wip

paddle/fluid/operators/detail/grpc_server.cc

@@ -241,6 +241,12 @@ void AsyncGRPCServer::RunSyncUpdate() {
   t_prefetch_.reset(new std::thread(
       std::bind(&AsyncGRPCServer::HandleRequest, this, cq_prefetch_.get(),
                 "cq_prefetch", prefetch_register)));
+
+  {
+    std::lock_guard<std::mutex> lock(this->mutex_ready_);
+    ready_ = 1;
+  }
+  condition_ready_.notify_all();
   // wait server
   server_->Wait();
   t_send_->join();

paddle/fluid/operators/detail/grpc_server.h

@@ -45,8 +45,9 @@ class RequestBase;
 class AsyncGRPCServer final {
  public:
   explicit AsyncGRPCServer(const std::string &address, bool sync_mode)
-      : address_(address), sync_mode_(sync_mode) {}
+      : address_(address), sync_mode_(sync_mode), ready_(0) {}
 
+  bool WaitServerReady();
   void RunSyncUpdate();
 
   // functions to sync server barrier status.
@@ -118,6 +119,10 @@ class AsyncGRPCServer final {
   framework::ProgramDesc *program_;
   framework::Executor *executor_;
   int selected_port_;
+
+  std::mutext mutex_ready_;
+  std::condition_variable condition_ready_;
+  int ready_;
 };
 
 };  // namespace detail

paddle/fluid/operators/listen_and_serv_op.cc

@@ -265,6 +265,23 @@ void ListenAndServOp::RunAsyncLoop(framework::Executor *executor,
   }  // while(true)
 }
 
+void ListenAndServOp::StartServerThread() {
+  server_thread_.reset(new std::thread(
+      std::bind(&ListenAndServOp::ServerThreadEntry, this, rpc_service_)));
+}
+
+void ListenAndServOp::ServerThreadEntry(
+    std::shared_ptr<detail::AsyncGRPCServer> service) {
+  service->RunSyncUpdate();
+  VLOG(4) << "RunServer thread end";
+
+  {
+    std::lock_guard<std::mutex> lock(this->barrier_mutex_);
+    barrier_cond_step_ = cond;
+  }
+  barrier_condition_.notify_all();
+}
+
 void ListenAndServOp::RunImpl(const framework::Scope &scope,
                               const platform::Place &dev_place) const {
   platform::DeviceContextPool &pool = platform::DeviceContextPool::Instance();
@@ -298,7 +315,7 @@ void ListenAndServOp::RunImpl(const framework::Scope &scope,
   // start the server listening after all member initialized.
   server_thread_.reset(new std::thread(RunServer, rpc_service_));
   VLOG(3) << "wait server thread to become ready...";
-  sleep(5);
+
   // Write to a file of server selected port for python use.
   SavePort(rpc_service_);
   if (sync_mode) {

paddle/fluid/operators/listen_and_serv_op.h

@@ -51,6 +51,10 @@ class ListenAndServOp : public framework::OperatorBase {
                     framework::Scope* recv_scope,
                     framework::BlockDesc* prefetch_block) const;
 
+  void StartServerThread();
+
+  void ServerThreadEntry();
+
   void Stop() override;
 
   void RunImpl(const framework::Scope& scope,
@@ -59,6 +63,8 @@ class ListenAndServOp : public framework::OperatorBase {
  protected:
   mutable std::shared_ptr<detail::AsyncGRPCServer> rpc_service_;
   mutable std::shared_ptr<std::thread> server_thread_;
+  std::mutext server_ready_mutex_;
+  std::condition_variable server_ready_;
 };
 
 }  // namespace operators