Move sync_mode device ctx from grpc server #10881

gongweibao · 2018-05-23T12:32:07Z

Fix part of #10804.

typhoonzero

Seems there are still compile errors, need to fix that first

typhoonzero · 2018-05-24T14:06:49Z

paddle/fluid/operators/detail/rpc_processor.h

+  const ReceivedMessage Get() { return var_recv_queue_.Pop(); }
+
+  void Push(const std::string& msg_name) {
+    var_recv_queue_.Push(std::make_pair(msg_name, nullptr));


maybe we can get rid of the queues, and have a member function Process which will do the "real" processing, like:

void Process(const std::string& msg_name, framework::Variable* var); virtual void ProcessImpl(const std::string& msg_name, framework::Variable* var);

Then, operators can define classes inherited from this class and write the ProcessImpl to do the job that when message arrives.

panyx0718 · 2018-05-24T14:25:09Z

paddle/fluid/operators/detail/grpc_server.cc

    } else {
-      request_.reset(new VariableResponse(scope, dev_ctx_, true));
+      request_.reset(new VariableResponse(rpc_processor->scope(),
+                                          rpc_processor->dev_ctx(), true));


request_.reset(new VariableResponse(rpc_processor->scope(), rpc_processor->dev_ctx(), !rpc_processor->sync_mode()));

panyx0718 · 2018-05-24T14:28:39Z

paddle/fluid/operators/detail/grpc_server.cc

-    executor_->RunPreparedContext(prefetch_ctx_, scope_);
-
-    SerializeToByteBuffer(var_name, var, *dev_ctx_, &reply_);
+    rpc_processor_->RequestPrefetch(request_.get(), &reply);


Done. Thanks!

panyx0718 · 2018-05-24T14:31:08Z

paddle/fluid/operators/detail/grpc_server.cc

+void AsyncGRPCServer::HandleRequest(::grpc::ServerCompletionQueue* cq,
+                                    int rpc_id,
+                                    std::function<void()> TryToRegisterNewOne) {
+  TryToRegisterNewOne();


this line is not necessary? We only register before the existing one is deleted?

Merge一时爽，debug火葬场！
Thanks!

Done.Thanks.

panyx0718 · 2018-05-24T14:44:11Z

paddle/fluid/operators/detail/grpc_server.cc

+
+    VLOG(3) << "HandleRequest for " << rpc_id << " get Next";
+
+    PADDLE_ENFORCE(tag);


the tag is now is an id starting from 0, smaller than the max buf size.

panyx0718 · 2018-05-24T14:48:53Z

paddle/fluid/operators/detail/grpc_server.cc

+    PADDLE_ENFORCE(tag);
+
+    if (rpc_processor_->sync_mode()) {
+      if (!is_shut_down_ &&


perhaps !is_shut_down_ should be in previous if? and this one should be if-else?

panyx0718 · 2018-05-24T14:53:49Z

paddle/fluid/operators/detail/grpc_server.cc

+void AsyncGRPCServer::WaitCond(const std::string& rpc_name) {
+  std::unique_lock<std::mutex> lock(this->barrier_mutex_);
+  auto it = barrier_.find(rpc_name);
+  assert(it != barrier_.end());


this 2 line is equal to barrier_.at()?

panyx0718 · 2018-05-24T15:00:01Z

paddle/fluid/operators/detail/grpc_server.cc

@@ -418,15 +396,29 @@ void AsyncGRPCServer::HandleRequest(
  }
 }

-void AsyncGRPCServer::WaitCond(int cond) {
+void AsyncGRPCServer::RegisterBarrier(const std::string& rpc_name,
+                                      int cond_id) {
  std::unique_lock<std::mutex> lock(this->barrier_mutex_);


do we really need a map? It seems we only have a few rpc methods. A enum is enough?

Done. A set is enough!

typhoonzero · 2018-05-28T05:39:42Z

paddle/fluid/operators/detail/rpc_processor.h

+  GRPCProcessorCtx() : exit_flag_(false), fan_in_(-1) { clear_to_init(); }
+  virtual ~GRPCProcessorCtx() {}
+
+  bool ProcessSendImpl(const std::string& msg_name, framework::Variable* var,


All handler functions can be saved in a unordered_map<std::string, std::function>, call the handler when RPC request arrives.

typhoonzero · 2018-05-28T05:41:55Z

paddle/fluid/operators/detail/rpc_processor.h

+  bool ProcessPrefetchImpl(const std::string& msg_name, framework::Scope* scope,
+                           framework::Variable** var);
+
+  void SetFanIn(int fan_in) { fan_in_ = fan_in; }


For easier to understand, can call this SetNumClients, then the WaitBarrier(std::string method) can wait all the clients have completed sending the barrier.

typhoonzero · 2018-05-28T05:43:40Z

paddle/fluid/operators/detail/rpc_processor.h

+
+  bool IsExit() {
+    std::unique_lock<std::mutex> lock(this->mutex_);
+    return exit_flag_;


Can make this atomic

exit_flag_ need be protected with the condition.

typhoonzero · 2018-05-28T05:45:51Z

paddle/fluid/operators/detail/rpc_processor.h

+  std::mutex mutex_;
+
+  // send
+  std::condition_variable condition_send_;


Only one condition is needed here for all status. The "condition" is the current "state" of the RCP server, condition will be changed only when client barrier reaches.

typhoonzero · 2018-05-28T05:49:17Z

paddle/fluid/operators/detail/rpc_processor.h

+
+ private:
+  // status
+  bool exit_flag_;


A handler class have no "exit_flag", should let RPC server do this, also the condition and state changing.

panyx0718 · 2018-05-28T06:53:05Z

paddle/fluid/operators/detail/grpc_server.cc


-    SerializeToByteBuffer(var_name, var, *dev_ctx_, &reply_);
+    local_scope_ = &rpc_processor_->scope()->NewScope();


Add a TODO to see if this local_scope_ is needed?

panyx0718 · 2018-05-28T07:03:39Z

paddle/fluid/operators/detail/grpc_server_test.cc

+  std::thread server_thread(
+      std::bind(&detail::AsyncGRPCServer::RunSyncUpdate, g_rpc_service.get()));
+
+  sleep(10);


Does WaitServerReady() work here?

panyx0718 · 2018-05-28T07:40:44Z

paddle/fluid/operators/detail/rpc_processor.cc

+namespace detail {
+
+void GRPCProcessorCtx::SetExit() {
+  LOG(WARNING) << "GRPCProcessorCtx SetExit!";


panyx0718 · 2018-05-28T08:01:38Z

paddle/fluid/operators/listen_and_serv_op.cc

-      exit_flag = true;
+  while (true) {
+    if (rpc_processor_->IsExit()) {
+      LOG(WARNING) << "get exit!rpc_processor break!";


panyx0718 · 2018-05-28T08:02:44Z

paddle/fluid/operators/listen_and_serv_op.cc

-      LOG(INFO) << "received terminate message and exit";
-      exit_flag = true;
+  while (true) {
+    if (rpc_processor_->IsExit()) {


has rpc_service been shutdown at this point? Maybe it should wait rpc_service?

Stop() call both of them exit. So can we judge only one?

panyx0718 · 2018-05-28T08:14:50Z

paddle/fluid/operators/detail/rpc_processor.cc

+      executor_->RunPreparedContext((*grad_to_prepared_ctx_)[msg_name].get(),
+                                    scope);
+    } catch (std::exception& e) {
+      LOG(ERROR) << "async: run sub program error " << e.what();


return false here?

Thanks. Done.

panyx0718 · 2018-05-28T08:16:25Z

paddle/fluid/operators/detail/rpc_processor.cc

+
+    if (var == nullptr) {
+      LOG(ERROR) << "sync: Can not find server side var: " << msg_name;
+      PADDLE_THROW("sync: Can not find server side var");


return false or throw?

Thanks. Done.

Yancey1989 · 2018-05-28T09:14:26Z

paddle/fluid/operators/detail/rpc_processor.cc

+  LOG(WARNING) << "GRPCProcessorCtx SetExit!";
+  std::unique_lock<std::mutex> lock(mutex_);
+  exit_flag_ = true;
+  condition_send_.notify_all();


It's better to unlock the mutex before executor notify_all, that will avoid deadlock. Reference https://baptiste-wicht.com/posts/2012/04/c11-concurrency-tutorial-advanced-locking-and-condition-variables.html

Done. Thanks!

Yancey1989 · 2018-05-28T09:28:00Z

paddle/fluid/operators/detail/rpc_processor.cc

+}
+
+void GRPCProcessorCtx::IncreaseBatchBarrierSend() {
+  std::unique_lock<std::mutex> lock(mutex_);


A condition variable should used with a mutex as a pair, and mutex_ is used anywhere, maybe you can use multiple mutex to project these condition variable.

Reserve only one! Thanks! Done!

Yancey1989 · 2018-05-28T09:33:00Z

paddle/fluid/operators/detail/grpc_server.h

+
+  GRPCProcessorCtx* rpc_processor_;
+
+  // barrier


Need more details about this comment, it's confusion with only one word barrier.

Change a readable name.
Thanks! Done!

typhoonzero · 2018-05-29T01:30:13Z

paddle/fluid/operators/detail/rpc_server.h

+  virtual void WaitCond(int cond) = 0;
+
+ protected:
+  std::string address_;


Maybe change to bind_address_ is better for a server?

typhoonzero · 2018-05-29T01:32:55Z

paddle/fluid/operators/detail/request_handler.h

+    VLOG(3) << "batch_barrier_:" << batch_barrier_;
+  }
+
+  void clear_to_init() {


Call this Reset

typhoonzero · 2018-05-29T01:35:22Z

paddle/fluid/operators/detail/request_handler.h

+  // barrier
+  void WaitBarrier() {
+    std::unique_lock<std::mutex> lock(this->mutex_);
+    condition_.wait(


Curious why "Barriers" can not put in RPCServer?

typhoonzero · 2018-05-29T02:42:24Z

paddle/fluid/operators/detail/request_handler.h

@@ -0,0 +1,158 @@
+// Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.


I've written an example at: https://gist.github.com/typhoonzero/fc2489231c8e3a29d9ac7c895c8f4aae, which shows how to abstract the rpcserver. This implement can actually remove the redundant "blocking queue" also.

Done.Thanks.

typhoonzero · 2018-05-30T05:10:14Z

Please fix the compile errors first

typhoonzero · 2018-05-30T11:10:04Z

paddle/fluid/operators/detail/request_handler.h

+  framework::Executor* executor() { return executor_; }
+  std::vector<framework::Variable*>& sparse_vars() { return sparse_vars_; }
+
+  // request handler


Maybe need detailed comments here, and also in the header files, so developers may get to know more about the interface.

typhoonzero · 2018-05-30T11:10:33Z

paddle/fluid/operators/detail/request_handler.h

+                     std::shared_ptr<framework::ExecutorPrepareContext>>*
+      grad_to_prepared_ctx_;
+
+  // get


typhoonzero · 2018-05-30T11:16:34Z

paddle/fluid/operators/detail/rpc_server.h

@@ -0,0 +1,91 @@
+// Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.


I think we can now move this file under fluid/framework as a general interface now.

Will do in next PR.

typhoonzero · 2018-05-30T11:27:04Z

paddle/fluid/operators/detail/rpc_server.cc

+}
+
+void RPCServer::SavePort() const {
+  // NOTE: default write file to /tmp/paddle.selected_port


Can delete this line now.

typhoonzero · 2018-05-30T11:28:40Z

paddle/fluid/operators/detail/rpc_server.cc

+                            RequestHandler* handler, int thread_num) {
+  rpc_call_map_[rpc_name] = handler;
+  rpc_thread_num_[rpc_name] = thread_num;
+  PADDLE_ENFORCE((thread_num >= 0 && thread_num <= 24),


24 is hard coded and maybe we don't need check from here?

typhoonzero · 2018-05-30T11:35:47Z

paddle/fluid/operators/detail/grpc_request_handler.cc

+namespace operators {
+namespace detail {
+
+bool GrpcRequestSendHandler::Handle(void* input, void* output) {


Maybe if we pass inputs below to the Handle function, so that the handler can not depend on the RPC implement:

bool Handle(const std::string& varname, Variable* var, Variable* outvar);

typhoonzero

LGTM basically, please fix those minor comments so that we can merge it.

typhoonzero · 2018-05-31T07:03:47Z

paddle/fluid/operators/detail/request_handler.h

+  //        SerializeToByteBuffer(varname, outvar,
+  //           *request_handler_->dev_ctx(), &reply_);
+  //    }
+  virtual bool Handle(const std::string& varname, framework::Scope* scope,


Curious we have a member scope and why need to pass a scope parameter?

RequestSend and RequestPrefetch uses the scope from VariableResponse.
Such as https://github.com/PaddlePaddle/Paddle/blob/develop/paddle/fluid/operators/detail/grpc_server.cc#L198.

typhoonzero · 2018-05-31T08:04:05Z

paddle/fluid/operators/detail/request_handler_impl.h

+namespace operators {
+namespace detail {
+
+class RequestSendHandler final : public RequestHandler {


request_handler_impl.cc seems not quiet large, so can omit the .cc file or just put it in the listen_and_serv operator.

Others depends them.

typhoonzero · 2018-05-31T08:08:27Z

paddle/fluid/operators/detail/grpc_server_test.cc

+  std::thread server_thread(
+      std::bind(&detail::AsyncGRPCServer::StartServer, g_rpc_service.get()));
+
+  // FIXME(gongwb): don't use hard time.


Call WaitServerReady

Should wait for data transformation end. So I add FIXME here.

typhoonzero · 2018-05-31T08:13:15Z

paddle/fluid/operators/detail/grpc_server.h

  void ShutdownQueue();
+  void ShutDownImpl() override;


ShutDownImpl, ShutdownQueue, HandleRequest these two functions can be private? They are only used in grpc_server class.

typhoonzero · 2018-05-31T08:15:49Z

paddle/fluid/operators/detail/request_handler.h

+  // The implemention is in request_handler_impl.
+  // example:
+  //    std::string varname = request_.varname();
+  //    VLOG(3) << "RequestGet " << varname;


Example have no need to add VLOG

typhoonzero · 2018-05-31T08:18:52Z

paddle/fluid/operators/listen_and_serv_op.cc

-      rpc_service_->ShutDown();
-      break;
-    }
+    sleep(1);


Why sleep1?

To avoid deed loop.

typhoonzero · 2018-05-31T08:19:37Z

paddle/fluid/operators/listen_and_serv_op.cc

-  rpc_service_.reset(new detail::AsyncGRPCServer(endpoint, sync_mode));
+  // request_handler_.reset(new detail::GRPCRequestSendHandler(sync_mode));
+  rpc_service_.reset(new detail::AsyncGRPCServer(endpoint, fan_in));
+  request_send_handler_.reset(new detail::RequestSendHandler(sync_mode));


request_send_handler => send_request_handler ?

和RequestSendHandler对应。

typhoonzero

LGTM, we can merge this work now finnally.

gongweibao added 3 commits May 23, 2018 07:59

init

fe4c8da

add

9885c8c

compile ok

c3a8c05

gongweibao changed the title ~~[WIP]Move sync_mode device ctx from grpc server~~ [WIP]Move sync_mode device ctx from grpc server May 23, 2018

gongweibao added 4 commits May 24, 2018 01:46

move queue

4c45baf

fix global var bug

42c2ad0

add rpc_id

33495e6

fix shared ptr bug

244617b

gongweibao changed the title ~~[WIP]Move sync_mode device ctx from grpc server~~ Move sync_mode device ctx from grpc server May 24, 2018

merge

64cb0e3

gongweibao requested review from jacquesqiao, Yancey1989, typhoonzero and panyx0718 May 24, 2018 05:33

typhoonzero reviewed May 24, 2018

View reviewed changes

panyx0718 reviewed May 24, 2018

View reviewed changes

gongweibao added 14 commits May 25, 2018 02:09

follow panxin comments

fe74105

add cond id

3a6fdd0

clean up queue

cfcffdb

fix gen_nccl

a556088

fix gen_nccl

16fd836

fix one

63c5e7c

can shut down

dc5f373

not find

6c8dc00

fix shutdown

80f081c

clean up

8665134

temp grc_server_test

bd1c538

fix req_id

fe0f4aa

add mutex

a653b27

fix mem leak

806bcd4

typhoonzero reviewed May 28, 2018

View reviewed changes

panyx0718 reviewed May 28, 2018

View reviewed changes

Yancey1989 reviewed May 28, 2018

View reviewed changes

gongweibao added 6 commits May 28, 2018 13:28

follow comments

c9a39a2

follow comments

b6d9fc4

fix gen

6503dac

fix get bug

1d8b37a

fix unittest

c72e911

fix get bug

941738a

typhoonzero reviewed May 29, 2018

View reviewed changes

gongweibao added 7 commits May 29, 2018 11:25

compile ok

d01aa77

add log

276044c

clean up

8d171a3

add more reqs

1cf100e

add paddle enforce

2e4fc55

merge

1cc99ff

add rpc_server.cc

f6ce3e3

gongweibao added 3 commits May 30, 2018 07:10

fix bug

9771071

fix sigterm

cdcafb2

clean up

ee91b3f

typhoonzero reviewed May 30, 2018

View reviewed changes

follow comments

8aa4921

typhoonzero reviewed May 31, 2018

View reviewed changes

gongweibao added 2 commits May 31, 2018 08:20

fix dead lock bug

9993820

follow comments

324c234

typhoonzero approved these changes Jun 1, 2018

View reviewed changes

gongweibao merged commit 4fb7cc7 into PaddlePaddle:develop Jun 1, 2018

gongweibao deleted the cleanprefetch branch June 11, 2018 10:53


		VLOG(3) << "HandleRequest for " << rpc_id << " get Next";

		PADDLE_ENFORCE(tag);


		SerializeToByteBuffer(var_name, var, *dev_ctx_, &reply_);
		local_scope_ = &rpc_processor_->scope()->NewScope();

		@@ -0,0 +1,158 @@
		// Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.

		@@ -0,0 +1,91 @@
		// Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.

Move sync_mode device ctx from grpc server #10881

Move sync_mode device ctx from grpc server #10881

Conversation

gongweibao commented May 23, 2018 • edited Loading

typhoonzero left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

typhoonzero commented May 30, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

typhoonzero left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gongweibao commented May 23, 2018 •

edited

Loading