~zhangmeng/face.git

			@@ -2,6 +2,7 @@

			import (
			"context"
			"os"
			"sync"
			"time"

			@@ -12,75 +13,16 @@
			"basic.com/valib/gogpu.git"
			)

			type channel struct {
			index int
			msgs *sdkhelper.LockList
			}

			//////////////////////////////////////////////////////////////////
			const (
			trackChnTimeout = time.Duration(10)
			)

			func (f *face) channelAvailable() int {
			f.mapChanLock.Lock()
			defer f.mapChanLock.Unlock()

			for i := 0; i < f.maxChannel; i++ {
			found := false
			for _, v := range f.mapChan {
			if v.index == i {
			found = true
			break
			}
			}
			if !found {
			return i
			}
			}
			return -1
			}

			func (f *face) channelRelease(id string) {
			f.mapChanLock.Lock()
			defer f.mapChanLock.Unlock()

			delete(f.mapChan, id)
			}

			func (f face) channelExist(id string) (channel, bool) {
			f.mapChanLock.Lock()
			defer f.mapChanLock.Unlock()

			c, ok := f.mapChan[id]
			return c, ok
			}

			func (f face) channelNew(id string) channel {

			index := f.channelAvailable()
			if index < 0 {
			return nil
			}

			f.mapChanLock.Lock()
			c := &channel{
			index: index,
			msgs: sdkhelper.NewLockList(3),
			}
			f.mapChan[id] = c
			f.mapChanLock.Unlock()

			return c
			}
			const maxTryBeforeReboot = 10

			type face struct {
			handle *SDKFace
			list *sdkhelper.LockList

			maxChannel int
			mapChan map[string]*channel
			mapChanLock sync.Mutex
			maxChannel int
			ftrackChans map[string]chan protomsg.SdkMessage
			ftrackChannels map[string]int
			ftrackChanStats []bool
			chnLock sync.Mutex

			fnLogger func(...interface{})

			@@ -90,6 +32,40 @@
			ipc2Rule string
			ruleMsgMaxCacheSize int
			reserved map[string]interface{}

			running bool
			rebootUntil int
			mtxRunning sync.Mutex
			}

			func (f *face) maybeReboot(ctx context.Context) {
			for {
			select {
			case <-ctx.Done():
			return
			default:
			f.mtxRunning.Lock()
			running := f.running
			f.mtxRunning.Unlock()

			if running {
			f.rebootUntil = 0

			f.mtxRunning.Lock()
			f.running = false
			f.mtxRunning.Unlock()

			} else {
			f.rebootUntil++
			f.fnLogger("Face No Running: ", f.rebootUntil)
			if f.rebootUntil > maxTryBeforeReboot {
			f.fnLogger("Face Too Long Running, Reboot")
			os.Exit(0)
			}
			}
			time.Sleep(time.Second)
			}
			}
			}

			// Create create sdk
			@@ -161,10 +137,11 @@

			return &face{
			handle: handle,
			list: sdkhelper.NewLockList(maxChan + maxChan/2),

			maxChannel: maxChan,
			mapChan: make(map[string]*channel, maxChan),
			maxChannel: maxChan,
			ftrackChans: make(map[string]chan protomsg.SdkMessage, maxChan),
			ftrackChannels: make(map[string]int, maxChan),
			ftrackChanStats: make([]bool, maxChan, maxChan),

			fnLogger: fn,

			@@ -174,6 +151,9 @@
			ipc2Rule: ipc2Rule,
			ruleMsgMaxCacheSize: ruleMaxSize,
			reserved: reserved,

			running: true,
			rebootUntil: maxTryBeforeReboot,
			}
			}

			@@ -187,93 +167,176 @@
			func Run(ctx context.Context, i interface{}) {
			s := i.(*face)

			chRcv, chSnd := sdkhelper.FlowCreate(ctx, s.id, s.shm, s.ipc2Rule, s.ruleMsgMaxCacheSize, s.fnLogger)
			go sdkhelper.FlowBatch(ctx, chRcv, chSnd, s.typ, s.list.Push, s.list.Drain, s.run, s.release, s.fnLogger)
			const (
			postPull = `_1`
			postPush = `_2`
			)
			ipcRcv := sdkhelper.GetIpcAddress(s.shm, s.id+postPull)
			ipcSnd := sdkhelper.GetIpcAddress(s.shm, s.id+postPush)
			chRcv := make(chan []byte, s.maxChannel)
			chSnd := make(chan sdkstruct.MsgSDK, s.maxChannel)

			rcver := sdkhelper.NewReciever(ipcRcv, chRcv, s.shm, s.fnLogger)
			snder := sdkhelper.NewSender(ipcSnd, chSnd, s.shm, s.fnLogger)
			torule := sdkhelper.NewToRule(s.ipc2Rule, s.ruleMsgMaxCacheSize, s.fnLogger)

			snder.ApplyCallbackFunc(torule.Push)

			go rcver.Run(ctx)
			go snder.Run(ctx)
			go torule.Run(ctx)

			go s.run(ctx, chRcv, chSnd)

			go s.maybeReboot(ctx)
			}

			func (f *face) run(msgs []protomsg.SdkMessage, out chan<- sdkstruct.MsgSDK, typ string) {
			//////////////////////////////////////////////////////////////////
			const (
			cacheFrameNum = 3
			trackChnTimeout = time.Duration(10)
			)

			for _, msg := range msgs {
			if c, ok := f.channelExist(msg.Cid); ok {
			c.msgs.Push(msg)
			} else {
			func (f *face) cleanChnStat() {
			f.chnLock.Lock()
			for i := 0; i < f.maxChannel; i++ {
			f.ftrackChanStats[i] = false
			}
			f.chnLock.Unlock()
			}

			nc := f.channelNew(msg.Cid)
			if nc == nil {
			f.fnLogger("TOO MUCH CHANNEL")
			sdkhelper.EjectResult(nil, msg, out)
			func (f *face) getAvailableChn() int {
			f.chnLock.Lock()
			defer f.chnLock.Unlock()

			for i := 0; i < f.maxChannel; i++ {
			if f.ftrackChanStats[i] == false {
			f.ftrackChanStats[i] = true
			return i
			}
			}
			return -1
			}

			func (f *face) releaseChn(chn int) {
			f.chnLock.Lock()
			f.ftrackChanStats[chn] = false
			f.chnLock.Unlock()
			}

			func (f *face) run(ctx context.Context, in <-chan []byte, out chan<- sdkstruct.MsgSDK) {

			chMsg := make(chan protomsg.SdkMessage)
			go sdkhelper.UnserilizeProto(ctx, in, chMsg, f.fnLogger)

			for {
			select {
			case <-ctx.Done():
			f.handle.Free()
			return
			case rMsg := <-chMsg:
			if !sdkhelper.ValidRemoteMessage(rMsg, f.typ, f.fnLogger) {
			f.fnLogger("FACE TRACK VALIDREMOTEMESSAGE INVALID")
			sdkhelper.EjectResult(nil, rMsg, out)
			continue
			}

			i := sdkhelper.UnpackImage(msg, f.typ, f.fnLogger)
			if i == nil {
			sdkhelper.EjectResult(nil, msg, out)
			continue
			if _, ok := f.ftrackChans[rMsg.Cid]; ok {
			f.fnLogger("Face Cache Size: ", len(f.ftrackChans))
			f.ftrackChans[rMsg.Cid] <- rMsg
			} else {

			f.ftrackChans[rMsg.Cid] = make(chan protomsg.SdkMessage, cacheFrameNum)
			chn := f.getAvailableChn()
			if chn < 0 {
			f.fnLogger("TOO MUCH CHANNEL")
			sdkhelper.EjectResult(nil, rMsg, out)
			continue
			}
			f.ftrackChannels[rMsg.Cid] = chn

			i := sdkhelper.UnpackImage(rMsg, f.typ, f.fnLogger)
			if i == nil {
			sdkhelper.EjectResult(nil, rMsg, out)
			continue
			}
			// conv to bgr24 and resize
			imgW, imgH := int(i.Width), int(i.Height)
			ret := f.handle.TrackerResize(imgW, imgH, chn)
			f.fnLogger("ResizeFaceTracker: cid: ", rMsg.Cid, " chan: ", chn, " wXh: ", imgW, "x", imgH, " result:", ret)
			go f.detectTrackOneChn(ctx, f.ftrackChans[rMsg.Cid], out, chn)
			f.ftrackChans[rMsg.Cid] <- rMsg
			}
			// conv to bgr24 and resize
			imgW, imgH := int(i.Width), int(i.Height)
			ret := f.handle.TrackerResize(imgW, imgH, nc.index)
			f.fnLogger("ResizeFaceTracker: cid: ", msg.Cid, " chan: ", nc.index, " wXh: ", imgW, "x", imgH, " result:", ret)
			go f.detectTrackOneChn(nc, out)
			nc.msgs.Push(msg)
			default:
			time.Sleep(time.Millisecond * 100)
			}
			}
			}

			func (f face) detectTrackOneChn(c channel, out chan<- sdkstruct.MsgSDK) {

			dtchn := c.index
			func (f *face) detectTrackOneChn(ctx context.Context, in <-chan protomsg.SdkMessage, out chan<- sdkstruct.MsgSDK, dtchn int) {
			tm := time.Now()
			sc := 0
			f.fnLogger("DETECTTRACKONECHN DTCHN: ", dtchn)
			var curCid string

			for {
			select {
			case <-ctx.Done():
			return

			case rMsg := <-in:

			if !sdkhelper.ValidRemoteMessage(rMsg, f.typ, f.fnLogger) {
			sdkhelper.EjectResult(nil, rMsg, out)
			continue
			}

			i := sdkhelper.UnpackImage(rMsg, f.typ, f.fnLogger)
			if i == nil \|\| i.Data == nil \|\| i.Width <= 0 \|\| i.Height <= 0 {
			sdkhelper.EjectResult(nil, rMsg, out)
			continue
			}

			curCid = i.Cid

			// conv to bgr24 and resize
			imgW, imgH := int(i.Width), int(i.Height)

			f.fnLogger("Face Start Run:", dtchn, "CAMERAID: ", rMsg.Cid)

			count, data, _ := f.handle.Run(i.Data, imgW, imgH, 3, dtchn)

			sdkhelper.EjectResult(data, rMsg, out)
			f.mtxRunning.Lock()
			f.running = true
			f.mtxRunning.Unlock()

			var id, name string
			if rMsg.Tasklab != nil {
			id, name = rMsg.Tasklab.Taskid, rMsg.Tasklab.Taskname
			}
			f.fnLogger("Chan:", dtchn, "CAMERAID: ", rMsg.Cid, " TASKID: ", id, " TASKNAME: ", name, " DETECT ", f.typ, " COUNT: ", count)

			sc++
			if sc == 25 {
			f.fnLogger("CHAN:%d, FACE RUN 25 FRAME USE TIME: ", dtchn, time.Since(tm))
			sc = 0
			tm = time.Now()
			}

			if time.Since(tm) > time.Second {
			f.fnLogger("CHAN: ", dtchn, " FACE RUN ", sc, " FRAME USE TIME: ", time.Since(tm))
			sc = 0
			tm = time.Now()
			}
			case <-time.After(trackChnTimeout * time.Second):
			f.fnLogger("######Timeout to get image, curCid:", curCid, " Chan:", dtchn)
			f.fnLogger("Timeout to get image, curCid:", curCid)
			if curCid != "" {
			f.channelRelease(curCid)
			delete(f.ftrackChans, curCid)
			f.releaseChn(dtchn)
			}
			return
			default:
			// f.fnLogger("DETECTTRACKONECHN DTCHN: ", dtchn)

			elems := c.msgs.Drain()
			var msgs []protomsg.SdkMessage
			for _, v := range elems {
			msgs = append(msgs, v.(protomsg.SdkMessage))
			}
			for _, rMsg := range msgs {

			if !sdkhelper.ValidRemoteMessage(rMsg, f.typ, f.fnLogger) {
			sdkhelper.EjectResult(nil, rMsg, out)
			f.fnLogger("Face!!!!!!SkdMessage Invalid: ", dtchn)

			continue
			}

			i := sdkhelper.UnpackImage(rMsg, f.typ, f.fnLogger)
			if i == nil \|\| i.Data == nil \|\| i.Width <= 0 \|\| i.Height <= 0 {
			sdkhelper.EjectResult(nil, rMsg, out)
			f.fnLogger("Face!!!!!!Unpack Image From SkdMessage Failed: ", dtchn)

			continue
			}

			curCid = i.Cid

			// conv to bgr24 and resize
			imgW, imgH := int(i.Width), int(i.Height)

			count, data, _ := f.handle.Run(i.Data, imgW, imgH, 3, dtchn)

			sdkhelper.EjectResult(data, rMsg, out)

			var id, name string
			if rMsg.Tasklab != nil {
			id, name = rMsg.Tasklab.Taskid, rMsg.Tasklab.Taskname
			}
			// f.fnLogger("Chan: ", dtchn, "CAMERAID: ", rMsg.Cid, " TASKID: ", id, " TASKNAME: ", name, " DETECT ", f.typ, " COUNT: ", count)
			}
			}
			}
			}