From 1df33ffacf98f09a33969a8c8ed4078247524d29 Mon Sep 17 00:00:00 2001
From: chenshijun <chenshijun@aiotlink.com>
Date: 星期五, 09 十月 2020 14:27:45 +0800
Subject: [PATCH] attach增加超时

---
 shmqueue.go |  108 +++++++++++++++++++++++++++++++++++------------------
 1 files changed, 71 insertions(+), 37 deletions(-)

diff --git a/shmqueue.go b/shmqueue.go
index 3495406..9c4f16c 100644
--- a/shmqueue.go
+++ b/shmqueue.go
@@ -3,7 +3,7 @@
 import (
 	"context"
 	"fmt"
-	"github.com/gen2brain/shm"
+	shm "basic.com/valib/goshm.git"
 	"reflect"
 	"runtime"
 	"sync/atomic"
@@ -12,7 +12,7 @@
 )
 
 const (
-	TimePeriodPutOrGet = time.Duration(10) * time.Microsecond
+	TimePeriodPutOrGet = time.Duration(5)*time.Millisecond //ms
 )
 
 //Element info
@@ -72,6 +72,9 @@
 	shmLen = shmblocks*uint32(unsafe.Sizeof(datainfo)) + uint32(unsafe.Sizeof(shmstruct))
 
 	data, shmid := CreateRawShm(ctx, int(shmLen), key)
+	if shmid == -1 {
+		return nil
+	}
 	q := bytes2shmEsQueue(data)
 	//init parameters
 	q.capacity = shmblocks
