修复解码器内因线程退出顺序错误导致的显存泄漏问题；

添加count_running_task和close_all_task接口

修复解码器内因线程退出顺序错误导致的显存泄漏问题；
添加count_running_task和close_all_task接口
Hu Chunming
1 parent 190b4e40
Showing 9 changed files with 87 additions and 20 deletions
src/ai_platform/MultiSourceProcess.cpp
src/ai_platform/MultiSourceProcess.h
src/ai_platform/stl_aiplatform.cpp
src/ai_platform/stl_aiplatform.h
src/decoder/dvpp/DvppDataMemory.hpp
src/decoder/dvpp/DvppDecoder.cpp
src/decoder/dvpp/DvppDecoder.h
src/decoder/dvpp/FFRecoderTaskManager.cpp
src/demo/demo.cpp
@@ -462,7 +462,7 @@ bool CMultiSourceProcess::AddTask(task_param _cur_task_param){
       // if(algor_face != algor_map->end()){
       //   const algor_basic_config_param_t *cur_param = ((algor_init_config_param_t *)(algor_face->second))->basic_param;
       //   LOG_INFO("face_snapshot, result_folder: {}  result_folder_little: {}", cur_param->result_folder, cur_param->result_folder_little);
-      // }
+    // }
     }
     LOG_INFO("started task {} successed!", config.name);
@@ -470,6 +470,11 @@ bool CMultiSourceProcess::AddTask(task_param _cur_task_param){
     return true;
 }
