Fix nccl trace

paddle/fluid/distributed/collective/process_group_nccl.cc

@@ -114,6 +114,11 @@ ProcessGroupNCCL::ProcessGroupNCCL(
       pg_timeout_(timeout) {
   LOG(INFO) << "ProcessGroupNCCL pg_timeout_ " << pg_timeout_;
 }
+ProcessGroupNCCL::~ProcessGroupNCCL() {
+  LOG(INFO) << "ProcessGroupNCCL destruct ";
+  auto& comm_task_manager = phi::distributed::CommTaskManager::GetInstance();
+  comm_task_manager.Stop();
+}
 
 void ProcessGroupNCCL::GroupStart() {
   NCCL_CHECK(phi::dynload::ncclGroupStart());

paddle/fluid/distributed/collective/process_group_nccl.h

@@ -78,6 +78,7 @@ class ProcessGroupNCCL final : public ProcessGroupWithStream {
                    int size,
                    int gid,
                    int64_t timeout = 20 * 1000);
+  ~ProcessGroupNCCL();
 
   std::string GetBackendName() const override { return "NCCL"; }
 

paddle/phi/core/distributed/comm_task_manager.cc

@@ -74,6 +74,17 @@ void CommTaskManager::CommTaskEnqueue(std::shared_ptr<CommTask> comm_task) {
   }
 }
 
+void CommTaskManager::Stop() {
+  terminated_.store(true);
+
+  LOG(INFO) << "CommTaskManager stopped begin.";
+  if (comm_task_loop_thread_.joinable()) {
+    comm_task_loop_thread_.join();
+    comm_task_list_cv_.notify_one();
+  }
+  LOG(INFO) << "CommTaskManager stopped.";
+}
+
 void CommTaskManager::CommTaskLoop() {
   bool done = false;
   while (!terminated_.load() || !done) {
@@ -100,7 +111,11 @@ void CommTaskManager::CommTaskLoop() {
         }
         iter = comm_task_list_.erase(iter);
       } else {
-        ++iter;
+        if (task->IsStarted() && task->IsCompleted()) {
+          iter = comm_task_list_.erase(iter);
+        } else {
+          ++iter;
+        }
       }
     }
 
@@ -131,6 +146,8 @@ void CommTaskManager::CommTaskLoop() {
     if (comm_task_list_.empty() && init_comm_task_map_.empty() &&
         start_comm_task_map_.empty()) {
       done = true;
+    } else {
+      done = false;
     }
   }
 }

paddle/phi/core/distributed/comm_task_manager.h

@@ -47,6 +47,7 @@ class CommTaskManager {
   }
 
   void CommTaskEnqueue(std::shared_ptr<CommTask> comm_task);
+  void Stop();
 
  private:
   void CommTaskLoop();