From e8bfcf3115506b2c4109d912dd1bffc7e8c44d9b Mon Sep 17 00:00:00 2001
From: zhangmeng <775834166@qq.com>
Date: 星期三, 27 十一月 2019 16:40:50 +0800
Subject: [PATCH] update
---
sdk/sdk.go | 88 +++++++++++++++++++++++++++++++++----------
1 files changed, 67 insertions(+), 21 deletions(-)
diff --git a/sdk/sdk.go b/sdk/sdk.go
index 8fdbc9c..8e5c723 100644
--- a/sdk/sdk.go
+++ b/sdk/sdk.go
@@ -1,16 +1,14 @@
package sdk
import (
- "fmt"
- // "golang.org/x/sys/unix"
-
- "github.com/gogo/protobuf/proto"
"taskpubsub/tasktag"
"taskpubsub/util"
+ "github.com/gogo/protobuf/proto"
+
"basic.com/pubsub/protomsg.git"
"basic.com/valib/deliver.git"
- "taskpubsub/logger"
+ "basic.com/valib/logger.git"
)
const (
@@ -33,13 +31,13 @@
}
func initInnerTopic() {
- for _, sendTopic := range innerSendTopic {
- createSdkSendServerAndListen(sendTopic)
- }
+ // for _, sendTopic := range innerSendTopic {
+ // createSdkSendServerAndListen(sendTopic)
+ // }
- for _, recvTopic := range innerRecvTopic {
- createSdkRecvServerAndListen(recvTopic)
- }
+ // for _, recvTopic := range innerRecvTopic {
+ // createSdkRecvServerAndListen(recvTopic)
+ // }
}
func Init(useShm bool) {
@@ -88,13 +86,15 @@
logger.Info("create", id)
}
- url := fmt.Sprintf("ipc:///tmp/%s%s.ipc", id, postPush)
+ url := "ipc:///tmp/" + id + postPush + ".ipc"
m := deliver.PushPull
- if shm{
+ if shm {
m = deliver.Shm
url = id + postPush
- // unix.Unlink("/dev/shm/" + url)
}
+
+ logger.Info("SDK URL: ", url)
+
socket, err := util.NewSocketListen(int(m), url, shm)
if err != nil {
delete(SdkMap, id)
@@ -112,13 +112,12 @@
logger.Info("create", id)
}
- url := fmt.Sprintf("ipc:///tmp/%s%s.ipc", id, postPull)
+ url := "ipc:///tmp/" + id + postPull + ".ipc"
m := deliver.PushPull
- if shm{
+ if shm {
m = deliver.Shm
url = id + postPull
- // unix.Unlink("/dev/shm/" + url)
}
socket, err := util.NewSocketListen(int(m), url, shm)
@@ -195,14 +194,24 @@
func GetSdkSendTopic(sdkmsg protomsg.SdkMessage) (sendTopic string) {
if int(sdkmsg.Tasklab.Index) < len(sdkmsg.Tasklab.Sdkinfos) {
sendTopic = sdkmsg.Tasklab.Sdkinfos[sdkmsg.Tasklab.Index].Ipcid
+ if "Yolo" != sdkmsg.Tasklab.Sdkinfos[sdkmsg.Tasklab.Index].Sdktype &&
+ "FaceDetect" != sdkmsg.Tasklab.Sdkinfos[sdkmsg.Tasklab.Index].Sdktype &&
+ "FaceCompare" != sdkmsg.Tasklab.Sdkinfos[sdkmsg.Tasklab.Index].Sdktype {
+ //if sdkmsg.Tasklab.Sdkinfos[sdkmsg.Tasklab.Index].Sdktype == "" {
+ logger.Info("----------Sdktype:yitusoutu")
+ logger.Info("鍒嗗彂鐨勪富棰�:", sendTopic, "!Sdktype锛�", sdkmsg.Tasklab.Sdkinfos[sdkmsg.Tasklab.Index].Sdktype)
+ }
} else {
sendTopic = "es"
}
- logger.Debug("鍒嗗彂鐨勪富棰�: ", sendTopic, "浣嶇疆锛�", int(sdkmsg.Tasklab.Index)+1, "/", len(sdkmsg.Tasklab.Sdkinfos))
- return
+
+ return sendTopic
}
func Recv(socket util.SocketContext) {
+
+ tryCount := 0
+
var repsdkmsg = protomsg.SdkMessage{}
for {
select {
@@ -211,6 +220,19 @@
return
default:
if msg, err := socket.Sock.Recv(); err != nil {
+ tryCount++
+ socket = util.MaybeRestartSocket(socket, &tryCount)
+
+ // if socket.UseSHM {
+ // if tryCount > util.SHMMaxTryCount {
+ // socket.Sock.Close()
+ // socket, err = util.NewSocketListen(socket.Mode, socket.URL, socket.UseSHM)
+ // logger.Info("SDK SEND SHM TRY :", util.SHMMaxTryCount, " RESTART IT")
+ // tryCount = 0
+ // continue
+ // }
+ // tryCount++
+ // }
continue
} else {
err = proto.Unmarshal(msg, &repsdkmsg)
@@ -221,6 +243,9 @@
repsdkmsg.Tasklab.Index++
//璋冪敤璁＄畻鍑芥暟锛� 鍒嗗彂缁欎笅涓�涓富棰�
nexttopic := GetSdkSendTopic(repsdkmsg)
+ if "facedetect-sdk-no-track" == nexttopic || "virtual-faceextract-sdk-pull" == nexttopic {
+ logger.Info("nexttopic:", nexttopic)
+ }
SdkMap[nexttopic] <- repsdkmsg
}
}
@@ -228,6 +253,8 @@
}
func Send(sdkid string, socket util.SocketContext, in chan protomsg.SdkMessage) {
+
+ tryCount := 0
for {
select {
@@ -243,10 +270,29 @@
}
if err := socket.Sock.Send(data); err != nil {
- logger.Error("failed send:sdkid=", sdkid)
+ if "facedetect-sdk-no-track" == sdkid || "virtual-faceextract-sdk-pull" == sdkid {
+ logger.Error("failed send:sdkid=", sdkid)
+ }
+
+ tryCount++
+ socket = util.MaybeRestartSocket(socket, &tryCount)
+
+ // if socket.UseSHM {
+ // if tryCount > util.SHMMaxTryCount {
+ // socket.Sock.Close()
+ // socket, err = util.NewSocketListen(socket.Mode, socket.URL, socket.UseSHM)
+ // logger.Info("SDK SEND SHM TRY :", util.SHMMaxTryCount, " RESTART IT")
+ // tryCount = 0
+ // continue
+ // }
+ // tryCount++
+ // }
+
continue
}
- logger.Debug(sdkid, " send success: ", len(data))
+ if "facedetect-sdk-no-track" == sdkid || "virtual-faceextract-sdk-pull" == sdkid {
+ logger.Info(sdkid, " send success: ", len(data))
+ }
} else {
logger.Debug(sdkid, " 涓婚鍏抽棴, 鍏抽棴send()")
return
--
Gitblit v1.8.0