debug

zhangmeng

2020-01-13 5aefcc3832d9a1660ec9b04aa4cf2a69ced1a84d

debug

 run.go

@@ -12,92 +12,14 @@
    "basic.com/valib/gogpu.git"
)

type channel struct {
    index int
    life  int
}

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) 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,
        life:  f.maxLife,
    }
    f.mapChan[id] = c
    f.mapChanLock.Unlock()

    return c
}

func (f *face) maintainChannel(ctx context.Context, ch <-chan string) {
    for {
        select {
        case <-ctx.Done():
            return
        case id := <-ch:
            f.mapChanLock.Lock()
            for k, v := range f.mapChan {
                v.life--
                if k == id {
                    v.life = f.maxLife
                }
            }
            for k, v := range f.mapChan {
                if v.life < 0 {
                    delete(f.mapChan, k)
                }
            }
            f.mapChanLock.Unlock()

        default:
            time.Sleep(10 * time.Millisecond)
        }
    }
}

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

    maxChannel int

    maxLife     int
    chLife      chan string
    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{})

@@ -178,12 +100,11 @@

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

        maxLife:    trckInterval * 10,
        chLife:     make(chan string, maxChan),
        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,

@@ -196,84 +117,171 @@
    }
}

func (f *face) release() {
    if f.handle != nil {
        f.handle.Free()
    }
}

// Run run
func Run(ctx context.Context, i interface{}) {
    s := i.(*face)

    go s.maintainChannel(ctx, s.chLife)
    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)
    chSnd := make(chan sdkstruct.MsgSDK)

    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)
    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)
}

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

    wg := &sync.WaitGroup{}
func (f *face) cleanChnStat() {
    f.chnLock.Lock()
    for i := 0; i < f.maxChannel; i++ {
        f.ftrackChanStats[i] = false
    }
    f.chnLock.Unlock()
}

    for _, msg := range msgs {
        f.chLife <- msg.Cid
func (f *face) getAvailableChn() int {
    f.chnLock.Lock()
    defer f.chnLock.Unlock()

        if c, ok := f.channelExist(msg.Cid); ok {
            wg.Add(1)
            go f.track(wg, &msg, c.index, out)
        } else {
    for i := 0; i < f.maxChannel; i++ {
        if f.ftrackChanStats[i] == false {
            f.ftrackChanStats[i] = true
            return i
        }
    }
    return -1
}

            nc := f.channelNew(msg.Cid)
            if nc == nil {
                f.fnLogger("TOO MUCH CHANNEL")
                sdkhelper.EjectResult(nil, msg, out)
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
            }

            wg.Add(1)
            go f.track(wg, &msg, nc.index, out)
            if _, ok := f.ftrackChans[rMsg.Cid]; ok {
                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
            }
        default:
            time.Sleep(time.Millisecond * 100)
        }
    }
    wg.Wait()
    f.fnLogger("######Wait for All track")
}

func (f *face) track(wg *sync.WaitGroup, msg *protomsg.SdkMessage, tchan int, out chan<- sdkstruct.MsgSDK) {
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

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

    rMsg := *msg
        case rMsg := <-in:

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

        return
            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)

            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("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)
            if curCid != "" {
                delete(f.ftrackChans, curCid)
                f.releaseChn(dtchn)
            }
            return

        }
    }

    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: ", tchan)

        return
    }

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

    ret := f.handle.TrackerResize(imgW, imgH, tchan)
    f.fnLogger("ResizeFaceTracker: cid: ", msg.Cid, " chan: ", tchan, " wXh: ", imgW, "x", imgH, " result:", ret)

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

    sdkhelper.EjectResult(data, rMsg, out)

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

}