communicator.cc 37.7 KB
Newer Older
Q
Qiao Longfei 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/* Copyright (c) 2019 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. */

#include "paddle/fluid/operators/distributed/communicator.h"
Q
Qiao Longfei 已提交
16
#include <gflags/gflags.h>
17
#include <paddle/fluid/framework/program_desc.h>
Q
Qiao Longfei 已提交
18
#include <chrono>  // NOLINT
19
#include <map>
Q
Qiao Longfei 已提交
20
#include <thread>  // NOLINT
21
#include <unordered_set>
Q
Qiao Longfei 已提交
22
#include "paddle/fluid/framework/eigen.h"
Q
Qiao Longfei 已提交
23 24
#include "paddle/fluid/framework/selected_rows.h"
#include "paddle/fluid/framework/tensor_util.h"
25
#include "paddle/fluid/framework/threadpool.h"
Q
Qiao Longfei 已提交
26
#include "paddle/fluid/framework/variable_helper.h"
C
Chengmo 已提交
27
#include "paddle/fluid/operators/distributed/distributed.h"
Q
Qiao Longfei 已提交
28 29 30
#include "paddle/fluid/operators/distributed/parameter_recv.h"
#include "paddle/fluid/operators/distributed/parameter_send.h"

31 32 33
DECLARE_int32(communicator_max_merge_var_num);
DECLARE_int32(communicator_send_queue_size);

Q
Qiao Longfei 已提交
34 35
DEFINE_bool(communicator_independent_recv_thread, true,
            "use an independent to recv vars from parameter server");
36
DEFINE_int32(communicator_min_send_grad_num_before_recv, 20,
37
             "max grad num to send before recv parameters");
38
DEFINE_int32(communicator_thread_pool_size, 5, "thread num to do send or recv");
Q
Qiao Longfei 已提交
39 40 41
DEFINE_int32(communicator_send_wait_times, 5,
             "times that send thread will wait if merge num does not reach "
             "max_merge_var_num");
42 43
DEFINE_bool(communicator_fake_rpc, false,
            "fake mode does not really send any thing");
44 45
DEFINE_bool(communicator_merge_sparse_grad, true,
            "merge sparse gradient before sending");
46 47
DEFINE_int32(communicator_merge_sparse_bucket, 2000,
             "number of threads for sparse var");
Q
Qiao Longfei 已提交
48

