Merge pull request #13224 from jacquesqiao/cherry-pick-async-handle-complete-message

fix async mode handle COMPLETE_MESSAGE (#13212)

paddle/fluid/operators/distributed/request_handler_impl.cc

@@ -39,8 +39,17 @@ bool RequestSendHandler::Handle(const std::string& varname,
                                 const std::string& out_var_name) {
   VLOG(4) << "RequestSendHandler:" << varname;
 
+  // Sync
+  if (varname == BATCH_BARRIER_MESSAGE) {
+    VLOG(3) << "sync: recv BATCH_BARRIER_MESSAGE";
+    rpc_server_->IncreaseBatchBarrier(kRequestSend);
+  } else if (varname == COMPLETE_MESSAGE) {
+    VLOG(3) << "sync: recv complete message";
+    rpc_server_->Complete();
+  } else {
     // Async
     if (!sync_mode_) {
+      VLOG(3) << "async process var: " << varname;
       rpc_server_->Profiler().OneStep();
       try {
         executor_->RunPreparedContext((*grad_to_prepared_ctx_)[varname].get(),
@@ -50,17 +59,7 @@ bool RequestSendHandler::Handle(const std::string& varname,
         return false;
       }
       return true;
-  }
-
-  // Sync
-  if (varname == BATCH_BARRIER_MESSAGE) {
-    VLOG(3) << "sync: recv BATCH_BARRIER_MESSAGE";
-    rpc_server_->IncreaseBatchBarrier(kRequestSend);
-  } else if (varname == COMPLETE_MESSAGE) {
-    VLOG(3) << "sync: recv complete message";
-    rpc_server_->Complete();
-  } else {
-    VLOG(3) << "sync: received var_name: " << varname;
+    } else {  // sync
       rpc_server_->WaitCond(kRequestSend);
       VLOG(3) << "sync: processing received var: " << varname;
 
@@ -68,11 +67,13 @@ bool RequestSendHandler::Handle(const std::string& varname,
         LOG(FATAL) << "sync: Can not find server side var: " << varname;
         return false;
       }
+
       if (invar->IsType<framework::SelectedRows>()) {
         std::unique_lock<std::mutex> lock(mutex_sparse_vars_);
         sparse_vars_.push_back(invar);
       }
     }
+  }
   return true;
 }