@@ -103,7 +106,10 @@
 func AttachQueue(ctx context.Context, key int) *EsQueueInfo {
 
 	var eqi EsQueueInfo
-	data, shmid := AttachRawShm(ctx, key)
+	data, shmid := AttachRawShmTimeout(ctx, key)
+	if shmid == -1 {
+		return nil
+	}
 	shmdata := bytes2shmEsQueue(data)
 	eqi.EsCaches = ptr2esCache(unsafe.Pointer(&shmdata.cache), int(shmdata.capacity))
 	//fmt.Println("AttachQueue EsCaches:", eqi.EsCaches)
@@ -146,14 +152,14 @@
 func (eqi *EsQueueInfo) QueueSize() int {
 	var putPos, getPos uint32
 	var quantity uint32
+
 	getPos = atomic.LoadUint32(&eqi.Queue.getPos)
 	putPos = atomic.LoadUint32(&eqi.Queue.putPos)
 
 	if putPos >= getPos {
 		quantity = putPos - getPos
 	} else {
-		//quantity = q.capMod + (putPos - getPos)
-		quantity = (eqi.Queue.capMod + (putPos - getPos)) % eqi.Queue.capMod
+		quantity = eqi.Queue.capMod + (putPos - getPos)
 	}
 
 	return int(quantity)
@@ -164,7 +170,8 @@
 	var elems []ElemInfo
 
 	for i := 0; i < len(eqi.EsCaches); i++ {
-		if eqi.EsCaches[i].value.PicId != 0 && eqi.EsCaches[i].value.InfoId != 0 {
+		if eqi.EsCaches[i].value.PicId != -1 && eqi.EsCaches[i].value.InfoId != -1 &&
+			eqi.EsCaches[i].value.PicId != 0 && eqi.EsCaches[i].value.InfoId != 0 {
 			elems = append(elems, eqi.EsCaches[i].value)
 		}
 	}
@@ -184,8 +191,7 @@
 	if putPos >= getPos {
 		posCnt = putPos - getPos
 	} else {
-		//posCnt = capMod + (putPos - getPos)
-		posCnt = (capMod + (putPos - getPos)) % capMod
+		posCnt = capMod + (putPos - getPos)
 	}
 
 	//todo
@@ -218,18 +224,29 @@
 }
 
 //PutForce 寮哄埗鍐欏叆锛屽け璐ュ垯缁х画閲嶅啓锛岀洿鍒版垚鍔熶负姝�
-func (eqi *EsQueueInfo) PutForce(val ElemInfo) int {
+func (eqi *EsQueueInfo) PutForce(ctx context.Context, val ElemInfo) int {
 	ok, qua := eqi.Put(val)
 	for !ok {
-		time.Sleep(TimePeriodPutOrGet)
-		ok, qua = eqi.Put(val)
+	loopB:
+		for {
+			select {
+			case <-ctx.Done():
+				return 0
+			default:
+				ok, qua = eqi.Put(val)
+				if ok {
+					break loopB
+				}
+				time.Sleep(TimePeriodPutOrGet)
+			}
+		}
 	}
 
 	return qua
 }
 
 //PutForceWithinTime 鍦╰imeout ms鐨勬椂闂村唴锛屽己鍒跺啓鍏ワ紝澶辫触鍒欑户缁噸鍐欙紝鐩村埌鎴愬姛涓烘鎴栬�呰秴鏃堕��鍑�
-func (eqi *EsQueueInfo) PutForceWithinTime(val ElemInfo, timeout int) (bool, int) {
+func (eqi *EsQueueInfo) PutForceWithinTime(ctx context.Context, val ElemInfo, timeout int) (bool, int) {
 	ok, qua := eqi.Put(val)
 	if ok {
 		return ok, qua
@@ -238,18 +255,21 @@
 	to := time.NewTimer(time.Duration(timeout) * time.Millisecond)
 	defer to.Stop()
 
-	for {
-		select {
-		case <-to.C:
-			return false, 0
-		default:
-			ok, qua = eqi.Put(val)
-			if ok {
-				return ok, qua
+	loopB:
+		for {
+			select {
+			case <-ctx.Done():
+				return false, 0
+			case <-to.C:
+				return false, 0
+			default:
+				ok, qua = eqi.Put(val)
+				if ok {
+					break loopB
+				}
+				time.Sleep(TimePeriodPutOrGet)
 			}
-			time.Sleep(TimePeriodPutOrGet)
 		}
-	}
 	return ok, qua
 }
 
@@ -286,7 +306,7 @@
 		putNo := atomic.LoadUint32(&cache.putNo)
 		if getPosNew == getNo && getNo == putNo-eqi.Queue.capacity {
 			val := cache.value
-			cache.value = ElemInfo{PicId: 0}
+			cache.value = ElemInfo{PicId: 0, InfoId:0}
 			atomic.AddUint32(&cache.getNo, eqi.Queue.capacity)
 			return val, true, int(posCnt - 1)
 		} else {
@@ -296,18 +316,29 @@
 }
 
 //GetForce 寮哄埗鑾峰彇鏁版嵁锛屽け璐ュ垯缁х画鑾峰彇锛岀洿鍒版垚鍔熶负姝�
-func (eqi *EsQueueInfo) GetForce() (ElemInfo, int) {
+func (eqi *EsQueueInfo) GetForce(ctx context.Context) (ElemInfo, int) {
 	val, ok, qua := eqi.Get()
 	for !ok {
-		time.Sleep(TimePeriodPutOrGet)
-		val, ok, qua = eqi.Get()
+	loopB:
+		for {
+			select {
+			case <-ctx.Done():
+				return ElemInfo{}, 0
+			default:
+				val, ok, qua = eqi.Get()
+				if ok {
+					break loopB
+				}
+				time.Sleep(TimePeriodPutOrGet)
+			}
+		}
 	}
 
 	return val, qua
 }
 
 //GetForceWithinTime 鍦╰imeout ms鐨勬椂闂村唴锛屽己鍒惰幏鍙栨暟鎹紝澶辫触鍒欑户缁幏鍙栵紝鐩村埌鎴愬姛涓烘
-func (eqi *EsQueueInfo) GetForceWithinTime(timeout int) (ElemInfo, bool, int) {
+func (eqi *EsQueueInfo) GetForceWithinTime(ctx context.Context, timeout int) (ElemInfo, bool, int) {
 	val, ok, qua := eqi.Get()
 	if ok {
 		return val, ok, qua
@@ -316,18 +347,21 @@
 	to := time.NewTimer(time.Duration(timeout) * time.Millisecond)
 	defer to.Stop()
 
-	for {
-		select {
-		case <-to.C:
-			return ElemInfo{}, false, 0
-		default:
-			val, ok, qua = eqi.Get()
-			if ok {
-				return val, ok, qua
+	loopB:
+		for {
+			select {
+			case <-ctx.Done():
+				return ElemInfo{}, false, 0
+			case <-to.C:
+				return ElemInfo{}, false, 0
+			default:
+				val, ok, qua = eqi.Get()
+				if ok {
+					break loopB
+				}
+				time.Sleep(TimePeriodPutOrGet)
 			}
-			time.Sleep(TimePeriodPutOrGet)
 		}
-	}
 	return val, ok, qua
 }
 

--
Gitblit v1.8.0