+int CMultiSourceProcess::CountRunningTask() {
+  DecoderManager* pDecManager = DecoderManager::getInstance();
+  return pDecManager->count();
+}
+
 bool CMultiSourceProcess::task_has_vpt_algor(const std::string &task_id){
     //! TODO: create enum iterator.
     auto algor_map = m_task_param_manager->get_task_other_param(task_id);
@@ -684,6 +689,11 @@ void CMultiSourceProcess::CloseAllTask(){
     LOG_INFO("CloseAllTask exit.");
 }
+void CMultiSourceProcess::CloseAllTask2() {
+  DecoderManager* pDecManager = DecoderManager::getInstance();
+  pDecManager->closeAllDecoder();
+}
+
 void CMultiSourceProcess::clear_finished_task(){// 清理已经结束的任务
   std::lock_guard<std::mutex> l1(m_FinishedTaskMtx);
@@ -46,6 +46,9 @@ public:
   bool FinishTask(const string taskID);
   void CloseAllTask();
   int SnapShot(task_param param);
+  int CountRunningTask();
+
+  void CloseAllTask2();
 #ifdef POST_USE_RABBITMQ
   int AddMqConn(mq_type_t mq_type, rabbitmq_conn_params_t mq_conn_param);
@@ -95,6 +95,17 @@ int tsl_aiplatform_release(void **handle)
 	return SUCCESS;
 }
+int count_running_task(void *handle)
+{
+	CMultiSourceProcess* tools = (CMultiSourceProcess*)handle;
+	return tools->CountRunningTask();
+}
+
+void close_all_task(void *handle) 
+{
+	CMultiSourceProcess* tools = (CMultiSourceProcess*)handle;
+	tools->CloseAllTask2();
+}
 const char* get_tsl_aiplatform_version()
 {
@@ -120,6 +120,11 @@ extern &quot;C&quot;
 	TSL_AIPLATFORM_API int tsl_aiplatform_release(void **handle);
+	TSL_AIPLATFORM_API int count_running_task(void *handle);
+
+	TSL_AIPLATFORM_API void close_all_task(void *handle);
+
+
 	/*************************************************************************
 	* FUNCTION: get_tsl_aiplatform_version
 	* PURPOSE: 获取SDK版本号
@@ -25,7 +25,7 @@ public:
     ~DvppDataMemory(){
         if (pHwRgb) {
-            int ret = acldvppFree((uint8_t*)pHwRgb);
+            int ret = acldvppFree(pHwRgb);
             if(ret != ACL_ERROR_NONE){
                 cout << "acldvppFree failed" << endl;
             }
@@ -335,6 +335,7 @@ void DvppDecoder::read_thread() {
     int frame_count = 0;
 	int ret = -1;
+    m_bExitDecodeThd = false;
     pthread_t m_decode_thread;
     pthread_create(&m_decode_thread,0,
         [](void* arg)
@@ -429,8 +430,11 @@ void DvppDecoder::read_thread() {
 		}
 	}
-    m_bRunning=false;
+    while(m_bRunning && m_pktQueue.size() > 0) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(5));
+    }
+    m_bExitDecodeThd = true;
 	if(m_decode_thread != 0){
 		pthread_join(m_decode_thread,0);
 	}
@@ -444,6 +448,8 @@ void DvppDecoder::read_thread() {
     }
     m_pktQueue_mutex.unlock();
+    m_bRunning=false;
+
     if(decode_finished_cbk) {
         decode_finished_cbk(m_finishedDecArg);
     }
@@ -618,6 +624,7 @@ void DvppDecoder::decode_thread(){
 		return;
 	}
+    m_bExitDisplayThd = false;
     pthread_t display_thread;
     pthread_create(&display_thread,0,
         [](void* arg)
@@ -655,7 +662,7 @@ void DvppDecoder::decode_thread(){
         uint64_t frame_count = 0;
         bool bBreak = false;
-        while (m_bRunning) {
+        while (!m_bExitDecodeThd) {
             if (m_bPause){
                 std::this_thread::sleep_for(std::chrono::milliseconds(3));
                 continue;
@@ -694,18 +701,26 @@ void DvppDecoder::decode_thread(){
         CHECK_NOT_RETURN(aclvdecDestroyChannel(vdecChannelDesc), "aclvdecDestroyChannel failed");
     }while(0);
-    
-    CHECK_NOT_RETURN(aclvdecDestroyChannelDesc(vdecChannelDesc), "aclvdecDestroyChannelDesc failed");
+    // 退出 report thread
+    while(m_bRunning && m_vdecQueue.size() > 0) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(5));
+    }
+    CHECK_NOT_RETURN(aclvdecDestroyChannelDesc(vdecChannelDesc), "aclvdecDestroyChannelDesc failed");
 	// report_thread 需后于destroy退出
-	m_bRunning = false;
     m_bExitReportThd = true;
-	CHECK_NOT_RETURN(pthread_join(report_thread, nullptr), "pthread_join failed");
-    CHECK_NOT_RETURN(pthread_join(display_thread, nullptr), "pthread_join failed");
+	CHECK_NOT_RETURN(pthread_join(report_thread, nullptr), "report_thread join failed");
+
+    // 退出 display thread
+    while(m_bRunning && m_decoded_data_queue.size() > 0) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(5));
+    }
+    m_bExitDisplayThd = true;
+    CHECK_NOT_RETURN(pthread_join(display_thread, nullptr), "display_thread join failed");
     // 最后清理一遍未解码的数据
     m_vdecQueue_mutex.lock();
-    if(m_vdecQueue.size() > 0){
+    while(m_vdecQueue.size() > 0){
         void* inputData = m_vdecQueue.front();
         acldvppFree(inputData);
         inputData = nullptr;
@@ -872,7 +887,7 @@ bool DvppDecoder::sendVdecEos(aclvdecChannelDesc *vdecChannelDesc) {
 void DvppDecoder::display_thread(){
     LOG_INFO("[{}]- display_thread start...", m_dec_name);
-    while(m_bRunning) {
+    while(!m_bExitDisplayThd) {
         m_decoded_data_queue_mtx.lock();
         if(m_decoded_data_queue.size() <= 0) {
             m_decoded_data_queue_mtx.unlock();
@@ -898,7 +913,7 @@ void DvppDecoder::display_thread(){
         mem = nullptr;
     }
-    LOG_INFO("[{}]- display_thread end.", m_dec_name);
+    LOG_INFO("[{}]- display_thread exit.", m_dec_name);
 }
 void DvppDecoder::release_dvpp(){
@@ -83,7 +83,10 @@ private:
     bool m_bFinished{false};
     bool m_bRunning{false};
     bool m_bPause{false};
+
     bool m_bExitReportThd{false};
+    bool m_bExitDisplayThd{false};
+    bool m_bExitDecodeThd{false};
     // 读取数据
     AVStream* stream{nullptr};
@@ -92,8 +92,6 @@ void FFRecoderTaskManager::cache_pkt(AVPacket* pkt, long long frame_nb){
     newDataPkt->pkt = new_pkt;
     newDataPkt->frame_nb = frame_nb;
-    m_pkt_list.emplace_back(newDataPkt);
-
     if(is_key_frame(pkt)){
         // 越来越大的值
         newDataPkt->isKeyFrame = true;
@@ -102,6 +100,8 @@ void FFRecoderTaskManager::cache_pkt(AVPacket* pkt, long long frame_nb){
         newDataPkt->isKeyFrame = false;
     }
+    m_pkt_list.emplace_back(newDataPkt);
+
     std::lock_guard<std::mutex> l_info(m_recoderinfo_list_mtx);
     if(m_recoderinfo_list.size() <= 0){
         // 没有任务的时候，维持500的长度
@@ -1105,11 +1105,11 @@ void test_gpu(int gpuID){
     // createTask(handle, algor_vec, 1);  
     // createTask(handle, algor_vec, 2);  
     // createTask(handle, algor_vec, 3);  
-    createTask(handle, algor_vec3, 4);  
-    createTask(handle, algor_vec3, 5);  
-    createTask(handle, algor_vec3, 6);  
-    createTask(handle, algor_vec3, 7);  
-    createTask(handle, algor_vec3, 8);  
+    // createTask(handle, algor_vec3, 4);  
+    // createTask(handle, algor_vec3, 5);  
+    // createTask(handle, algor_vec3, 6);  
+    // createTask(handle, algor_vec3, 7);  
+    // createTask(handle, algor_vec3, 8);  
     // createTask(handle, algor_vec3, 9);  
     // createTask(handle, algor_vec3, 10);  
     // createTask(handle, algor_vec3, 11);  
@@ -1148,7 +1148,27 @@ void test_gpu(int gpuID){
     finish_task(handle, (char*)task_id2.data(), 0);
     finish_task(handle, (char*)task_id1.data(), 0);
   }*/
-    while (getchar() != 'q');
+
+  char ch = 'a';
+  while (ch != 'q') {
+    ch = getchar();
+    switch (ch)
+    {
+    case 'a':
+      createTask(handle, algor_vec3, 4, false);  
+      createTask(handle, algor_vec3, 5, false);  
+      createTask(handle, algor_vec3, 6, false);  
+      createTask(handle, algor_vec3, 7, false);  
+      createTask(handle, algor_vec3, 8, false);  
+      break;
+    case 'c':
+      close_all_task(handle);
+      break;
+    default:
+      break;
+    }
+    
+  }
     // finish_task(handle, (char*)task_id.data(), 0);