update

zhangmeng

2020-01-20 4324306f529b9bc62d7e818c0b12ff822687bb47

update

 common/helper.go

New file
@@ -0,0 +1,96 @@
package common

import (
    "context"
    "encoding/json"
    "fmt"
    "io/ioutil"
    "strconv"
    "time"

    "basic.com/pubsub/protomsg.git"
    "github.com/gogo/protobuf/proto"
)

// GetIpcAddress get ipc
func GetIpcAddress(shm bool, id string) string {
    if shm {
        return id
    }
    return `ipc:///tmp/` + id + `.ipc`
}

// SubConfig sub
type SubConfig struct {
    SoFile string            `json:"so_file_path"`
    Env    string            `json:"runtime"`
    Param  map[string]string `json:"param"`
}

// SdkConfig sdk
type SdkConfig struct {
    SoFile string            `json:"so_file_path"`
    Env    string            `json:"runtime"`
    Param  map[string]string `json:"param"`
    Sub    *SubConfig        `json:"sub"`
}

// ReadConfig conf
func ReadConfig(file string) (SdkConfig, error) {
    data, err := ioutil.ReadFile(file)
    if err != nil {
        return SdkConfig{}, fmt.Errorf("READ SDK CONFIG FILE %s ERROR", file)
    }

    //读取的数据为json格式，需要进行解码
    var v SdkConfig
    err = json.Unmarshal(data, &v)

    return v, err
}

// Atoi atoi
func Atoi(s string) int {
    i, _ := strconv.Atoi(s)
    return i
}

// UnserilizeProto un
func UnserilizeProto(ctx context.Context, data <-chan []byte, out chan<- protomsg.SdkMessage, fn func(...interface{})) {
    for {
        select {
        case <-ctx.Done():
            return
        case d := <-data:
            if len(d) < 100 {
                continue
            }
            msg := protomsg.SdkMessage{}
            if err := proto.Unmarshal(d, &msg); err != nil {
                fn(err, " msg 处理异常")
                continue
            }

            out <- msg

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

// UnpackImage unpack
func UnpackImage(msg protomsg.SdkMessage, fnName string, fn func(...interface{})) *protomsg.Image {
    // 反序列化数据得到sdk入参
    i := &protomsg.Image{}
    err := proto.Unmarshal(msg.Data, i)
    if err != nil {
        fn(fnName, " protobuf decode CameraImage error: ", err.Error())
        return nil
    }
    if i.Data == nil {
        fn(fnName, " protomsg.Image data null")
        return nil
    }
    return i
}

 common/reboot.go

New file
@@ -0,0 +1,64 @@
package common

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

// Disturber stop
type Disturber struct {
    mtx    sync.Mutex
    live   bool
    until  int
    maxTry int
}

// NewDisturber new
func NewDisturber(maxTry int) *Disturber {
    return &Disturber{
        live:   true,
        until:  0,
        maxTry: maxTry,
    }
}

// Prevent prevent
func (d *Disturber) Prevent() {
    d.mtx.Lock()
    defer d.mtx.Unlock()
    d.live = true
}

// MaybeReboot reboot
func (d *Disturber) MaybeReboot(ctx context.Context, fn func(...interface{})) {
    d.live = true
    for {
        select {
        case <-ctx.Done():
            return
        default:
            d.mtx.Lock()
            running := d.live
            d.mtx.Unlock()

            if running {
                d.until = 0

                d.mtx.Lock()
                d.live = false
                d.mtx.Unlock()

            } else {
                d.until++
                fn("!!!!!!No Running: ", d.until)
                if d.until > d.maxTry {
                    fn("!!!!!!Too Long Running, Reboot: ", d.maxTry)
                    os.Exit(0)
                }
            }
            time.Sleep(time.Second)
        }
    }
}

 run.go

@@ -48,7 +48,7 @@

    "reid/rpc"

    "basic.com/libgowrapper/sdkhelper.git"
    "reid/common"

    "basic.com/valib/gogpu.git"

@@ -68,7 +68,7 @@
// Create Reid
func Create(config string, typ, id string, gpu int, shm bool, ipc2Rule string, ruleMaxSize int, fn func(...interface{}), reserved map[string]interface{}) interface{} {

    cfg, err := sdkhelper.ReadConfig(config)
    cfg, err := common.ReadConfig(config)
    if err != nil {
        fn("Reid SDK Create Error When Read Config: ", err)
        return nil
@@ -88,7 +88,7 @@
        }
    }

    gpuM := sdkhelper.Atoi(cfg.Param[sGPU])
    gpuM := common.Atoi(cfg.Param[sGPU])

    rGPU := gpu

@@ -123,8 +123,8 @@
    ipcSnd := s.ipc + postPush
    ipcRcv := s.ipc + postPull

    sndURL := sdkhelper.GetIpcAddress(true, ipcSnd)
    rcvURL := sdkhelper.GetIpcAddress(true, ipcRcv)
    sndURL := common.GetIpcAddress(true, ipcSnd)
    rcvURL := common.GetIpcAddress(true, ipcRcv)

    chSnd := make(chan []byte, 3)
    chRcv := make(chan []byte, 3)
@@ -133,7 +133,7 @@
    go recv.Run(ctx)

    chMsg := make(chan protomsg.SdkMessage, 3)
    go sdkhelper.UnserilizeProto(ctx, chRcv, chMsg, s.fnLogger)
    go common.UnserilizeProto(ctx, chRcv, chMsg, s.fnLogger)

    send := rpc.NewSender(sndURL, chSnd, true, s.fnLogger)
    go send.Run(ctx)
@@ -147,7 +147,7 @@
                s.fnLogger("reid !!!!!! Recv Msg From Humantrack Error")
                continue
            }
            i := sdkhelper.UnpackImage(msg, "reid", s.fnLogger)
            i := common.UnpackImage(msg, "reid", s.fnLogger)
            if i == nil || i.Data == nil || i.Width <= 0 || i.Height <= 0 {
                s.fnLogger("reid !!!!!! Unpack Image From Humantrack Msg Failed")
                continue