rpc_client.h 3.2 KB
Newer Older
G
gongweibao 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
// Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#pragma once

#include <string>
W
Wu Yi 已提交
18
#include "gflags/gflags.h"
G
gongweibao 已提交
19 20 21 22 23

#include "paddle/fluid/framework/data_type.h"
#include "paddle/fluid/framework/lod_tensor.h"
#include "paddle/fluid/framework/scope.h"

G
gongweibao 已提交
24
DECLARE_int32(rpc_deadline);
W
Wu Yi 已提交
25

G
gongweibao 已提交
26 27
namespace paddle {
namespace operators {
28
namespace distributed {
G
gongweibao 已提交
29 30 31

class RPCClient {
 public:
G
gongweibao 已提交
32 33
  RPCClient() {}
  virtual ~RPCClient() {}
G
gongweibao 已提交
34 35 36 37
  virtual bool AsyncSendVar(const std::string& ep,
                            const platform::DeviceContext& ctx,
                            const framework::Scope& scope,
                            const std::string& var_name,
G
gongweibao 已提交
38
                            int64_t time_out = FLAGS_rpc_deadline) = 0;
G
gongweibao 已提交
39 40 41 42 43

  virtual bool AsyncGetVar(const std::string& ep,
                           const platform::DeviceContext& ctx,
                           const framework::Scope& scope,
                           const std::string& var_name,
G
gongweibao 已提交
44
                           int64_t time_out = FLAGS_rpc_deadline) = 0;
G
gongweibao 已提交
45 46 47 48 49 50

  virtual bool AsyncPrefetchVar(const std::string& ep,
                                const platform::DeviceContext& ctx,
                                const framework::Scope& scope,
                                const std::string& in_var_name,
                                const std::string& out_var_name,
G
gongweibao 已提交
51
                                int64_t time_out = FLAGS_rpc_deadline) = 0;
G
gongweibao 已提交
52

G
gongweibao 已提交
53 54
  virtual void AsyncSendBatchBarrier(const std::string& ep,
                                     int64_t time_out = FLAGS_rpc_deadline) = 0;
G
gongweibao 已提交
55

G
gongweibao 已提交
56 57
  virtual void AsyncSendFetchBarrier(const std::string& ep,
                                     int64_t time_out = FLAGS_rpc_deadline) = 0;
G
gongweibao 已提交
58

T
tangwei12 已提交
59 60 61
  virtual void AsyncCheckpointNotify(const std::string& ep,
                                     const std::string& dir,
                                     int64_t time_out = FLAGS_rpc_deadline) = 0;
T
tangwei12 已提交
62

W
Wu Yi 已提交
63 64 65 66 67
  // SendComplete tells all the server that current trainer have no more data
  // to train, so that the pserver can reduce it's barrier count, and continue
  // to train with other trainers.
  virtual void SendComplete() = 0;

G
gongweibao 已提交
68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91
  virtual void Wait() = 0;

  template <typename T>
  static RPCClient* GetInstance() {
    std::call_once(init_flag_, &RPCClient::Init<T>);
    return rpc_client_.get();
  }

  // Init is called by GetInstance.
  template <typename T>
  static void Init() {
    if (rpc_client_.get() == nullptr) {
      rpc_client_.reset(new T());
      rpc_client_->InitImpl();
    }
  }

 protected:
  virtual void InitImpl() {}

 private:
  static std::once_flag init_flag_;
  static std::unique_ptr<RPCClient> rpc_client_;
};
92
}  // namespace distributed
G
gongweibao 已提交
93 94
}  // namespace operators
}  // namespace paddle