brpc_ps_server.cc 21.0 KB
Newer Older
T
tangwei12 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
// 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.

#include "paddle/fluid/distributed/service/brpc_ps_server.h"
16
#include <netdb.h>
T
tangwei12 已提交
17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33
#include <thread>  // NOLINT
#include "Eigen/Dense"
#include "butil/endpoint.h"
#include "iomanip"
#include "paddle/fluid/distributed/table/table.h"
#include "paddle/fluid/framework/archive.h"
#include "paddle/fluid/platform/profiler.h"

namespace paddle {
namespace distributed {

int32_t BrpcPsServer::initialize() {
  auto &service_config = _config.downpour_server_param().service_param();
  if (!service_config.has_service_class()) {
    LOG(ERROR) << "miss service_class in ServerServiceParameter";
    return -1;
  }
T
tangwei12 已提交
34 35
  auto *service =
      CREATE_PSCORE_CLASS(PsBaseService, service_config.service_class());
T
tangwei12 已提交
36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61
  if (service == NULL) {
    LOG(ERROR) << "service is unregistered, service_name:"
               << service_config.service_class();
    return -1;
  }

  _service.reset(service);
  if (service->configure(this) != 0 || service->initialize() != 0) {
    LOG(ERROR) << "service initialize failed, service_name:"
               << service_config.service_class();
    return -1;
  }
  if (_server.AddService(service, brpc::SERVER_DOESNT_OWN_SERVICE) != 0) {
    LOG(ERROR) << "service add to brpc failed, service:"
               << service_config.service_class();
    return -1;
  }
  return 0;
}

uint64_t BrpcPsServer::start(const std::string &ip, uint32_t port) {
  std::unique_lock<std::mutex> lock(mutex_);

  std::string ip_port = ip + ":" + std::to_string(port);
  VLOG(3) << "server of rank " << _rank << " starts at " << ip_port;
  brpc::ServerOptions options;
T
tangwei12 已提交
62 63 64 65

  int num_threads = std::thread::hardware_concurrency();
  auto trainers = _environment->get_trainers();
  options.num_threads = trainers > num_threads ? trainers : num_threads;
T
tangwei12 已提交
66 67

  if (_server.Start(ip_port.c_str(), &options) != 0) {
68 69 70 71 72 73 74 75 76
    VLOG(0) << "BrpcPsServer start failed, ip_port= " << ip_port
            << " , Try Again.";

    std::string int_ip_port = GetIntTypeEndpoint(ip, port);

    if (_server.Start(int_ip_port.c_str(), &options) != 0) {
      LOG(ERROR) << "BrpcPsServer start failed, ip_port= " << int_ip_port;
      return 0;
    }
T
tangwei12 已提交
77
  }
78

T
tangwei12 已提交
79 80 81 82 83 84 85 86 87 88 89 90 91 92 93
  VLOG(0) << "BrpcPsServer::start registe_ps_server";
  _environment->registe_ps_server(ip, port, _rank);
  VLOG(0) << "BrpcPsServer::start wait";
  cv_.wait(lock, [&] { return stoped_; });

  PSHost host;
  host.ip = ip;
  host.port = port;
  host.rank = _rank;
  VLOG(0) << "BrpcPsServer::start return host.rank";
  return host.rank;
}

int32_t BrpcPsServer::port() { return _server.listen_address().port; }

T
tangwei12 已提交
94
int32_t BrpcPsService::initialize() {
T
tangwei12 已提交
95
  _is_initialize_shard_info = false;
T
tangwei12 已提交
96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115
  _service_handler_map[PS_STOP_SERVER] = &BrpcPsService::stop_server;
  _service_handler_map[PS_PULL_DENSE_TABLE] = &BrpcPsService::pull_dense;
  _service_handler_map[PS_PUSH_DENSE_TABLE] = &BrpcPsService::push_dense;
  _service_handler_map[PS_PULL_SPARSE_TABLE] = &BrpcPsService::pull_sparse;
  _service_handler_map[PS_PUSH_SPARSE_TABLE] = &BrpcPsService::push_sparse;
  _service_handler_map[PS_SAVE_ONE_TABLE] = &BrpcPsService::save_one_table;
  _service_handler_map[PS_SAVE_ALL_TABLE] = &BrpcPsService::save_all_table;
  _service_handler_map[PS_SHRINK_TABLE] = &BrpcPsService::shrink_table;
  _service_handler_map[PS_LOAD_ONE_TABLE] = &BrpcPsService::load_one_table;
  _service_handler_map[PS_LOAD_ALL_TABLE] = &BrpcPsService::load_all_table;
  _service_handler_map[PS_CLEAR_ONE_TABLE] = &BrpcPsService::clear_one_table;
  _service_handler_map[PS_CLEAR_ALL_TABLE] = &BrpcPsService::clear_all_table;
  _service_handler_map[PS_PUSH_DENSE_PARAM] = &BrpcPsService::push_dense_param;
  _service_handler_map[PS_PRINT_TABLE_STAT] = &BrpcPsService::print_table_stat;
  _service_handler_map[PS_PULL_GEO_PARAM] = &BrpcPsService::pull_geo_param;
  _service_handler_map[PS_PUSH_SPARSE_PARAM] =
      &BrpcPsService::push_sparse_param;
  _service_handler_map[PS_BARRIER] = &BrpcPsService::barrier;
  _service_handler_map[PS_START_PROFILER] = &BrpcPsService::start_profiler;
  _service_handler_map[PS_STOP_PROFILER] = &BrpcPsService::stop_profiler;
116
  _service_handler_map[PS_PUSH_GLOBAL_STEP] = &BrpcPsService::push_global_step;
T
tangwei12 已提交
117 118 119 120 121 122 123 124 125 126 127 128 129 130 131

  // shard初始化,server启动后才可从env获取到server_list的shard信息
  initialize_shard_info();

  return 0;
}

#define CHECK_TABLE_EXIST(table, request, response)        \
  if (table == NULL) {                                     \
    std::string err_msg("table not found with table_id:"); \
    err_msg.append(std::to_string(request.table_id()));    \
    set_response_code(response, -1, err_msg.c_str());      \
    return -1;                                             \
  }

T
tangwei12 已提交
132
int32_t BrpcPsService::initialize_shard_info() {
T
tangwei12 已提交
133 134 135 136 137 138 139 140 141 142 143 144 145 146 147
  if (!_is_initialize_shard_info) {
    std::lock_guard<std::mutex> guard(_initialize_shard_mutex);
    if (_is_initialize_shard_info) {
      return 0;
    }
    size_t shard_num = _server->environment()->get_ps_servers().size();
    auto &table_map = *(_server->table());
    for (auto itr : table_map) {
      itr.second->set_shard(_rank, shard_num);
    }
    _is_initialize_shard_info = true;
  }
  return 0;
}

T
tangwei12 已提交
148 149 150 151
void BrpcPsService::service(google::protobuf::RpcController *cntl_base,
                            const PsRequestMessage *request,
                            PsResponseMessage *response,
                            google::protobuf::Closure *done) {
T
tangwei12 已提交
152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178
  brpc::ClosureGuard done_guard(done);
  std::string log_label("ReceiveCmd-");
  if (!request->has_table_id()) {
    set_response_code(*response, -1, "PsRequestMessage.tabel_id is required");
    return;
  }

  response->set_err_code(0);
  response->set_err_msg("");
  auto *table = _server->table(request->table_id());
  brpc::Controller *cntl = static_cast<brpc::Controller *>(cntl_base);
  auto itr = _service_handler_map.find(request->cmd_id());
  if (itr == _service_handler_map.end()) {
    std::string err_msg(
        "undefined cmd_id, should match PsCmdID in ps.proto, cmd_id:");
    err_msg.append(std::to_string(request->cmd_id()));
    set_response_code(*response, -1, err_msg.c_str());
    return;
  }
  serviceHandlerFunc handler_func = itr->second;
  int service_ret = (this->*handler_func)(table, *request, *response, cntl);
  if (service_ret != 0) {
    response->set_err_code(service_ret);
    response->set_err_msg("server internal error");
  }
}

T
tangwei12 已提交
179 180 181
int32_t BrpcPsService::pull_dense(Table *table, const PsRequestMessage &request,
                                  PsResponseMessage &response,
                                  brpc::Controller *cntl) {
T
tangwei12 已提交
182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206
  platform::RecordEvent record_event("PsService->pull_dense");
  CHECK_TABLE_EXIST(table, request, response)
  if (request.params_size() < 1) {
    set_response_code(
        response, -1,
        "PsRequestMessage.datas is requeired at least 1 for num of dense");
    return 0;
  }
  uint32_t num = *(const uint32_t *)request.params(0).c_str();
  if (num < 0) {
    set_response_code(response, -1,
                      "PsRequestMessage.datas[0] is invalid, num must >= 0");
    return 0;
  }

  std::vector<float> res_data;
  res_data.resize(num * table->value_accesor()->select_size() / sizeof(float));
  table->pull_dense(res_data.data(), num);

  cntl->response_attachment().append((char *)res_data.data(),
                                     res_data.size() * sizeof(float));

  return 0;
}

T
tangwei12 已提交
207 208 209 210
int32_t BrpcPsService::push_dense_param(Table *table,
                                        const PsRequestMessage &request,
                                        PsResponseMessage &response,
                                        brpc::Controller *cntl) {
T
tangwei12 已提交
211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233
  platform::RecordEvent record_event("PsService->push_dense_param");
  CHECK_TABLE_EXIST(table, request, response)
  thread_local std::string push_buffer;
  auto &req_io_buffer = cntl->request_attachment();
  auto req_buffer_size = req_io_buffer.size();
  if (req_buffer_size < 1) {
    set_response_code(response, -1, "req attachment is empty");
    return 0;
  }
  push_buffer.resize(0);
  push_buffer.reserve(req_buffer_size);
  const char *data = (const char *)cntl->request_attachment().fetch(
      const_cast<char *>(push_buffer.data()), req_buffer_size);

  uint32_t num = *(const uint32_t *)data;

  const float *values = (const float *)(data + sizeof(uint32_t));
  if (table->push_dense_param(values, num) != 0) {
    set_response_code(response, -1, "push_dense_param failed");
  }
  return 0;
}

T
tangwei12 已提交
234 235 236
int32_t BrpcPsService::push_dense(Table *table, const PsRequestMessage &request,
                                  PsResponseMessage &response,
                                  brpc::Controller *cntl) {
T
tangwei12 已提交
237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259
  platform::RecordEvent record_event("PsService->push_dense");
  CHECK_TABLE_EXIST(table, request, response)
  auto req_buffer_size = request.data().size();
  if (req_buffer_size < 1) {
    // set_response_code(response, 0, "push dense data is empty");
    return 0;
  }

  /*
  Push Content:
  |--num--|---valuesData---|
  |--4B---|----------------|
  */
  uint32_t num = *(const uint32_t *)(request.data().data());
  const float *values =
      (const float *)(request.data().data() + sizeof(uint32_t));
  if (table->push_dense(values, num) != 0) {
    set_response_code(response, -1, "push_dense failed");
  }

  return 0;
}

T
tangwei12 已提交
260 261 262
int32_t BrpcPsService::barrier(Table *table, const PsRequestMessage &request,
                               PsResponseMessage &response,
                               brpc::Controller *cntl) {
T
tangwei12 已提交
263 264 265 266 267 268 269 270 271 272 273 274 275 276 277
  CHECK_TABLE_EXIST(table, request, response)

  if (request.params_size() < 1) {
    set_response_code(response, -1,
                      "PsRequestMessage.params is requeired at "
                      "least 1 for num of sparse_key");
    return 0;
  }

  auto trainer_id = request.client_id();
  auto barrier_type = request.params(0);
  table->barrier(trainer_id, barrier_type);
  return 0;
}

T
tangwei12 已提交
278 279 280 281
int32_t BrpcPsService::push_sparse_param(Table *table,
                                         const PsRequestMessage &request,
                                         PsResponseMessage &response,
                                         brpc::Controller *cntl) {
T
tangwei12 已提交
282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309
  platform::RecordEvent record_event("PsService->push_sparse_param");
  CHECK_TABLE_EXIST(table, request, response)
  auto &push_data = request.data();
  if (push_data.size() < 1) {
    // set_response_code(response, 0, "push sparse data is empty");
    return 0;
  }
  if (request.params_size() < 1) {
    set_response_code(response, -1,
                      "PsRequestMessage.params is requeired at "
                      "least 1 for num of sparse_key");
    return 0;
  }
  uint32_t num = *(uint32_t *)(request.params(0).c_str());
  /*
  Push Content:
  |---keysData---|---valuesData---|
  |---8*{num}B---|----------------|
  */
  const uint64_t *keys = (const uint64_t *)push_data.data();
  const float *values =
      (const float *)(push_data.data() + sizeof(uint64_t) * num);
  if (table->push_sparse_param(keys, values, num) != 0) {
    set_response_code(response, -1, "push_sparse_param error");
  }
  return 0;
}

T
tangwei12 已提交
310 311 312 313
int32_t BrpcPsService::pull_geo_param(Table *table,
                                      const PsRequestMessage &request,
                                      PsResponseMessage &response,
                                      brpc::Controller *cntl) {
T
tangwei12 已提交
314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332
  platform::RecordEvent record_event("PsService->pull_geo_param");
  CHECK_TABLE_EXIST(table, request, response)
  thread_local std::string push_sparse_request_buffer;

  auto trainer_id = request.client_id();

  std::vector<float> values;
  std::vector<uint64_t> ids;
  table->pull_geo_param(trainer_id, &values, &ids);

  uint32_t num = ids.size();
  cntl->response_attachment().append((char *)(&num), sizeof(uint32_t));
  cntl->response_attachment().append((char *)ids.data(),
                                     ids.size() * sizeof(uint64_t));
  cntl->response_attachment().append((char *)values.data(),
                                     values.size() * sizeof(float));
  return 0;
}

T
tangwei12 已提交
333 334 335 336
int32_t BrpcPsService::pull_sparse(Table *table,
                                   const PsRequestMessage &request,
                                   PsResponseMessage &response,
                                   brpc::Controller *cntl) {
T
tangwei12 已提交
337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370
  platform::RecordEvent record_event("PsService->pull_sparse");
  CHECK_TABLE_EXIST(table, request, response)
  thread_local std::string push_sparse_request_buffer;
  auto &req_io_buffer = cntl->request_attachment();
  auto req_buffer_size = req_io_buffer.size();
  if (req_buffer_size < 1) {
    set_response_code(response, -1, "req attachment is empty");
    return 0;
  }
  if (request.params_size() < 1) {
    set_response_code(response, -1,
                      "PsRequestMessage.params is requeired at "
                      "least 1 for num of sparse_key");
    return 0;
  }
  uint32_t num = *(uint32_t *)(request.params(0).c_str());
  push_sparse_request_buffer.resize(0);
  push_sparse_request_buffer.reserve(req_buffer_size);
  const char *data = (const char *)cntl->request_attachment().fetch(
      const_cast<char *>(push_sparse_request_buffer.data()), req_buffer_size);
  /*
  Attachment Content:
  |---keysData---|
  |---8*{num}B---|
  */
  const uint64_t *keys = (const uint64_t *)data;
  std::vector<float> res_data;
  res_data.resize(num * table->value_accesor()->select_size() / sizeof(float));
  table->pull_sparse(res_data.data(), keys, num);
  cntl->response_attachment().append((char *)res_data.data(),
                                     res_data.size() * sizeof(float));
  return 0;
}

T
tangwei12 已提交
371 372 373 374
int32_t BrpcPsService::push_sparse(Table *table,
                                   const PsRequestMessage &request,
                                   PsResponseMessage &response,
                                   brpc::Controller *cntl) {
T
tangwei12 已提交
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
  platform::RecordEvent record_event("PsService->push_sparse");
  CHECK_TABLE_EXIST(table, request, response)
  auto &push_data = request.data();
  if (push_data.size() < 1) {
    // set_response_code(response, 0, "push sparse data is empty");
    return 0;
  }
  if (request.params_size() < 1) {
    set_response_code(response, -1,
                      "PsRequestMessage.params is requeired at "
                      "least 1 for num of sparse_key");
    return 0;
  }
  uint32_t num = *(uint32_t *)(request.params(0).c_str());
  /*
  Push Content:
  |---keysData---|---valuesData---|
  |---8*{num}B---|----------------|
  */
  const uint64_t *keys = (const uint64_t *)push_data.data();
  const float *values =
      (const float *)(push_data.data() + sizeof(uint64_t) * num);
  if (table->push_sparse(keys, values, num) != 0) {
    set_response_code(response, -1, "push_sparse error");
  }
  return 0;
}

T
tangwei12 已提交
403 404 405 406
int32_t BrpcPsService::print_table_stat(Table *table,
                                        const PsRequestMessage &request,
                                        PsResponseMessage &response,
                                        brpc::Controller *cntl) {
T
tangwei12 已提交
407 408 409 410 411 412 413 414 415 416
  CHECK_TABLE_EXIST(table, request, response)
  std::pair<int64_t, int64_t> ret = table->print_table_stat();
  paddle::framework::BinaryArchive ar;
  ar << ret.first << ret.second;
  std::string table_info(ar.Buffer(), ar.Length());
  response.set_data(table_info);

  return 0;
}

T
tangwei12 已提交
417 418 419 420
int32_t BrpcPsService::load_one_table(Table *table,
                                      const PsRequestMessage &request,
                                      PsResponseMessage &response,
                                      brpc::Controller *cntl) {
T
tangwei12 已提交
421 422 423 424 425 426 427 428 429 430 431 432 433 434
  CHECK_TABLE_EXIST(table, request, response)
  if (request.params_size() < 2) {
    set_response_code(
        response, -1,
        "PsRequestMessage.datas is requeired at least 2 for path & load_param");
    return -1;
  }
  if (table->load(request.params(0), request.params(1)) != 0) {
    set_response_code(response, -1, "table load failed");
    return -1;
  }
  return 0;
}

T
tangwei12 已提交
435 436 437 438
int32_t BrpcPsService::load_all_table(Table *table,
                                      const PsRequestMessage &request,
                                      PsResponseMessage &response,
                                      brpc::Controller *cntl) {
T
tangwei12 已提交
439 440 441 442 443 444 445 446 447 448
  auto &table_map = *(_server->table());
  for (auto &itr : table_map) {
    if (load_one_table(itr.second.get(), request, response, cntl) != 0) {
      LOG(ERROR) << "load table[" << itr.first << "] failed";
      return -1;
    }
  }
  return 0;
}

T
tangwei12 已提交
449 450 451 452
int32_t BrpcPsService::save_one_table(Table *table,
                                      const PsRequestMessage &request,
                                      PsResponseMessage &response,
                                      brpc::Controller *cntl) {
T
tangwei12 已提交
453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470
  CHECK_TABLE_EXIST(table, request, response)
  if (request.params_size() < 2) {
    set_response_code(
        response, -1,
        "PsRequestMessage.datas is requeired at least 2, path&mode");
    return -1;
  }
  table->flush();

  int32_t feasign_size = 0;
  feasign_size = table->save(request.params(0), request.params(1));
  if (feasign_size < 0) {
    set_response_code(response, -1, "table save failed");
    return -1;
  }
  return feasign_size;
}

T
tangwei12 已提交
471 472 473 474
int32_t BrpcPsService::save_all_table(Table *table,
                                      const PsRequestMessage &request,
                                      PsResponseMessage &response,
                                      brpc::Controller *cntl) {
T
tangwei12 已提交
475 476 477 478 479 480 481 482 483 484 485 486 487 488
  auto &table_map = *(_server->table());
  int32_t all_feasign_size = 0;
  int32_t feasign_size = 0;

  for (auto &itr : table_map) {
    feasign_size = save_one_table(itr.second.get(), request, response, cntl);
    if (feasign_size < 0) {
      LOG(ERROR) << "save table[" << itr.first << "] failed";
      return -1;
    }
  }
  return 0;
}

T
tangwei12 已提交
489 490 491 492
int32_t BrpcPsService::shrink_table(Table *table,
                                    const PsRequestMessage &request,
                                    PsResponseMessage &response,
                                    brpc::Controller *cntl) {
T
tangwei12 已提交
493 494 495 496 497 498 499 500
  CHECK_TABLE_EXIST(table, request, response)
  table->flush();
  if (table->shrink() != 0) {
    set_response_code(response, -1, "table shrink failed");
  }
  return 0;
}

T
tangwei12 已提交
501 502 503 504
int32_t BrpcPsService::clear_one_table(Table *table,
                                       const PsRequestMessage &request,
                                       PsResponseMessage &response,
                                       brpc::Controller *cntl) {
T
tangwei12 已提交
505 506 507 508 509 510
  CHECK_TABLE_EXIST(table, request, response)
  table->flush();
  table->clear();
  return 0;
}

T
tangwei12 已提交
511 512 513 514
int32_t BrpcPsService::clear_all_table(Table *table,
                                       const PsRequestMessage &request,
                                       PsResponseMessage &response,
                                       brpc::Controller *cntl) {
T
tangwei12 已提交
515 516 517 518 519 520 521 522 523
  auto &table_map = *(_server->table());
  for (auto &itr : table_map) {
    if (clear_one_table(itr.second.get(), request, response, cntl) != 0) {
      return -1;
    }
  }
  return 0;
}

T
tangwei12 已提交
524 525 526 527
int32_t BrpcPsService::stop_server(Table *table,
                                   const PsRequestMessage &request,
                                   PsResponseMessage &response,
                                   brpc::Controller *cntl) {
T
tangwei12 已提交
528 529 530 531 532 533 534 535 536
  auto *p_server = _server;
  std::thread t_stop([p_server]() {
    p_server->stop();
    LOG(INFO) << "Server Stoped";
  });
  t_stop.detach();
  return 0;
}

T
tangwei12 已提交
537 538 539 540
int32_t BrpcPsService::stop_profiler(Table *table,
                                     const PsRequestMessage &request,
                                     PsResponseMessage &response,
                                     brpc::Controller *cntl) {
T
tangwei12 已提交
541 542 543 544 545
  platform::DisableProfiler(platform::EventSortingKey::kDefault,
                            string::Sprintf("server_%s_profile", _rank));
  return 0;
}

T
tangwei12 已提交
546 547 548 549
int32_t BrpcPsService::start_profiler(Table *table,
                                      const PsRequestMessage &request,
                                      PsResponseMessage &response,
                                      brpc::Controller *cntl) {
T
tangwei12 已提交
550 551 552 553
  platform::EnableProfiler(platform::ProfilerState::kCPU);
  return 0;
}

T
tangwei12 已提交
554 555 556 557
int32_t BrpcPsService::push_global_step(Table *table,
                                        const PsRequestMessage &request,
                                        PsResponseMessage &response,
                                        brpc::Controller *cntl) {
558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574
  CHECK_TABLE_EXIST(table, request, response);
  auto req_buffer_size = request.data().size();
  if (req_buffer_size < 1) {
    set_response_code(response, 0, "run_program data is empty");
    return 0;
  }
  uint32_t num = *(const uint32_t *)(request.data().data());
  const int64_t *values =
      (const int64_t *)(request.data().data() + sizeof(uint32_t));
  auto trainer_id = request.client_id();
  if (table->push_dense(values, trainer_id) != 0) {
    set_response_code(response, -1, "run_program failed");
  }

  return 0;
}

T
tangwei12 已提交
575 576
}  // namespace distributed
}  // namespace paddle