20220719更新僵尸进程处理

Send_tranfer_oss.py

@@ -291,7 +291,7 @@ def test5(par):
                                 msg['error']='video uploading failure'        
                     if  upLoaded:       
                         if  typename=='结束': msg["status"]="success"
-                        else: msg["status"]="success_timeout"
+                        else: msg["status"]="timeout"
                     else:
                         msg["status"]='failed'    
                     time_s2 = time.time()                          

conf/master.json

@@ -1,8 +1,8 @@
 {
 "par":{
         "server1":"212.129.223.66:19092",
-        "server":"101.132.127.1:19092",
+        "server2":"101.132.127.1:19092",
-	"server3":"192.168.11.242:9092",
+	"server":"192.168.11.242:9092",
         "topic": ["alg-online-tasks","alg-offline-tasks","alg-task-results"],
         "group_id":"testWw",
         "kafka":"mintors/kafka",  

conf/send_oss.json

@@ -14,6 +14,6 @@
          "AId":"LTAI5tE7KWN9fsuGU7DyfYF4",
           "ASt":"yPPCyfsqWgrTuoz5H4sisY0COclx8E"
             },
-    "kafkaPar":{"boostServer1":["192.168.11.242:9092"]  ,"boostServer":["101.132.127.1:19092"],  "boostServer3":["212.129.223.66:9092"] ,"topic":"alg-task-results"} 
+    "kafkaPar":{"boostServer":["192.168.11.242:9092"]  ,"boostServer2":["101.132.127.1:19092"],  "boostServer3":["212.129.223.66:9092"] ,"topic":"alg-task-results"} 
 
 }

master.log

@@ -0,0 +1,2 @@
+{'server1': '212.129.223.66:19092', 'server2': '101.132.127.1:19092', 'server': '192.168.11.242:9092', 'topic': ['alg-online-tasks', 'alg-offline-tasks', 'alg-task-results'], 'group_id': 'testWw', 'kafka': 'mintors/kafka', 'modelJson': 'conf/model.json', 'logDir': 'logs/master', 'StreamWaitingTime': 240, 'logPrintInterval': 60}
+2022-07-19 15:11:14.016 [INFO][master:detector][434][logs.master.detector.log]- detector process starts

master.py

@@ -545,6 +545,13 @@ def detector(par):
                
         time.sleep(timeSleep)
         
+        ####检查gpu子进程是否结束，如果结束要join(),否则会产生僵尸进程###
+        #taskStatus['pidInfos'][gpuProcess.pid] = {'gpuProcess':gpuProcess,'type':'onLine','taskInfos':taskInfos}
+        for key in list(taskStatus['pidInfos'].keys()):
+            if not taskStatus['pidInfos'][key]['gpuProcess'].is_alive():
+                taskStatus['pidInfos'][key]['gpuProcess'].join()
+                taskStatus['pidInfos'].pop(key)
+                
     print('########Program End#####')
     
 

producer.py

@@ -12,8 +12,8 @@ def producer_demo():
     
     #topic_on='thsw';topic_off='thsw2';
     #server=['212.129.223.66:19092'];
-    server=['101.132.127.1:19092']
+    #server=['101.132.127.1:19092']
-
+    server=['192.168.11.242:9092']
     topic_on='alg-online-tasks';topic_off='alg-offline-tasks'
     # 假设生产的消息为键值对（不是一定要键值对），且序列化方式为json
     producer = KafkaProducer(
@@ -25,8 +25,8 @@ def producer_demo():
         #pull_channel = "rtmp://live.play.t-aaron.com/live/THSA"
         #push_channel = 'rtmp://live.push.t-aaron.com/live/THSB'
         #pull_channel = 'rtmp://live.play.t-aaron.com/live/THSAa_hd'
-        pull_channel = 'http://live.play.t-aaron.com/live/THSAa_hd.m3u8'
+        pull_channel = 'http://live.play.t-aaron.com/live/THSAl_hd.m3u8'
-        push_channel = "rtmp://live.push.t-aaron.com/live/THSBa"
+        push_channel = "rtmp://live.push.t-aaron.com/live/THSBd"
     else:
         pull_channel = "rtmp://demoplay.yunhengzhizao.cn/live/THSA_HD5M"
         push_channel = "rtmp://127.0.0.1:1935/live/test"