Q
Qiao Longfei 已提交
49 50 51 52
namespace paddle {
namespace operators {
namespace distributed {

Q
Qiao Longfei 已提交
53 54 55 56 57 58
inline double GetCurrentUS() {
  struct timeval time;
  gettimeofday(&time, NULL);
  return 1e+6 * time.tv_sec + time.tv_usec;
}

59 60 61 62 63 64 65
template <typename T>
inline void VSUB(int n, const T *x, const T *y, T *z) {
  for (int i = 0; i < n; ++i) {
    z[i] = x[i] - y[i];
  }
}

T
tangwei12 已提交
66
std::once_flag Communicator::init_flag_;
67
std::shared_ptr<Communicator> Communicator::communicator_(nullptr);
Q
can run  
Qiao Longfei 已提交
68

T
tangwei12 已提交
69 70 71 72 73 74 75
void AsyncCommunicator::InitImpl(const RpcCtxMap &send_varname_to_ctx,
                                 const RpcCtxMap &recv_varname_to_ctx,
                                 Scope *recv_scope) {
  send_varname_to_ctx_ = std::move(send_varname_to_ctx);
  recv_varname_to_ctx_ = std::move(recv_varname_to_ctx);
  recv_scope_ = std::move(recv_scope);

Q
Qiao Longfei 已提交
76 77 78 79 80
  // get all send information from graph, build vars_to_send
  VLOG(0) << "communicator_independent_recv_thread: "
          << FLAGS_communicator_independent_recv_thread;
  VLOG(0) << "communicator_send_queue_size: "
          << FLAGS_communicator_send_queue_size;
81 82
  VLOG(0) << "communicator_min_send_grad_num_before_recv: "
          << FLAGS_communicator_min_send_grad_num_before_recv;
Q
Qiao Longfei 已提交
83 84
  VLOG(0) << "communicator_thread_pool_size: "
          << FLAGS_communicator_thread_pool_size;
85 86
  VLOG(0) << "communicator_send_wait_times: "
          << FLAGS_communicator_send_wait_times;
Q
Qiao Longfei 已提交
87
  VLOG(0) << "communicator_max_merge_var_num: "
88 89
          << FLAGS_communicator_max_merge_var_num;
  VLOG(0) << "communicator_fake_rpc: " << FLAGS_communicator_fake_rpc;
90 91
  VLOG(0) << "communicator_merge_sparse_grad: "
          << FLAGS_communicator_merge_sparse_grad;
92 93
  VLOG(0) << "communicator_is_sgd_optimizer: "
          << FLAGS_communicator_is_sgd_optimizer;
94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112

  if (send_varname_to_ctx.size() == 0) {
    VLOG(0) << "nothing need to be send, will not start send_thread";
  } else {
    send_scope_.reset(new Scope());
    for (auto &iter : send_varname_to_ctx_) {
      send_varname_to_queue_[iter.first] =
          std::make_shared<BlockingQueue<std::shared_ptr<Variable>>>(
              FLAGS_communicator_send_queue_size);
    }
    send_threadpool_.reset(
        new ::ThreadPool(FLAGS_communicator_thread_pool_size));
  }

  if (recv_varname_to_ctx.size() == 0) {
    VLOG(0) << "nothing need to be received, will not start recv_thread";
  } else {
    recv_threadpool_.reset(
        new ::ThreadPool(FLAGS_communicator_thread_pool_size));
Q
Qiao Longfei 已提交
113 114 115
  }
}

T
tangwei12 已提交
116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132
void AsyncCommunicator::InitImpl(const paddle::framework::ProgramDesc &program,
                                 Scope *param_scope) {
  using RpcCtxMap = operators::distributed::RpcCtxMap;
  VLOG(3) << "ProcessGraph";
  RpcCtxMap send_varname_to_ctx;
  RpcCtxMap recv_varname_to_ctx;
  for (auto *op : program.Block(0).AllOps()) {
    VLOG(3) << "node name " << op->Type();
    if (op->Type() == "send") {
      auto send_var_name = op->Input("X")[0];
      auto send_varnames = boost::get<std::vector<std::string>>(
          op->GetNullableAttr("send_varnames"));
      auto epmap =
          boost::get<std::vector<std::string>>(op->GetNullableAttr("epmap"));
      auto height_section =
          boost::get<std::vector<int64_t>>(op->GetNullableAttr("sections"));
      auto trainer_id = boost::get<int>(op->GetNullableAttr("trainer_id"));
1
123malin 已提交
133 134 135 136 137 138
      auto merge_add = boost::get<bool>(op->GetNullableAttr("merge_add"));
      if (!merge_add) {
        merge_add = FLAGS_communicator_is_sgd_optimizer;
      }
      auto use_send_handler =
          boost::get<bool>(op->GetNullableAttr("use_send_handler"));
T
tangwei12 已提交
139
      send_varname_to_ctx[send_var_name] = operators::distributed::RpcContext(
1
123malin 已提交
140 141
          send_var_name, send_varnames, epmap, height_section, trainer_id,
          merge_add, use_send_handler);
T
tangwei12 已提交
142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167
      VLOG(3) << "find and init an send op: "
              << send_varname_to_ctx[send_var_name];
    } else if (op->Type() == "recv") {
      auto do_not_run = boost::get<int>(op->GetNullableAttr("do_not_run"));
      PADDLE_ENFORCE_GT(do_not_run, 0, "recv should not run!");
      auto recv_var_name = op->Output("Out")[0];
      auto recv_varnames = boost::get<std::vector<std::string>>(
          op->GetNullableAttr("recv_varnames"));
      auto epmap =
          boost::get<std::vector<std::string>>(op->GetNullableAttr("epmap"));
      auto trainer_id = boost::get<int>(op->GetNullableAttr("trainer_id"));
      recv_varname_to_ctx[recv_var_name] = operators::distributed::RpcContext(
          recv_var_name, recv_varnames, epmap, {}, trainer_id);
    }
  }

  // init communicator here
  if (send_varname_to_ctx.size() == 0 && recv_varname_to_ctx.size() == 0) {
    LOG(WARNING) << "no var need to send and recv!!";
  }

  operators::distributed::AsyncCommunicator::InitImpl(
      send_varname_to_ctx, recv_varname_to_ctx, param_scope);
}

AsyncCommunicator::~AsyncCommunicator() {
168 169 170 171
  if (FLAGS_v >= 3) {
    std::string msg("~Communicator");
    fwrite(msg.c_str(), msg.length(), 1, stdout);
  }
Q
Qiao Longfei 已提交
172 173 174
  running_ = false;
  if (send_thread_) send_thread_->join();
  if (recv_thread_) recv_thread_->join();
175 176 177 178
  if (FLAGS_v >= 3) {
    std::string msg("~Communicator done");
    fwrite(msg.c_str(), msg.length(), 1, stdout);
  }
Q
Qiao Longfei 已提交
179 180
}

T
tangwei12 已提交
181
void AsyncCommunicator::SendThread() {
Q
Qiao Longfei 已提交
182
  VLOG(3) << "SendThread start!";
Q
Qiao Longfei 已提交
183 184 185
  while (running_) {
    std::vector<std::future<void>> task_futures;
    task_futures.reserve(send_varname_to_ctx_.size());
Q
Qiao Longfei 已提交
186
    VLOG(3) << "run send graph";
Q
Qiao Longfei 已提交
187
    auto before_run_send_graph = GetCurrentUS();
Q
Qiao Longfei 已提交
188
    for (auto &iter : send_varname_to_queue_) {
Q
Qiao Longfei 已提交
189 190
      auto &var_name = iter.first;
      auto &var_queue = iter.second;
Q
Qiao Longfei 已提交
191
      if (var_queue->Size() > 0) {
Q
Qiao Longfei 已提交
192
        auto send_task = [this, &var_name, &var_queue] {
Q
Qiao Longfei 已提交
193
          VLOG(3) << var_name << " merge and send";
Q
Qiao Longfei 已提交
194
          std::vector<std::shared_ptr<Variable>> vars;
195 196
          int merged_var_num = 0;
          int wait_times = 0;
Q
Qiao Longfei 已提交
197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214
          while (merged_var_num < FLAGS_communicator_max_merge_var_num) {
            if (var_queue->Size() == 0) {
              VLOG(3) << "wait_times -> " << wait_times;
              if (wait_times >= FLAGS_communicator_send_wait_times) {
                break;
              }
              std::this_thread::sleep_for(std::chrono::milliseconds(10));
              wait_times++;
              continue;
            } else {
              wait_times = 0;

              vars.push_back(var_queue->Pop());
              // only count the send number of the first var
              if (var_name == send_varname_to_queue_.begin()->first) {
                grad_num_.fetch_add(1, std::memory_order_relaxed);
              }
              merged_var_num++;
215
            }
Q
Qiao Longfei 已提交
216
          }
Q
Qiao Longfei 已提交
217
          auto before_merge = GetCurrentUS();
1
123malin 已提交
218 219 220 221 222 223 224
          auto &ctx = send_varname_to_ctx_.at(var_name);
          if (ctx.use_send_handler) {
            MergeVars<float>(var_name, vars, send_scope_.get(), ctx.merge_add);
          } else {
            MergeVars<int64_t>(var_name, vars, send_scope_.get(),
                               ctx.merge_add);
          }
Q
Qiao Longfei 已提交
225
          auto after_merge = GetCurrentUS();
Q
Qiao Longfei 已提交
226 227
          VLOG(3) << "merge " << merged_var_num << " " << var_name
                  << " use time " << after_merge - before_merge;
Q
Qiao Longfei 已提交
228
          auto send_functor = distributed::ParameterSend<float>();
229
          if (!FLAGS_communicator_fake_rpc) {
230
            send_functor(ctx, *send_scope_, true, 1);
231
          }
Q
Qiao Longfei 已提交
232 233 234
          auto after_send = GetCurrentUS();
          VLOG(3) << "send " << var_name << " use time "
                  << after_send - after_merge;
Q
Qiao Longfei 已提交
235 236 237
        };
        task_futures.emplace_back(
            send_threadpool_->enqueue(std::move(send_task)));
Q
Qiao Longfei 已提交
238
      } else {
239
        VLOG(4) << var_name << " queue empty";
Q
Qiao Longfei 已提交
240
      }
Q
Qiao Longfei 已提交
241 242 243
    }
    for (auto &task_f : task_futures) {
      task_f.wait();
Q
Qiao Longfei 已提交
244
    }
Q
Qiao Longfei 已提交
245
    auto after_run_send_graph = GetCurrentUS();
246 247 248

    VLOG(3) << "run send graph use time "
            << after_run_send_graph - before_run_send_graph;
T
tangwei12 已提交
249
    Recv();
Q
Qiao Longfei 已提交
250
  }
251
  VLOG(0) << "communicator stopped, send thread exit";
Q
Qiao Longfei 已提交
252 253
}

T
tangwei12 已提交
254
void AsyncCommunicator::RecvThread() {
Q
Qiao Longfei 已提交
255
  VLOG(3) << "RecvThread start!";
Q
Qiao Longfei 已提交
256
  while (running_) {
257
    int grad_num = grad_num_.load();
258
    if (grad_num > FLAGS_communicator_min_send_grad_num_before_recv) {
259 260 261 262 263 264
      VLOG(1) << "current grad num " << grad_num;
      RecvAll();
      grad_num_.store(0);
    } else {
      std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
Q
Qiao Longfei 已提交
265
  }
266
  VLOG(0) << "communicator stopped, recv thread exit";
Q
Qiao Longfei 已提交
267 268
}

T
tangwei12 已提交
269 270
void AsyncCommunicator::Send(const std::string &var_name,
                             const framework::Scope &scope) {
Q
Qiao Longfei 已提交
271 272 273 274
  VLOG(3) << "communicator send " << var_name;
  // push var into send queue by var_name
  auto *grad_var = scope.FindVar(var_name);
  PADDLE_ENFORCE(grad_var->IsInitialized(), "grad var should be inited");
275 276 277 278 279
  if (grad_var->IsType<framework::SelectedRows>() &&
      !FLAGS_communicator_merge_sparse_grad) {
    auto send_functor = distributed::ParameterSend<float>();
    auto &ctx = send_varname_to_ctx_.at(var_name);
    if (!FLAGS_communicator_fake_rpc) {
280
      send_functor(ctx, scope, true, 1);
281 282 283 284 285 286 287 288
    }
  } else {
    auto tmp_grad_var = std::make_shared<Variable>();
    framework::CopyVariable(*grad_var, tmp_grad_var.get());
    auto &queue = send_varname_to_queue_.at(var_name);
    VLOG(3) << "send " << var_name << " queue size " << queue->Size();
    queue->Push(tmp_grad_var);
  }
Q
Qiao Longfei 已提交
289 290
}

T
tangwei12 已提交
291 292 293
void AsyncCommunicator::Recv() {
  if (FLAGS_communicator_independent_recv_thread) {
    return;
294 295
  }

T
tangwei12 已提交
296 297 298 299 300 301
  auto grad_num = grad_num_.load();
  if (grad_num > 0) {
    RecvAll();
    grad_num_.store(0);
  } else {
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
302 303 304
  }
}

T
tangwei12 已提交
305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326
void AsyncCommunicator::RecvAll() {
  VLOG(3) << "parallel run recv graph";
  if (!running_) return;
  auto before_send = GetCurrentUS();
  std::vector<std::future<void>> task_futures;
  task_futures.reserve(recv_varname_to_ctx_.size());
  for (auto &iter : recv_varname_to_ctx_) {
    auto recv_task = [this, &iter] {
      auto &var_name = iter.first;
      VLOG(4) << "recv var " << var_name;
      auto recv_functor = distributed::ParameterRecv<float>();
      if (!FLAGS_communicator_fake_rpc) {
        recv_functor(iter.second, *recv_scope_);
      }
    };
    task_futures.emplace_back(recv_threadpool_->enqueue(std::move(recv_task)));
  }
  for (auto &task : task_futures) {
    task.wait();
  }
  auto after_recv = GetCurrentUS();
  VLOG(1) << "run recv graph use time " << after_recv - before_send;
327 328
}

T
tangwei12 已提交
329
void AsyncCommunicator::Start() {
330 331 332 333 334 335 336 337
  VLOG(0) << "Communicator start";
  if (!communicator_) {
    VLOG(0) << "Communicator is not inited, do nothing";
  } else {
    VLOG(1) << "start send thread and recv thread";
    running_ = true;
    // start send and recv thread
    send_thread_.reset(
T
tangwei12 已提交
338
        new std::thread(std::bind(&AsyncCommunicator::SendThread, this)));
339 340
    if (FLAGS_communicator_independent_recv_thread) {
      recv_thread_.reset(
T
tangwei12 已提交
341
          new std::thread(std::bind(&AsyncCommunicator::RecvThread, this)));
342 343 344 345
    }
  }
}

T
tangwei12 已提交
346
void AsyncCommunicator::Stop() {
347 348 349 350 351 352 353 354 355 356 357 358 359 360 361
  VLOG(0) << "Communicator stop";
  running_ = false;
  if (!communicator_) {
    VLOG(0) << "Communicator is not inited, do nothing";
  } else {
    if (send_thread_) {
      VLOG(1) << "stop send thread";
      send_thread_->join();
      send_thread_.reset(nullptr);
    }
    if (recv_thread_) {
      VLOG(1) << "stop recv thread";
      recv_thread_->join();
      recv_thread_.reset(nullptr);
    }
Q
Qiao Longfei 已提交
362
  }
363
  VLOG(0) << "Communicator stop done";
Q
Qiao Longfei 已提交
364 365
}

366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447
void AsyncCommunicator::Send(const std::vector<std::string> &sparse_var_names,
                             const std::vector<std::string> &sparse_var_tables,
                             const framework::Scope &scope) {}

void AsyncCommunicator::InitImpl(
    const paddle::framework::ProgramDesc &program, Scope *param_scope,
    std::map<std::string, std::map<std::string, std::vector<std::string>>>
        &vars_info,
    const int &trainers, const int &geo_need_push_nums) {}

GeoSgdCommunicator::~GeoSgdCommunicator() {
  if (FLAGS_v >= 3) {
    std::string msg("~Geo Sgd Communicator");
    fwrite(msg.c_str(), msg.length(), 1, stdout);
  }
  running_ = false;
  if (send_thread_) send_thread_->join();
  if (FLAGS_v >= 3) {
    std::string msg("~Geo Sgd Communicator done");
    fwrite(msg.c_str(), msg.length(), 1, stdout);
  }
}

void GeoSgdCommunicator::InitImpl(
    const paddle::framework::ProgramDesc &program, Scope *training_scope,
    std::map<std::string, std::map<std::string, std::vector<std::string>>>
        &vars_info,
    const int &trainers, const int &geo_need_push_nums) {
  training_scope_ = std::move(training_scope);
  trainer_nums_ = std::move(trainers);
  geo_need_push_nums_ = std::move(geo_need_push_nums);

  // get all send information from graph, build vars_to_send
  VLOG(0) << "communicator_independent_recv_thread: "
          << FLAGS_communicator_independent_recv_thread;
  VLOG(0) << "communicator_send_queue_size: "
          << FLAGS_communicator_send_queue_size;
  VLOG(0) << "communicator_min_send_grad_num_before_recv: "
          << FLAGS_communicator_min_send_grad_num_before_recv;
  VLOG(0) << "communicator_thread_pool_size: "
          << FLAGS_communicator_thread_pool_size;
  VLOG(0) << "communicator_send_wait_times: "
          << FLAGS_communicator_send_wait_times;
  VLOG(0) << "communicator_max_merge_var_num: "
          << FLAGS_communicator_max_merge_var_num;
  VLOG(0) << "communicator_fake_rpc: " << FLAGS_communicator_fake_rpc;
  VLOG(0) << "communicator_merge_sparse_grad: "
          << FLAGS_communicator_merge_sparse_grad;
  VLOG(0) << "Trainer nums: " << trainer_nums_;
  VLOG(0) << "geo_sgd_push_before_local_train_nums: " << geo_need_push_nums_;
  VLOG(0) << "communicator_merge_sparse_bucket "
          << FLAGS_communicator_merge_sparse_bucket;

  // process var info from transpiler
  for (auto &iter : vars_info) {
    // change var name in delta scope: "var" -> "var.delta"
    std::string var_name = iter.first;
    std::string send_var_name = VarToDeltaVar(var_name);
    std::vector<std::string> vars_names = iter.second["var_names"];
    std::vector<std::string> send_var_names;
    for (auto origin_var_name : vars_names) {
      send_var_names.push_back(VarToDeltaVar(origin_var_name));
    }

    // get vars section for split
    std::vector<std::string> vars_sections_str = iter.second["sections"];
    std::vector<int64_t> vars_sections_int = {};
    for (std::string str : vars_sections_str) {
      int64_t str2i = std::stol(str.c_str());
      vars_sections_int.push_back(str2i);
    }

    std::vector<std::string> vars_epmap = iter.second["epmap"];

    // record var is sparse or not
    bool is_sparse = iter.second["is_sparse"].front() == std::string("True");
    var_list_[var_name] = is_sparse;

    send_varname_to_ctx_[send_var_name] = operators::distributed::RpcContext(
        send_var_name, send_var_names, vars_epmap, vars_sections_int, 0);
    recv_varname_to_ctx_[var_name] = operators::distributed::RpcContext(
        var_name, vars_names, vars_epmap, vars_sections_int, 0);
C
Chengmo 已提交
448 449 450 451 452 453 454 455

    // record sparse section
    if (is_sparse) {
      need_thread_nums_ +=
          send_varname_to_ctx_[send_var_name].height_sections.size();
      absolute_section_[var_name] = operators::ToAbsoluteSection(
          send_varname_to_ctx_[send_var_name].height_sections);
    }
456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524
  }

  if (send_varname_to_ctx_.size() == 0 && recv_varname_to_ctx_.size() == 0) {
    LOG(WARNING) << "no var need to send and recv!!";
  }

  send_threadpool_.reset(new ::ThreadPool(FLAGS_communicator_thread_pool_size));
  need_push_queue_ =
      std::make_shared<BlockingQueue<std::shared_ptr<SparseIdsMap>>>(
          geo_need_push_nums);
  delta_scope_.reset(new Scope());
  old_scope_.reset(new Scope());
  pserver_scope_.reset(new Scope());
}

void GeoSgdCommunicator::Start() {
  VLOG(0) << "Geo Sgd Communicator start";
  if (!communicator_) {
    VLOG(0) << "Geo Sgd Communicator is not inited, do nothing";
  } else {
    VLOG(0) << "start send thread ";
    running_ = true;
    // start send and recv thread
    send_thread_.reset(
        new std::thread(std::bind(&GeoSgdCommunicator::SendThread, this)));
  }
}

void GeoSgdCommunicator::Stop() {
  VLOG(0) << "Geo Sgd Communicator stop";
  running_ = false;
  if (!communicator_) {
    VLOG(0) << "Geo Sgd Communicator is not inited, do nothing";
  } else {
    if (send_thread_) {
      VLOG(1) << "stop send thread";
      send_thread_->join();
      send_thread_.reset(nullptr);
    }
  }
  VLOG(0) << "Geo Sgd Communicator stop done";
}

void GeoSgdCommunicator::Send(const std::string &var_name,
                              const framework::Scope &scope) {
  // when execute trainer startup program, recv parameter from pserver
  // training_scope & pserver_scope param will copy it
  if (var_name == "param_init") {
    for (auto &iter : var_list_) {
      // For sparse param, old_scope store LoDTensor,
      // pserver_scope store SelectedRows.
      auto local_var_name = iter.first;
      if (var_list_[local_var_name] == true) {
        GeoSgdSparseParamInit(training_scope_, pserver_scope_.get(),
                              local_var_name);
      } else {
        GeoSgdDenseParamInit(training_scope_, pserver_scope_.get(),
                             local_var_name);
      }
      GeoSgdDenseParamInit(training_scope_, old_scope_.get(), local_var_name);
    }
  }
}

void GeoSgdCommunicator::Send(const std::vector<std::string> &sparse_var_names,
                              const std::vector<std::string> &sparse_var_tables,
                              const framework::Scope &scope) {
  // SparseIdsMap = std::unordered_map<std::string,std::unordered_set<int64_t>>
  std::shared_ptr<SparseIdsMap> ids_table = std::make_shared<SparseIdsMap>();
C
Chengmo 已提交
525
  auto before_run_send = GetCurrentUS();
526 527 528
  for (size_t i = 0; i < sparse_var_tables.size(); i++) {
    if (ids_table->find(sparse_var_tables[i]) == ids_table->end()) {
      // create empty set for new sparse var
C
Chengmo 已提交
529 530 531 532 533 534
      auto splited_var_nums =
          recv_varname_to_ctx_[sparse_var_tables[i]].splited_var_names.size();
      ids_table->insert(
          std::pair<std::string, std::vector<std::unordered_set<int64_t>>>(
              sparse_var_tables[i],
              std::vector<std::unordered_set<int64_t>>{splited_var_nums}));
535 536 537 538 539 540
    }
    auto *var = scope.FindVar(sparse_var_names[i]);
    auto var_tensor = var->Get<framework::LoDTensor>();
    int element_number = var_tensor.numel();
    int *var_mutable_data = var_tensor.mutable_data<int>(var_tensor.place());
    // insert ids which has not been record
541
    for (int j = 0; j < element_number; j++) {
C
Chengmo 已提交
542 543 544
      auto ep_idx = GetSectionIndex(var_mutable_data[j],
                                    absolute_section_[sparse_var_tables[i]]);
      ids_table->at(sparse_var_tables[i])[ep_idx].insert(var_mutable_data[j]);
545 546 547 548 549
      VLOG(4) << "Sparse var " << sparse_var_tables[i] << " insert "
              << var_mutable_data[j];
    }
  }
  need_push_queue_->Push(ids_table);
C
Chengmo 已提交
550 551
  auto after_run_send = GetCurrentUS();
  VLOG(3) << "run send_op use time " << after_run_send - before_run_send;
552 553 554 555 556 557 558 559 560 561
}

void GeoSgdCommunicator::SendThread() {
  VLOG(0) << "SendThread start!";
  auto before_run_training = GetCurrentUS();

  while (running_) {
    std::vector<std::future<void>> task_futures;
    task_futures.reserve(send_varname_to_ctx_.size());

562
    int wait_times = 0;
C
Chengmo 已提交
563
    while (ids_send_vec_.size() < geo_need_push_nums_) {
564 565
      VLOG(4) << "ids_send_vec_ Size: " << ids_send_vec_.size();
      if (need_push_queue_->Size() > 0) {
C
Chengmo 已提交
566
        wait_times = 0;
567 568
        ids_send_vec_.push_back(*(need_push_queue_->Pop()));
        VLOG(4) << "ids_send_vec_ pushed";
C
Chengmo 已提交
569 570 571 572 573 574 575 576
      } else if (need_push_queue_->Size() == 0) {
        VLOG(3) << "wait_times -> " << wait_times;
        if (wait_times >= FLAGS_communicator_send_wait_times) {
          break;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
        wait_times++;
        continue;
577 578 579 580 581 582 583 584 585 586 587 588
      }
    }

    if (ids_send_vec_.size() >= geo_need_push_nums_) {
      auto after_run_training = GetCurrentUS();
      VLOG(3) << "run Training use time "
              << after_run_training - before_run_training;
      before_run_training = GetCurrentUS();
      VLOG(3) << "Start send after get need_push_num";

      for (auto &iter : send_varname_to_ctx_) {
        auto &var_name = iter.first;
C
Chengmo 已提交
589 590 591 592 593 594 595 596 597 598 599 600 601 602 603
        if (var_list_[DeltaVarToVar(var_name)] == true) {
          // sparse var: merge->send->recv
          for (auto &splited_var_name : iter.second.splited_var_names) {
            auto send_task = [this, &var_name, &splited_var_name] {
              auto before_run_geo = GetCurrentUS();
              auto ids_set =
                  SparseIdsMerge(ids_send_vec_, var_name, splited_var_name);
              SendUpdateSparseVars(var_name, splited_var_name, ids_set);
              RecvUpdateSparseVars(var_name, splited_var_name);
              auto after_run_geo = GetCurrentUS();
              VLOG(1) << "run GEO-SGD var " << splited_var_name << " use time "
                      << after_run_geo - before_run_geo;
            };
            task_futures.emplace_back(
                send_threadpool_->enqueue(std::move(send_task)));
604
          }
C
Chengmo 已提交
605 606 607 608 609 610 611 612 613 614 615 616
        } else {
          auto send_task = [this, &var_name] {
            auto before_run_geo = GetCurrentUS();
            SendUpdateDenseVars(var_name);
            RecvUpdateDenseVars(var_name);
            auto after_run_geo = GetCurrentUS();
            VLOG(3) << "run GEO-SGD var " << var_name << " use time "
                    << after_run_geo - before_run_geo;
          };
          task_futures.emplace_back(
              send_threadpool_->enqueue(std::move(send_task)));
        }
617
      }
C
Chengmo 已提交
618 619 620 621
      for (auto &task_f : task_futures) {
        task_f.wait();
      }
      ids_send_vec_.clear();
622 623 624 625 626
    }
  }
}

std::unordered_set<int64_t> GeoSgdCommunicator::SparseIdsMerge(
C
Chengmo 已提交
627 628
    const std::vector<SparseIdsMap> &ids_send_vec, const std::string &var_name,
    const std::string &splited_var_name) {
629
  // every batch has some sparse id, merge them into one unoredered_set
C
Chengmo 已提交
630 631
  VLOG(3) << "Sparse Ids merge var: " << var_name
          << " splited var: " << splited_var_name;
632
  auto before_run_ids_merge_ = GetCurrentUS();
C
Chengmo 已提交
633 634
  auto origin_var_name = DeltaVarToVar(var_name);
  auto splited_var_index = GetSplitedVarIndex(var_name, splited_var_name);
635
  std::unordered_set<int64_t> ids_set;
C
Chengmo 已提交
636

637
  for (auto ids_map : ids_send_vec) {
C
Chengmo 已提交
638
    for (auto id : ids_map[origin_var_name][splited_var_index]) {
639 640 641 642
      ids_set.insert(id);
    }
  }
  auto after_run_ids_merge_ = GetCurrentUS();
C
Chengmo 已提交
643 644
  VLOG(3) << "run SparseIdsMerge " << splited_var_name << " has nums "
          << ids_set.size() << " use time "
645 646 647 648 649 650 651
          << after_run_ids_merge_ - before_run_ids_merge_;
  return ids_set;
}

void GeoSgdCommunicator::SendUpdateDenseVars(const std::string &var_name) {
  // calc var_delata = (var_training - var_old)/trainer_nums
  // calc var_old += var_delta
C
Chengmo 已提交
652 653
  // var_name: param.delta
  auto origin_var_name = DeltaVarToVar(var_name);
654 655
  auto before_run_send_dense = GetCurrentUS();

C
Chengmo 已提交
656
  auto *var_x = training_scope_->FindVar(origin_var_name);
657 658
  auto var_x_tensor = var_x->Get<framework::LoDTensor>();

C
Chengmo 已提交
659
  auto *var_y = old_scope_->FindVar(origin_var_name);
660 661 662 663 664 665
  auto var_y_tensor = var_y->Get<framework::LoDTensor>();

  auto cpu_ctx = paddle::platform::CPUDeviceContext();
  auto dims = var_x_tensor.dims();

  // create temp var for sub
C
Chengmo 已提交
666
  auto *var_y_sub = old_scope_->Var(var_name);
667 668 669 670
  framework::CopyVariable(*var_y, var_y_sub);
  auto var_y_sub_tensor = var_y_sub->Get<framework::LoDTensor>();

  // create delta var in delta scope
C
Chengmo 已提交
671
  auto *var_z = delta_scope_->Var(var_name);
672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701
  auto *var_z_tensor = var_z->GetMutable<framework::LoDTensor>();
  var_z_tensor->mutable_data<float>(dims, var_x_tensor.place());
  var_z_tensor->set_lod(var_x_tensor.lod());

  math::SetConstant<paddle::platform::CPUDeviceContext, float> constant_functor;
  constant_functor(cpu_ctx, var_z_tensor, static_cast<float>(0));

  // calc sub = var_training - var_old
  auto blas = math::GetBlas<paddle::platform::CPUDeviceContext, float>(cpu_ctx);
  blas.SCAL(var_y_sub_tensor.numel(), -1,
            var_y_sub_tensor.mutable_data<float>(var_y_sub_tensor.place()));
  blas.VADD(var_x_tensor.numel(),
            var_x_tensor.mutable_data<float>(var_x_tensor.place()),
            var_y_sub_tensor.mutable_data<float>(var_y_sub_tensor.place()),
            var_z_tensor->mutable_data<float>(var_z_tensor->place()));

  // calc var_delta = sub / trainer_nums
  float trainer_param = 1.0 / static_cast<float>(trainer_nums_);
  blas.SCAL(var_z_tensor->numel(), trainer_param,
            var_z_tensor->mutable_data<float>(var_z_tensor->place()));

  // calc var_old += var_delta
  blas.VADD(var_y_tensor.numel(),
            var_y_tensor.mutable_data<float>(var_y_tensor.place()),
            var_z_tensor->mutable_data<float>(var_z_tensor->place()),
            var_y_tensor.mutable_data<float>(var_y_tensor.place()));

  auto after_run_send_dense = GetCurrentUS();
  VLOG(3) << "run send update dense var " << var_name << " use time "
          << after_run_send_dense - before_run_send_dense;
C
Chengmo 已提交
702 703 704 705 706 707 708 709 710

  auto send_functor = distributed::ParameterSend<float>();
  auto &ctx = send_varname_to_ctx_.at(var_name);

  auto before_send_dense = GetCurrentUS();
  send_functor(ctx, *delta_scope_.get(), true, 1);
  auto after_send_denxe = GetCurrentUS();
  VLOG(3) << "send " << var_name << " use time "
          << after_send_denxe - before_send_dense;
711 712 713
}

void GeoSgdCommunicator::SendUpdateSparseVars(
C
Chengmo 已提交
714 715
    const std::string &var_name, const std::string &splited_var_name,
    const std::unordered_set<int64_t> &ids_table) {
716 717
  // calc var_delata = (var_training - var_old)/trainer_nums
  // calc var_old += var_delta
C
Chengmo 已提交
718 719
  // var_name: param.delta, splited_var_name: param.block0.delta
  // origin_var_name: param
720 721 722
  auto before_run_send_sparse = GetCurrentUS();

  auto ids_num = ids_table.size();
C
Chengmo 已提交
723 724 725 726
  VLOG(4) << "Sparse Ids nums is : " << ids_num;
  auto origin_var_name = DeltaVarToVar(var_name);

  auto *var_x = training_scope_->FindVar(origin_var_name);
727 728
  auto var_x_tensor = var_x->Get<framework::LoDTensor>();

C
Chengmo 已提交
729
  auto *var_y = old_scope_.get()->FindVar(origin_var_name);
730 731 732 733 734 735 736 737
  auto var_y_tensor = var_y->Get<framework::LoDTensor>();

  auto dims = var_x_tensor.dims();
  auto row_numel = dims[1];

  float *x_value = var_x_tensor.mutable_data<float>(var_x_tensor.place());
  float *y_value = var_y_tensor.mutable_data<float>(var_y_tensor.place());

C
Chengmo 已提交
738
  auto *var_z = delta_scope_->Var(splited_var_name);
739 740 741 742 743 744 745
  auto *var_z_select_rows = var_z->GetMutable<framework::SelectedRows>();
  auto *var_z_value = var_z_select_rows->mutable_value();
  var_z_value->Resize({static_cast<int64_t>(ids_num), row_numel});
  auto *z_value = var_z_value->mutable_data<float>(var_x_tensor.place());

  std::vector<int64_t> new_rows;
  new_rows.insert(new_rows.begin(), ids_table.begin(), ids_table.end());
C
Chengmo 已提交
746 747 748 749

  auto cpu_ctx = paddle::platform::CPUDeviceContext();
  auto blas = math::GetBlas<paddle::platform::CPUDeviceContext, float>(cpu_ctx);
  float avg = 1 / static_cast<float>(trainer_nums_);
750
  for (size_t y = 0; y < new_rows.size(); y++) {
C
Chengmo 已提交
751 752 753 754 755 756 757 758 759 760 761
    auto ids = new_rows[y];

    float *x_val = x_value + ids * row_numel;
    float *y_val = y_value + ids * row_numel;
    float *z_val = z_value + y * row_numel;

    std::vector<float> row_delta(row_numel, 0);
    VSUB<float>(row_numel, x_val, y_val, row_delta.data());
    blas.SCAL(row_numel, avg, row_delta.data());
    blas.VADD(row_numel, row_delta.data(), y_val, y_val);
    blas.VCOPY(row_numel, row_delta.data(), z_val);
762
  }
C
Chengmo 已提交
763

764
  auto after_run_send_sparse = GetCurrentUS();
C
Chengmo 已提交
765
  VLOG(3) << "run send update sparse var " << splited_var_name << " use time "
766
          << after_run_send_sparse - before_run_send_sparse;
C
Chengmo 已提交
767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783

  auto splited_var_index = GetSplitedVarIndex(var_name, splited_var_name);
  std::vector<int64_t> send_rows;
  send_rows.reserve(new_rows.size());
  for (auto idx : new_rows) {
    send_rows.push_back(idx -
                        absolute_section_[origin_var_name][splited_var_index]);
  }
  var_z_select_rows->set_rows(send_rows);
  var_z_select_rows->set_height(
      send_varname_to_ctx_[var_name].height_sections[splited_var_index]);

  auto before_send_sparse = GetCurrentUS();
  RpcSend(var_name, splited_var_name, splited_var_index);
  auto after_send_sparse = GetCurrentUS();
  VLOG(3) << "send " << splited_var_name << " has nums " << new_rows.size()
          << " use time " << after_send_sparse - before_send_sparse;
784 785
}

C
Chengmo 已提交
786
void GeoSgdCommunicator::RecvUpdateDenseVars(const std::string &var_name) {
787 788
  // calc var_training += var_pserver - var_old
  // calc var_old = var_pserver
C
Chengmo 已提交
789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843
  // var_name: param.delta

  // step1: recv dense var from pserver
  auto origin_var_name = DeltaVarToVar(var_name);

  auto before_run_recv = GetCurrentUS();
  auto recv_functor = distributed::ParameterRecv<float>();
  recv_functor(recv_varname_to_ctx_[origin_var_name], *pserver_scope_.get());
  auto after_run_recv = GetCurrentUS();
  VLOG(3) << "recv var " << origin_var_name << " use time "
          << after_run_recv - before_run_recv;

  // step2: update dense var
  auto before_run_update = GetCurrentUS();
  auto *var_x = training_scope_->FindVar(origin_var_name);
  auto var_x_tensor = var_x->Get<framework::LoDTensor>();

  auto *var_y = old_scope_->FindVar(origin_var_name);
  auto var_y_tensor = var_y->Get<framework::LoDTensor>();

  auto *var_y_sub = old_scope_->Var(origin_var_name);
  framework::CopyVariable(*var_y, var_y_sub);
  auto var_y_sub_tensor = var_y_sub->Get<framework::LoDTensor>();

  auto *var_z = pserver_scope_.get()->FindVar(origin_var_name);
  auto var_z_tensor = var_z->Get<framework::LoDTensor>();

  auto cpu_ctx = paddle::platform::CPUDeviceContext();
  auto blas = math::GetBlas<paddle::platform::CPUDeviceContext, float>(cpu_ctx);
  // calc sub = pserver - old
  blas.SCAL(var_y_sub_tensor.numel(), -1,
            var_y_sub_tensor.mutable_data<float>(var_y_sub_tensor.place()));
  blas.VADD(var_y_tensor.numel(),
            var_y_sub_tensor.mutable_data<float>(var_y_sub_tensor.place()),
            var_z_tensor.mutable_data<float>(var_z_tensor.place()),
            var_y_sub_tensor.mutable_data<float>(var_y_sub_tensor.place()));
  // calc recv += sub
  blas.VADD(var_x_tensor.numel(),
            var_x_tensor.mutable_data<float>(var_x_tensor.place()),
            var_y_sub_tensor.mutable_data<float>(var_y_sub_tensor.place()),
            var_x_tensor.mutable_data<float>(var_x_tensor.place()));
  // calc old = pserver
  framework::CopyVariable(*var_z, var_y);
  auto after_run_update = GetCurrentUS();
  VLOG(3) << "dese var update " << origin_var_name << " use time "
          << after_run_update - before_run_update;
}

void GeoSgdCommunicator::RecvUpdateSparseVars(
    const std::string &var_name, const std::string &splited_var_name) {
  // step 1: recv splited var from pserver
  auto splited_var_index = GetSplitedVarIndex(var_name, splited_var_name);
  auto origin_var_name = DeltaVarToVar(var_name);
  auto origin_splited_var_name = DeltaVarToVar(splited_var_name);

844
  auto before_run_recv = GetCurrentUS();
C
Chengmo 已提交
845 846 847 848
  RpcRecv(origin_var_name, origin_splited_var_name, splited_var_index);
  auto after_run_recv = GetCurrentUS();
  VLOG(3) << "recv var " << origin_splited_var_name << " use time "
          << after_run_recv - before_run_recv;
849

C
Chengmo 已提交
850 851 852
  // step 2: update sparse var
  auto before_run_update = GetCurrentUS();
  auto *var_x = training_scope_->FindVar(origin_var_name);
853
  auto var_x_tensor = var_x->Get<framework::LoDTensor>();
C
Chengmo 已提交
854
  auto dims = var_x_tensor.dims();
855 856
  float *x_value = var_x_tensor.mutable_data<float>(var_x_tensor.place());

C
Chengmo 已提交
857
  auto *var_y = old_scope_->FindVar(origin_var_name);
858 859 860
  auto var_y_tensor = var_y->Get<framework::LoDTensor>();
  float *y_value = var_y_tensor.mutable_data<float>(var_y_tensor.place());

C
Chengmo 已提交
861 862 863 864 865 866 867 868 869 870
  auto *var_z = pserver_scope_.get()->FindVar(origin_splited_var_name);
  auto var_z_slr = var_z->GetMutable<framework::SelectedRows>();
  auto row_size = var_z_slr->rows().size();

  std::vector<int64_t> new_rows;
  new_rows.reserve(row_size);

  for (auto ids : var_z_slr->rows()) {
    new_rows.push_back(ids +
                       absolute_section_[origin_var_name][splited_var_index]);
871 872
  }

C
Chengmo 已提交
873 874 875 876 877 878
  auto *new_value = var_z_slr->mutable_value();
  auto row_numel = dims[1];
  auto *z_value = new_value->mutable_data<float>(var_x_tensor.place());

  auto cpu_ctx = paddle::platform::CPUDeviceContext();
  auto blas = math::GetBlas<paddle::platform::CPUDeviceContext, float>(cpu_ctx);
879
  for (size_t y = 0; y < new_rows.size(); y++) {
C
Chengmo 已提交
880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896
    std::vector<float> row_delta(row_numel, 0);

    auto ids = new_rows[y];

    float *x_val = x_value + ids * row_numel;
    float *y_val = y_value + ids * row_numel;
    float *z_val = z_value + y * row_numel;

    VSUB(row_numel, z_val, y_val, row_delta.data());
    blas.VADD(row_numel, row_delta.data(), x_val, x_val);
    blas.VCOPY(row_numel, z_val, y_val);
  }

  auto after_run_update = GetCurrentUS();
  VLOG(3) << "sparse var recv update " << origin_splited_var_name << " has num "
          << new_rows.size() << " use time "
          << after_run_update - before_run_update;
897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928
}

void GeoSgdCommunicator::GeoSgdSparseParamInit(framework::Scope *scope_x,
                                               framework::Scope *scope_y,
                                               const std::string var_name) {
  // create selectedrows var from lodtensor var info
  auto *var_x = scope_x->Var(var_name);
  auto *var_y = scope_y->Var(var_name);

  auto var_x_tensor = var_x->Get<framework::LoDTensor>();
  auto *var_y_select_rows = var_y->GetMutable<framework::SelectedRows>();

  auto dims = var_x_tensor.dims();
  auto rows = dims[0];
  auto row_numel = dims[1];

  var_y_select_rows->set_height(rows);
  std::vector<int64_t> new_rows{};
  var_y_select_rows->set_rows(new_rows);
  auto *var_y_value = var_y_select_rows->mutable_value();
  var_y_value->Resize({rows, row_numel});
  var_y_value->mutable_data<float>(var_x_tensor.place());
}

void GeoSgdCommunicator::GeoSgdDenseParamInit(framework::Scope *scope_x,
                                              framework::Scope *scope_y,
                                              const std::string var_name) {
  auto *var_x = scope_x->Var(var_name);
  auto *var_y = scope_y->Var(var_name);
  framework::CopyVariable(*var_x, var_y);
}

C
Chengmo 已提交
929 930 931 932 933 934 935 936 937 938 939
void GeoSgdCommunicator::RpcSend(const std::string &origin_var_name,
                                 const std::string &splited_var_name,
                                 const size_t &splited_var_index) {
  auto trainer_id = send_varname_to_ctx_[origin_var_name].trainer_id;
  auto endpoint =
      send_varname_to_ctx_[origin_var_name].epmap[splited_var_index];

  platform::DeviceContextPool &pool = platform::DeviceContextPool::Instance();
  auto &cpu_ctx_send = *pool.Get(platform::CPUPlace());
  distributed::RPCClient *rpc_client =
      distributed::RPCClient::GetInstance<RPCCLIENT_T>(trainer_id);
940 941 942
  auto handle = rpc_client->AsyncSendVar(endpoint, cpu_ctx_send,
                                         *delta_scope_.get(), splited_var_name);
  handle->Wait();
C
Chengmo 已提交
943 944 945 946 947 948 949 950 951 952 953 954
}

void GeoSgdCommunicator::RpcRecv(const std::string &var_name,
                                 const std::string &splited_var_name,
                                 const size_t &splited_var_index) {
  auto train_id = recv_varname_to_ctx_[var_name].trainer_id;
  auto endpoint = recv_varname_to_ctx_[var_name].epmap[splited_var_index];
  platform::DeviceContextPool &pool = platform::DeviceContextPool::Instance();
  auto &cpu_ctx_recv = *pool.Get(platform::CPUPlace());
  distributed::RPCClient *rpc_client =
      distributed::RPCClient::GetInstance<RPCCLIENT_T>(train_id);
  pserver_scope_->Var(splited_var_name);
955 956 957 958
  auto handle = rpc_client->AsyncGetVar(endpoint, cpu_ctx_recv,
                                        *pserver_scope_.get(), splited_var_name,
                                        splited_var_name, splited_var_name);
  handle->Wait();
C
Chengmo 已提交
959 960 961 962
}

void GeoSgdCommunicator::Recv() {}

963 964 965 966 967 968 969
void GeoSgdCommunicator::InitImpl(const RpcCtxMap &send_varname_to_ctx,
                                  const RpcCtxMap &recv_varname_to_ctx,
                                  Scope *recv_scope) {}

void GeoSgdCommunicator::InitImpl(const paddle::framework::ProgramDesc &program,
                                  Scope *recv_scope) {}

Q
Qiao Longfei 已提交
970 971 972
}  // namespace distributed
}  // namespace operators
}  // namespace paddle