Fix grpc bugs

cmake/external/grpc.cmake

@@ -33,7 +33,7 @@ ExternalProject_Add(
     extern_grpc
     DEPENDS protobuf zlib
     GIT_REPOSITORY "https://github.com/grpc/grpc.git"
-    GIT_TAG "v1.7.x"
+    GIT_TAG "v1.8.x"
     PREFIX          ${GRPC_SOURCES_DIR}
     UPDATE_COMMAND  ""
     CONFIGURE_COMMAND ""

paddle/operators/detail/grpc_client.cc

@@ -87,7 +87,7 @@ bool RPCClient::AsyncGetVariable(const std::string& ep,
   return true;
 }
 
-bool RPCClient::wait() {
+bool RPCClient::Wait() {
   bool ok = true;
 
   while (true) {
@@ -96,7 +96,6 @@ bool RPCClient::wait() {
     }
 
     if (!Proceed()) {
-      LOG(ERROR) << "Get meets CompletionQueue error";
       return false;
     }
   }
@@ -110,22 +109,25 @@ bool RPCClient::Proceed() {
 
   // request counts.
   if (!cq_.Next(&tag, &ok)) {
+    LOG(ERROR) << "Get meets CompletionQueue error";
     return false;
   }
-  req_count_--;
 
   GPR_ASSERT(ok);
   PADDLE_ENFORCE(tag);
 
   // TODO(gongwb): add more retries.
   ClientBase* c = static_cast<ClientBase*>(tag);
   if (!c->status_.ok()) {
+    LOG(ERROR) << "proc param error:" << c->var_h_.String()
+               << " grpc error:" << c->status_.error_message();
     delete c;
-    return true;
+    return false;
   }
 
   c->Process();
   delete c;
+  req_count_--;
   return true;
 }
 
@@ -135,8 +137,12 @@ std::shared_ptr<grpc::Channel> RPCClient::GetChannel(const std::string& ep) {
     return it->second;
   }
 
+  grpc::ChannelArguments args;
+  args.SetMaxSendMessageSize(std::numeric_limits<int>::max());
+  args.SetMaxReceiveMessageSize(std::numeric_limits<int>::max());
+
   auto ch = std::shared_ptr<grpc::Channel>(
-      grpc::CreateChannel(ep, grpc::InsecureChannelCredentials()));
+      grpc::CreateCustomChannel(ep, grpc::InsecureChannelCredentials(), args));
 
   channels_[ep] = ch;
   return ch;

paddle/operators/detail/grpc_client.h

@@ -130,7 +130,7 @@ class RPCClient {
                         const framework::Scope& scope,
                         const std::string& var_name,
                         int64_t time_out = 600 * 1000);
-  bool wait();
+  bool Wait();
 
  private:
   bool Proceed();

paddle/operators/detail/grpc_server.cc

@@ -28,12 +28,15 @@ class RequestBase {
  public:
   explicit RequestBase(sendrecv::SendRecvService::AsyncService* service,
                        grpc::ServerCompletionQueue* cq)
-      : service_(service), cq_(cq), status_(PROCESS) {}
+      : service_(service), cq_(cq), status_(PROCESS) {
+    PADDLE_ENFORCE(cq_);
+  }
   virtual ~RequestBase() {}
   virtual void Process() { assert(false); }
 
   CallStatus Status() { return status_; }
   void SetStatus(CallStatus status) { status_ = status; }
+  virtual std::string GetReqName() { assert(false); }
 
  protected:
   grpc::ServerContext ctx_;
@@ -56,12 +59,14 @@ class RequestSend final : public RequestBase {
 
   virtual ~RequestSend() {}
 
+  virtual std::string GetReqName() { return request_.varname(); }
+
   virtual void Process() {
     MessageWithName msg_with_name =
         std::make_pair(request_.varname(), std::move(request_));
     queue_->Push(std::move(msg_with_name));
-    // TODO(gongwb): check var's info.
     responder_.Finish(reply_, grpc::Status::OK, this);
+    status_ = FINISH;
   }
 
  protected:
@@ -81,13 +86,16 @@ class RequestGet final : public RequestBase {
 
   virtual ~RequestGet() {}
 
+  virtual std::string GetReqName() { return request_.varname(); }
+
   virtual void Process() {
     // proc request.
     std::string var_name = request_.varname();
     auto* var = scope_->FindVar(var_name);
     SerializeToMessage(var_name, var, platform::CPUDeviceContext(), &reply_);
     // TODO(gongwb): check var's info.
     responder_.Finish(reply_, grpc::Status::OK, this);
+    status_ = FINISH;
   }
 
  protected:
@@ -100,6 +108,8 @@ class RequestGet final : public RequestBase {
 void AsyncGRPCServer::RunSyncUpdate() {
   grpc::ServerBuilder builder;
   builder.AddListeningPort(address_, grpc::InsecureServerCredentials());
+  builder.SetMaxSendMessageSize(std::numeric_limits<int>::max());
+  builder.SetMaxReceiveMessageSize(std::numeric_limits<int>::max());
   builder.RegisterService(&service_);
 
   cq_send_ = builder.AddCompletionQueue();
@@ -159,18 +169,6 @@ void AsyncGRPCServer::TryToRegisterNewGetOne() {
   VLOG(4) << "create Requestget status:" << get->Status();
 }
 
-void AsyncGRPCServer::SetFinishOrDelete(RequestBase*& last) {
-  std::unique_lock<std::mutex> lock(cq_mutex_);
-  if (is_shut_down_) {
-    delete last;
-    last = NULL;
-    return;
-  }
-
-  last->SetStatus(FINISH);
-  return;
-}
-
 void AsyncGRPCServer::HandleRequest(bool wait, grpc::ServerCompletionQueue* cq,
                                     std::string cq_name,
                                     std::function<void()> TryToRegisterNewOne) {
@@ -184,13 +182,19 @@ void AsyncGRPCServer::HandleRequest(bool wait, grpc::ServerCompletionQueue* cq,
       break;
     }
 
+    PADDLE_ENFORCE(tag);
     if (wait && !done_) {
       Wait();
     }
 
     RequestBase* base = (RequestBase*)tag;
+    // reference:
+    // https://github.com/tensorflow/tensorflow/issues/5596
+    // https://groups.google.com/forum/#!topic/grpc-io/xftlRy-IQwM
+    // https://groups.google.com/forum/#!topic/grpc-io/ywATt88Ef_I
     if (!ok) {
-      VLOG(4) << cq_name << " recv no regular event";
+      LOG(WARNING) << cq_name << " recv no regular event:argument name"
+                   << base->GetReqName();
       TryToRegisterNewOne();
       delete base;
       continue;
@@ -201,7 +205,6 @@ void AsyncGRPCServer::HandleRequest(bool wait, grpc::ServerCompletionQueue* cq,
         VLOG(4) << cq_name << " status:" << base->Status();
         TryToRegisterNewOne();
         base->Process();
-        SetFinishOrDelete(base);
         break;
       }
       case FINISH: {

paddle/operators/detail/grpc_server.h

@@ -60,7 +60,6 @@ class AsyncGRPCServer final : public sendrecv::SendRecvService::Service {
                      std::function<void()> TryToRegisterNewOne);
   void TryToRegisterNewSendOne();
   void TryToRegisterNewGetOne();
-  void SetFinishOrDelete(RequestBase *&last);
   void ShutdownQueue();
 
  private:

paddle/operators/recv_op.cc

@@ -96,6 +96,8 @@ class RecvOp : public framework::OperatorBase {
     rpc_service_->Reset();
     // TODO(typhoonzero): change this to a while_op for every cluster-batch.
     bool exit_flag = false;
+    VLOG(4) << "param_count:" << param_count
+            << " trainer_count:" << trainer_count;
     while (!exit_flag) {
       // TODO(gognwb): simply this loop.
       // Get from multiple trainers, we don't care about order in which

paddle/operators/send_op.cc

@@ -48,7 +48,7 @@ class SendOp : public framework::OperatorBase {
       client_.AsyncGetVariable(epmap[i], ctx, scope, outs[i]);
     }
 
-    client_.wait();
+    PADDLE_ENFORCE(client_.Wait());
   }
 
  private: