fleet.h 12.4 KB
Newer Older
T
tangwei12 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25
/* Copyright (c) 2020 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 <atomic>
#include <ctime>
#include <map>
#include <memory>
#include <random>
#include <string>
#include <unordered_map>
#include <vector>

26 27
#include "paddle/fluid/distributed/ps/service/communicator/communicator_common.h"
#include "paddle/fluid/distributed/ps/service/ps_service/service.h"
T
tangwei12 已提交
28 29 30 31 32 33 34 35 36
#include "paddle/fluid/framework/archive.h"
#include "paddle/fluid/framework/io/fs.h"
#include "paddle/fluid/framework/io/shell.h"
#include "paddle/fluid/framework/program_desc.h"
#include "paddle/fluid/framework/scope.h"
#include "paddle/fluid/framework/tensor.h"
#include "paddle/fluid/framework/variable_helper.h"
#include "paddle/fluid/platform/macros.h"  // for DISABLE_COPY_AND_ASSIGN

37 38 39 40 41 42 43 44
namespace paddle {
namespace framework {
class Scope;
class SelectedRows;
class Variable;
}  // namespace framework
}  // namespace paddle

T
tangwei12 已提交
45 46 47
namespace paddle {
namespace distributed {

48 49
class PSCore;

T
tangwei12 已提交
50 51 52
using framework::LoDTensor;
using framework::Scope;
using framework::Variable;
53
using phi::SelectedRows;
T
tangwei12 已提交
54 55 56

using RpcCtxMap = std::unordered_map<std::string, CommContext>;

Z
zhaocaibei123 已提交
57
class FleetWrapper {
T
tangwei12 已提交
58 59 60 61 62 63 64 65 66 67 68 69 70 71
 public:
  virtual ~FleetWrapper() {}
  FleetWrapper() {
    scale_sparse_gradient_with_batch_size_ = true;
    // trainer sleep some time for pserver core dump
    sleep_seconds_before_fail_exit_ = 300;
    // pserver request server timeout ms
    client2client_request_timeout_ms_ = 500000;
    // pserver connect server timeout_ms
    client2client_connect_timeout_ms_ = 10000;
    // pserver request max retry
    client2client_max_retry_ = 3;
  }

72 73 74 75 76 77 78 79 80 81
  // TODO(zhaocaibei123: later)
  int32_t CopyTable(const uint64_t src_table_id, const uint64_t dest_table_id);

  int32_t CopyTableByFeasign(const uint64_t src_table_id,
                             const uint64_t dest_table_id,
                             const std::vector<uint64_t>& feasign_list);

  typedef std::function<void(int, int)> HeterCallBackFunc;
  int RegisterHeterCallback(HeterCallBackFunc handler);

T
tangwei12 已提交
82
  // set client to client communication config
83 84
  void SetClient2ClientConfig(int request_timeout_ms,
                              int connect_timeout_ms,
T
tangwei12 已提交
85 86 87 88 89
                              int max_retry);

  // Pull sparse variables from server in sync mode
  // Param<in>: scope, table_id, var_names, fea_keys, fea_dim, var_emb_names
  // Param<out>: fea_values
90 91
  void PullSparseVarsSync(const Scope& scope,
                          const uint64_t table_id,
T
tangwei12 已提交
92 93 94 95 96 97
                          const std::vector<std::string>& var_names,
                          std::vector<uint64_t>* fea_keys,
                          std::vector<std::vector<float>>* fea_values,
                          int fea_dim,
                          const std::vector<std::string>& var_emb_names);

98 99 100 101
  // Pull sparse variables from server in async mode
  // Param<in>: scope, table_id, var_names, fea_keys, fea_dim
  // Param<out>: fea_values std::future
  std::future<int32_t> PullSparseVarsAsync(
102 103
      const Scope& scope,
      const uint64_t table_id,
104 105
      const std::vector<std::string>& var_names,
      std::vector<uint64_t>* fea_keys,
106 107
      std::vector<std::vector<float>>* fea_values,
      int fea_dim);
108

T
tangwei12 已提交
109 110
  // Pull sparse variables from server in sync mode
  // pull immediately to tensors
111 112 113
  // is_training is true means training, false means inference, the behavior is
  // different on pserver

114 115 116 117
  void PullSparseToTensorSync(const uint64_t table_id,
                              int fea_dim,
                              uint64_t padding_id,
                              platform::Place place,
118
                              bool is_training,
T
tangwei12 已提交
119 120 121 122 123 124
                              std::vector<const LoDTensor*>* inputs,  // NOLINT
                              std::vector<LoDTensor*>* outputs);      // NOLINT

  // pull dense variables from server in sync mod
  // Param<in>: scope, table_id, var_names
  // Param<out>: void
125 126
  void PullDenseVarsSync(const Scope& scope,
                         const uint64_t table_id,
T
tangwei12 已提交
127 128 129 130 131
                         const std::vector<std::string>& var_names);

  // pull dense variables from server in async mod
  // Param<in>: scope, table_id, var_names
  // Param<out>: pull_dense_status
132 133
  void PullDenseVarsAsync(const Scope& scope,
                          const uint64_t table_id,
T
tangwei12 已提交
134 135 136 137 138
                          const std::vector<std::string>& var_names,
                          std::vector<std::future<int32_t>>* pull_dense_status,
                          bool in_cpu);

  // push dense parameters(not gradients) to server in sync mode
139 140
  void PushDenseParamSync(const Scope& scope,
                          const uint64_t table_id,
T
tangwei12 已提交
141 142
                          const std::vector<std::string>& var_names);

143 144
  void PushDenseVarsAsync(const Scope& scope,
                          const uint64_t table_id,
T
tangwei12 已提交
145 146
                          const std::vector<std::string>& var_names,
                          std::vector<std::future<int32_t>>* push_sparse_status,
147 148
                          float scale_datanorm,
                          int batch_size);
T
tangwei12 已提交
149 150

  // push dense variables to server in sync mode
151 152
  void PushDenseVarsSync(Scope* scope,
                         const uint64_t table_id,
T
tangwei12 已提交
153 154 155
                         const std::vector<std::string>& var_names);

  void PushSparseVarsAsync(
156 157 158
      const Scope& scope,
      const uint64_t table_id,
      const std::string& grad,
T
tangwei12 已提交
159 160 161 162 163 164
      std::vector<std::future<int32_t>>* push_sparse_status);
  // This is specially designed for click/show stats in server
  // Param<in>: scope, table_id, fea_keys, fea_labels, sparse_key_names,
  //            sparse_grad_names, batch_size, use_cvm, dump_slot
  // Param<out>: push_values, push_sparse_status
  void PushSparseVarsWithLabelAsync(
165 166
      const Scope& scope,
      const uint64_t table_id,
T
tangwei12 已提交
167 168 169
      const std::vector<uint64_t>& fea_keys,
      const std::vector<float>& fea_labels,
      const std::vector<std::string>& sparse_key_names,
170 171
      const std::vector<std::string>& sparse_grad_names,
      const int emb_dim,
T
tangwei12 已提交
172 173
      std::vector<std::vector<float>>* push_values,
      std::vector<std::future<int32_t>>* push_sparse_status,
174 175 176 177 178
      const int batch_size,
      const bool use_cvm,
      const bool dump_slot,
      std::vector<uint64_t>* sparse_push_keys,
      const bool no_cvm);
T
tangwei12 已提交
179 180 181

  // Push sparse variables to server in async mode
  void PushSparseFromTensorWithLabelAsync(
182 183 184 185 186 187 188 189
      const Scope& scope,
      const uint64_t table_id,
      int fea_dim,
      uint64_t padding_id,
      bool scale_sparse,
      const std::string& accesor,
      const std::string& click_name,
      platform::Place place,
T
tangwei12 已提交
190 191 192
      const std::vector<std::string>& input_names,
      std::vector<const LoDTensor*>* inputs,    // NOLINT
      std::vector<const LoDTensor*>* outputs);  // NOLINT
193

194 195 196 197
  void PushSparseFromTensorAsync(const uint64_t table_id,
                                 int fea_dim,
                                 uint64_t padding_id,
                                 platform::Place place,
Z
zhaocaibei123 已提交
198
                                 std::vector<const LoDTensor*>* inputs,
199
                                 std::vector<int>& slots,  // NOLINT
Z
zhaocaibei123 已提交
200 201
                                 const LoDTensor* shows,
                                 const LoDTensor* clicks,
202 203
                                 std::vector<LoDTensor*>* outputs,
                                 bool use_cvm_op = false);
T
tangwei12 已提交
204 205 206 207 208
  // Push sparse variables to server in Async mode
  // Param<In>: scope, table_id, fea_keys, sparse_grad_names
  // Param<Out>: push_values, push_sparse_status

  // init server
209 210
  void LoadSparseOnServer(const std::string& path,
                          const std::string& meta,
T
tangwei12 已提交
211 212 213 214
                          uint32_t table_id);
  // init server
  // void InitServer(const std::string& dist_desc,
  //                 const std::vector<uint64_t>& host_sign_list, int index);
215 216
  void InitServer(
      const std::string& dist_desc,
217 218 219
      const std::vector<std::string>& host_sign_list,
      int index,
      int trainers,
220
      const std::vector<framework::ProgramDesc>& server_sub_program = {});
T
tangwei12 已提交
221 222
  // init trainer
  void InitWorker(const std::string& dist_desc,
223 224
                  const std::vector<std::string>& host_sign_list,
                  int index);
T
tangwei12 已提交
225 226 227 228 229 230 231 232 233

  // stop server
  void StopServer();
  // finalize worker to make worker can be stop
  void FinalizeWorker();
  // run server with ip port
  uint64_t RunServer(const std::string& ip, uint32_t port);
  // get client info
  std::vector<uint64_t> GetClientsInfo();
234 235
  // set client info
  int SetClients(std::vector<uint64_t>& host_sign_list);  // NOLINT
T
tangwei12 已提交
236 237 238 239 240 241 242 243 244 245 246
  // create client to client connection
  void CreateClient2ClientConnection();
  // flush all push requests
  void ClientFlush();

  // barrier with barrier table
  void BarrierWithTable(uint32_t barrier_type);

  void PrintTableStat(const uint64_t table_id);
  // mode = 0, load all feature
  // mode = 1, load delta feature, which means load diff
Z
zhaocaibei123 已提交
247
  void LoadModel(const std::string& path, const int mode);
T
tangwei12 已提交
248 249
  // mode = 0, load all feature
  // mode = 1, load delta feature, which means load diff
250 251
  void LoadModelOneTable(const uint64_t table_id,
                         const std::string& path,
T
tangwei12 已提交
252 253 254 255 256 257
                         const int mode);
  // mode = 0, save all feature
  // mode = 1, save delta feature, which means save diff
  void SaveModel(const std::string& path, const int mode);
  // mode = 0, save all feature
  // mode = 1, save delta feature, which means save diff
258 259
  void SaveModelOneTable(const uint64_t table_id,
                         const std::string& path,
T
tangwei12 已提交
260
                         const int mode);
261 262 263 264

  // recv table from server and save it in LodTensor
  void RecvAndSaveTable(const uint64_t table_id, const std::string& path);

T
tangwei12 已提交
265 266 267 268 269
  // clear all models, release their memory
  void ClearModel();
  // clear one table
  void ClearOneTable(const uint64_t table_id);
  // shrink sparse table
270
  void ShrinkSparseTable(int table_id, int threshold);
T
tangwei12 已提交
271
  // shrink dense table
272 273 274 275
  void ShrinkDenseTable(int table_id,
                        Scope* scope,
                        std::vector<std::string> var_list,
                        float decay,
T
tangwei12 已提交
276 277 278 279 280 281
                        int emb_dim);

  typedef std::function<int32_t(int, int, const std::string&)> MsgHandlerFunc;
  // register client to client communication
  int RegisterClientToClientMsgHandler(int msg_type, MsgHandlerFunc handler);
  // send client to client message
282 283
  std::future<int32_t> SendClientToClientMsg(int msg_type,
                                             int to_client_id,
T
tangwei12 已提交
284 285 286 287 288 289 290 291 292 293 294 295
                                             const std::string& msg);

  // FleetWrapper singleton
  static std::shared_ptr<FleetWrapper> GetInstance() {
    if (NULL == s_instance_) {
      s_instance_.reset(new paddle::distributed::FleetWrapper());
    }
    return s_instance_;
  }
  // this performs better than rand_r, especially large data
  std::default_random_engine& LocalRandomEngine();

296 297 298
  // for init worker
  void InitGFlag(const std::string& gflags);

Z
zhaocaibei123 已提交
299
  double GetCacheThreshold(int table_id);
300 301 302
  void CacheShuffle(int table_id,
                    const std::string& path,
                    const int mode,
Z
zhaocaibei123 已提交
303 304
                    const double cache_threshold);
  int32_t SaveCache(int table_id, const std::string& path, const int mode);
Z
zhaocaibei123 已提交
305 306
  void Revert();
  void CheckSavePrePatchDone();
Z
zhaocaibei123 已提交
307

T
tangwei12 已提交
308
  static std::shared_ptr<paddle::distributed::PSCore> pserver_ptr_;
309
  static std::shared_ptr<paddle::distributed::PSClient> worker_ptr_;
T
tangwei12 已提交
310 311 312

 private:
  static std::shared_ptr<FleetWrapper> s_instance_;
313
  paddle::distributed::PaddlePSEnvironment ps_env_;
314 315 316
  size_t GetAbsoluteSum(size_t start,
                        size_t end,
                        size_t level,
T
tangwei12 已提交
317 318 319 320
                        const framework::LoD& lod);

 protected:
  static bool is_initialized_;
Z
zhaocaibei123 已提交
321
  std::map<uint64_t, std::vector<paddle::distributed::Region>> regions_;
T
tangwei12 已提交
322 323 324 325 326 327 328 329 330 331
  bool scale_sparse_gradient_with_batch_size_;
  int32_t sleep_seconds_before_fail_exit_;
  int client2client_request_timeout_ms_;
  int client2client_connect_timeout_ms_;
  int client2client_max_retry_;
  DISABLE_COPY_AND_ASSIGN(FleetWrapper);
};

}  // end namespace distributed
}  // end namespace paddle