From 467d125ef6ac819cc1e8bd1a8b5dcd00d491f624 Mon Sep 17 00:00:00 2001
From: zhangmeng <775834166@qq.com>
Date: 星期三, 30 十月 2019 09:13:39 +0800
Subject: [PATCH] bug fixed memvirt leak
---
sdk/sdk.go | 88 ++++++++++++++++++++++++++++++++++----------
1 files changed, 68 insertions(+), 20 deletions(-)
diff --git a/sdk/sdk.go b/sdk/sdk.go
index 6353513..ac4788e 100644
--- a/sdk/sdk.go
+++ b/sdk/sdk.go
@@ -1,11 +1,10 @@
package sdk
import (
- "fmt"
-
- "github.com/gogo/protobuf/proto"
"taskpubsub/tasktag"
"taskpubsub/util"
+
+ "github.com/gogo/protobuf/proto"
"basic.com/pubsub/protomsg.git"
"basic.com/valib/deliver.git"
@@ -13,12 +12,14 @@
)
const (
- postPush = "_1.ipc"
- postPull = "_2.ipc"
+ postPush = "_1"
+ postPull = "_2"
)
var SocketManage = make(map[string]util.SocketContext)
var SdkMap = make(map[string]chan protomsg.SdkMessage)
+
+var shm bool = false
var innerRecvTopic = []string{
"facedetect-sdk-no-track", //to sdk-no-track 浠ュ浘鎼滃浘
@@ -30,16 +31,18 @@
}
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() {
+func Init(useShm bool) {
+
+ shm = useShm
logger.Info("============= init sdk info =====================")
for _, sdkid := range util.Sdklist { // 鍒涘缓sdk server
@@ -53,8 +56,6 @@
// es
SdkMap["es"] = make(chan protomsg.SdkMessage)
logger.Info("create es channel: ")
- logger.Info("SdkMap:", SdkMap)
-
go DealEsTopic()
go autoUpdateSdk(util.Sdkflag)
@@ -85,8 +86,16 @@
logger.Info("create", id)
}
- url := fmt.Sprintf("ipc:///tmp/%s%s", id, postPush)
- socket, err := util.NewSocketListen(int(deliver.PushPull), url)
+ url := "ipc:///tmp/" + id + postPush + ".ipc"
+ m := deliver.PushPull
+ if shm {
+ m = deliver.Shm
+ url = id + postPush
+ }
+
+ logger.Info("SDK URL: ", url)
+
+ socket, err := util.NewSocketListen(int(m), url, shm)
if err != nil {
delete(SdkMap, id)
logger.Error(id, "create socket error!")
@@ -103,8 +112,16 @@
logger.Info("create", id)
}
- url := fmt.Sprintf("ipc:///tmp/%s%s", id, postPull)
- socket, err := util.NewSocketListen(int(deliver.PushPull), url)
+ url := "ipc:///tmp/" + id + postPull + ".ipc"
+ m := deliver.PushPull
+ if shm {
+ m = deliver.Shm
+ url = id + postPull
+
+ }
+
+ socket, err := util.NewSocketListen(int(m), url, shm)
+
if err != nil {
delete(SdkMap, id)
logger.Error(id, "create socket error!")
@@ -179,8 +196,8 @@
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 == "" {
+ "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)
}
@@ -192,6 +209,9 @@
}
func Recv(socket util.SocketContext) {
+
+ tryCount := 0
+
var repsdkmsg = protomsg.SdkMessage{}
for {
select {
@@ -200,6 +220,18 @@
return
default:
if msg, err := socket.Sock.Recv(); err != nil {
+ if socket.UseSHM {
+ if tryCount > util.SHMMaxTryCount {
+ socket.Sock.Close()
+ socket, err = util.NewSocketListen(socket.Mode, socket.URL, socket.UseSHM)
+ logger.Info("SDK RECV SHM TRY :", tryCount, " RESTART IT")
+
+ tryCount = 0
+
+ continue
+ }
+ tryCount++
+ }
continue
} else {
err = proto.Unmarshal(msg, &repsdkmsg)
@@ -221,6 +253,8 @@
func Send(sdkid string, socket util.SocketContext, in chan protomsg.SdkMessage) {
+ tryCount := 0
+
for {
select {
case <-socket.Context.Done():
@@ -238,6 +272,20 @@
if "facedetect-sdk-no-track" == sdkid || "virtual-faceextract-sdk-pull" == sdkid {
logger.Error("failed send:sdkid=", sdkid)
}
+
+ 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
}
if "facedetect-sdk-no-track" == sdkid || "virtual-faceextract-sdk-pull" == sdkid {
--
Gitblit v1.8.0