From 82a262ef5cf721e5a236c8b1d2ab3ff92ca33122 Mon Sep 17 00:00:00 2001
From: zhangmeng <775834166@qq.com>
Date: 星期五, 24 九月 2021 11:40:20 +0800
Subject: [PATCH] bug fixed withintime

---
 shmqueue.go |  134 +++++++++++++++++++++++++-------------------
 1 files changed, 75 insertions(+), 59 deletions(-)

diff --git a/shmqueue.go b/shmqueue.go
index 2cc6908..cb2c12a 100644
--- a/shmqueue.go
+++ b/shmqueue.go
@@ -1,9 +1,9 @@
 package shmqueue
 
 import (
+	shm "basic.com/valib/goshm.git"
 	"context"
 	"fmt"
-	"github.com/gen2brain/shm"
 	"reflect"
 	"runtime"
 	"sync/atomic"
@@ -12,7 +12,7 @@
 )
 
 const (
-	TimePeriodPutOrGet = time.Duration(1) * 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)
@@ -139,25 +145,21 @@
 
 // Capacity queue 瀹归噺锛屾�诲ぇ灏�
 func (eqi *EsQueueInfo) QueueCapacity() int {
-	var capMod uint32
-	capMod = atomic.LoadUint32(&eqi.Queue.capMod)
-	return int(capMod)
+	return int(eqi.Queue.capMod)
 }
 
 // Size queue 瀹為檯澶у皬
 func (eqi *EsQueueInfo) QueueSize() int {
-	var putPos, getPos, capMod uint32
+	var putPos, getPos uint32
 	var quantity uint32
 
-	capMod = atomic.LoadUint32(&eqi.Queue.capMod)
 	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 = (capMod + (putPos - getPos)) % capMod
+		quantity = eqi.Queue.capMod + (putPos - getPos)
 	}
 
 	return int(quantity)
@@ -179,19 +181,17 @@
 
 // Put 鍗曟鍐欏叆锛屽彲鑳藉け璐�
 func (eqi *EsQueueInfo) Put(val ElemInfo) (bool, int) {
-	var putPos, putPosNew, getPos, posCnt, capMod ,capacity uint32
+	var putPos, putPosNew, getPos, posCnt uint32
 	var cache *esCache
+	capMod := eqi.Queue.capMod
 
-	capacity = atomic.LoadUint32(&eqi.Queue.capacity)
-	capMod = atomic.LoadUint32(&eqi.Queue.capMod)
 	getPos = atomic.LoadUint32(&eqi.Queue.getPos)
 	putPos = atomic.LoadUint32(&eqi.Queue.putPos)
 
 	if putPos >= getPos {
 		posCnt = putPos - getPos
 	} else {
-		//posCnt = capMod + (putPos - getPos)
-		posCnt = (capMod + (putPos - getPos)) % capMod
+		posCnt = capMod + (putPos - getPos)
 	}
 
 	//todo
@@ -215,7 +215,7 @@
 		putNo := atomic.LoadUint32(&cache.putNo)
 		if putPosNew == putNo && getNo == putNo {
 			cache.value = val
-			atomic.AddUint32(&cache.putNo, capacity)
+			atomic.AddUint32(&cache.putNo, eqi.Queue.capacity)
 			return true, int(posCnt + 1)
 		} else {
 			runtime.Gosched()
@@ -252,34 +252,42 @@
 		return ok, qua
 	}
 
-	to := time.NewTimer(time.Duration(timeout) * time.Millisecond)
-	defer to.Stop()
-
-	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)
-			}
+	count := timeout/int(TimePeriodPutOrGet) + 1
+	for i := 0; i < count; i++ {
+		if ok, qua := eqi.Put(val); ok {
+			return ok, qua
 		}
-	return ok, qua
+		time.Sleep(TimePeriodPutOrGet)
+	}
+	return false, 0
+
+	// 	to := time.NewTimer(time.Duration(timeout) * time.Millisecond)
+	// 	defer to.Stop()
+
+	// 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)
+	// 		}
+	// 	}
+	// 	return ok, qua
 }
 
 // Get 鍗曟鑾峰彇锛屽彲鑳藉け璐�
 func (eqi *EsQueueInfo) Get() (ElemInfo, bool, int) {
-	var putPos, getPos, getPosNew, posCnt, capMod, capacity uint32
+	var putPos, getPos, getPosNew, posCnt uint32
 	var cache *esCache
+	capMod := eqi.Queue.capMod
 
-	capacity = atomic.LoadUint32(&eqi.Queue.capacity)
-	capMod = atomic.LoadUint32(&eqi.Queue.capMod)
 	putPos = atomic.LoadUint32(&eqi.Queue.putPos)
 	getPos = atomic.LoadUint32(&eqi.Queue.getPos)
 
@@ -305,10 +313,10 @@
 	for {
 		getNo := atomic.LoadUint32(&cache.getNo)
 		putNo := atomic.LoadUint32(&cache.putNo)
-		if getPosNew == getNo && getNo == putNo-capacity {
+		if getPosNew == getNo && getNo == putNo-eqi.Queue.capacity {
 			val := cache.value
-			cache.value = ElemInfo{PicId: 0}
-			atomic.AddUint32(&cache.getNo, capacity)
+			cache.value = ElemInfo{PicId: 0, InfoId: 0}
+			atomic.AddUint32(&cache.getNo, eqi.Queue.capacity)
 			return val, true, int(posCnt - 1)
 		} else {
 			runtime.Gosched()
@@ -344,26 +352,34 @@
 	if ok {
 		return val, ok, qua
 	}
-
-	to := time.NewTimer(time.Duration(timeout) * time.Millisecond)
-	defer to.Stop()
-
-	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)
-			}
+	count := timeout/int(TimePeriodPutOrGet) + 1
+	for i := 0; i < count; i++ {
+		if val, ok, qua := eqi.Get(); ok {
+			return val, ok, qua
 		}
-	return val, ok, qua
+		time.Sleep(TimePeriodPutOrGet)
+	}
+	return ElemInfo{}, false, 0
+
+	// 	to := time.NewTimer(time.Duration(timeout) * time.Millisecond)
+	// 	defer to.Stop()
+
+	// 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)
+	// 		}
+	// 	}
+	// 	return val, ok, qua
 }
 
 // round 鍒版渶杩戠殑2鐨勫�嶆暟

--
Gitblit v1.8.0