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