debug send/recv and add logs

zhangmeng

2020-02-24 d9b98e947f7b5aae886ff02ec27bd92aa34feaaa

debug send/recv and add logs

 rpc/recv.go

@@ -8,8 +8,6 @@
    "basic.com/valib/deliver.git"
)

const mode = deliver.PushPull

// Reciever recv from ipc
type Reciever struct {
    ctx    context.Context
@@ -36,7 +34,7 @@
    if r.shm {
        r.runShm(ctx)
    } else {
        r.run(ctx, deliver.NewClient(mode, r.ipcURL))
        r.run(ctx, deliver.NewServer(deliver.PushPull, r.ipcURL))
    }
}

@@ -54,19 +52,25 @@
            if r.shm {
                if d, err := i.Recv(); err != nil {
                    i.Close()
                    r.fnLogger("Reciever RECV ERROR: ", err)
                    r.fnLogger("Reciever RECV From:", r.ipcURL, " ERROR: ", err)

                    c, err := deliver.NewServerWithError(deliver.Shm, r.ipcURL)
                loopR:
                    for {
                        select {
                        case <-ctx.Done():
                            return
                        default:
                        if err == nil {
                            break
                                break loopR
                        }
                        r.fnLogger("Reciever CREATE FAILED : ", err)
                        time.Sleep(time.Second)
                        c, err = deliver.NewServerWithError(deliver.Shm, r.ipcURL)
                            r.fnLogger("Recver ANALYSIS CREATE:", r.ipcURL, "  FAILED : ", err)
                        }
                    }
                    i = c
                    r.fnLogger("Reciever CREATE SHM")
                    r.fnLogger("Reciever CREATE SHM:", r.ipcURL)
                } else {
                    if d != nil {
                        count++
@@ -100,13 +104,20 @@

func (r *Reciever) runShm(ctx context.Context) {
    c, err := deliver.NewServerWithError(deliver.Shm, r.ipcURL)
loopRBegin:
    for {
        select {
        case <-ctx.Done():
            return
        default:
        if err == nil {
            break
                break loopRBegin
        }
        r.fnLogger("Reciever CLIENT CREATE FAILED : ", err)
        time.Sleep(1 * time.Second)
        c, err = deliver.NewServerWithError(deliver.Shm, r.ipcURL)
            r.fnLogger("Recver CLIENT CREATE", r.ipcURL, "FAILED : ", err)
    }
    }

    r.run(ctx, c)
}

 rpc/send.go

@@ -12,7 +12,6 @@
    ipcURL string
    in     <-chan []byte
    shm    bool
    fn     func([]byte, bool)

    fnLogger func(...interface{})
}
@@ -23,7 +22,6 @@
        ipcURL:   ipcURL,
        in:       in,
        shm:      shm,
        fn:       nil,
        fnLogger: fn,
    }
}
@@ -34,7 +32,7 @@
    if s.shm {
        s.runShm(ctx)
    } else {
        i := deliver.NewClient(mode, s.ipcURL)
        i := deliver.NewServer(deliver.PushPull, s.ipcURL)
        if i == nil {
            s.fnLogger("sender 2 pubsub nng create error")
            return
@@ -55,18 +53,27 @@
            if s.shm {
                if err := i.Send(d); err != nil {
                    i.Close()
                    s.fnLogger("ANALYSIS SENDER ERROR: ", err)
                    s.fnLogger("SENDER To:", s.ipcURL, " ERROR: ", err)

                    c, err := deliver.NewServerWithError(deliver.Shm, s.ipcURL)
                loopS:
                    for {
                        select {
                        case <-ctx.Done():
                            return
                        default:
                        if err == nil {
                            break
                                break loopS
                        }
                        time.Sleep(time.Second)
                        c, err = deliver.NewServerWithError(deliver.Shm, s.ipcURL)
                        s.fnLogger("CLIENT CREATE FAILED : ", err)
                            s.fnLogger("Sender CLIENT CREATE:", s.ipcURL, " FAILED : ", err)
                    }

                    }

                    i = c
                    s.fnLogger("Sender Create Shm:", s.ipcURL)
                } else {

                }
@@ -86,13 +93,20 @@

func (s *Sender) runShm(ctx context.Context) {
    c, err := deliver.NewServerWithError(deliver.Shm, s.ipcURL)
loopSBegin:
    for {
        select {
        case <-ctx.Done():
            return
        default:
        if err == nil {
            break
                break loopSBegin
        }
        time.Sleep(1 * time.Second)
        c, err = deliver.NewServerWithError(deliver.Shm, s.ipcURL)
        s.fnLogger("CLIENT CREATE FAILED : ", err)
            s.fnLogger("Sender CLIENT CREATE:", s.ipcURL, "FAILED : ", err)
    }
    }

    s.run(ctx, c)
}

 run.go

@@ -154,7 +154,7 @@
            }

            sdkInfo := msg.Tasklab.Sdkinfos[int(msg.Tasklab.Index)]
            s.fnLogger("reid !!!!!! Recv From Humantrack SDK Result Length: ", len(sdkInfo.Sdkdata))
            s.fnLogger("reid~~~~~~Recv From Humantrack SDK Result Length: ", len(sdkInfo.Sdkdata))

            res := &protomsg.HumanTrackResult{}
            if err := proto.Unmarshal(sdkInfo.Sdkdata, res); err != nil {
@@ -181,7 +181,7 @@
            if len(res.Result) > 0 {
                if out, err := proto.Marshal(res); err == nil {
                    msg.Tasklab.Sdkinfos[int(msg.Tasklab.Index)].Sdkdata = out
                    s.fnLogger("reid !!!!!! Send To Humantrack Result Length:", len(out))
                    s.fnLogger("reid~~~~~~Send To Humantrack Result Length:", len(out))
                }
            }

@@ -190,7 +190,7 @@
                    s.fnLogger("reid !!!!!! proto.Marshal Failed To Marshal proto.SdkMessage")
                    continue
                }
                // s.fnLogger("reid !!!!!! MSG Send Back To Humantrack Length:", len(data))
                s.fnLogger("reid~~~~~~MSG Send Back To Humantrack Length:", len(data))

                chSnd <- data
            } else {