brpc_service_dense_sgd_test.cc 10.1 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 <unistd.h>
16

T
tangwei12 已提交
17 18 19 20
#include <string>
#include <thread>  // NOLINT

#include "gtest/gtest.h"
21 22
#include "paddle/fluid/distributed/ps/service/brpc_ps_client.h"
#include "paddle/fluid/distributed/ps/service/brpc_ps_server.h"
23
#include "paddle/fluid/framework/program_desc.h"
T
tangwei12 已提交
24 25
#include "paddle/fluid/framework/scope.h"
#include "paddle/fluid/platform/place.h"
26
#include "paddle/phi/kernels/funcs/math_function.h"
T
tangwei12 已提交
27

28 29 30 31 32 33 34 35 36 37
namespace paddle {
namespace distributed {
class DownpourBrpcClosure;
class PSClient;
class PSServer;
}  // namespace distributed
namespace framework {
class Variable;
}  // namespace framework
}  // namespace paddle
T
tangwei12 已提交
38

39
namespace phi {
40
class DenseTensor;
41
}  // namespace phi
42

T
tangwei12 已提交
43 44 45 46 47 48 49 50 51 52 53
namespace framework = paddle::framework;
namespace platform = paddle::platform;
namespace operators = paddle::operators;
namespace memory = paddle::memory;
namespace distributed = paddle::distributed;

void CreateVarsOnScope(framework::Scope* scope, platform::CPUPlace* place) {
  auto x_var = scope->Var("x");
  x_var->GetMutable<framework::LoDTensor>();
}

54 55
void InitTensorsOnClient(framework::Scope* scope,
                         platform::CPUPlace* place,
T
tangwei12 已提交
56 57 58 59 60 61
                         int64_t rows_numel) {
  CreateVarsOnScope(scope, place);

  auto x_var = scope->Var("x")->GetMutable<framework::LoDTensor>();
  float* x_ptr =
      x_var->mutable_data<float>(framework::DDim({1, rows_numel}), *place);
62 63
  for (int64_t i = 0; i < rows_numel; ++i)
    x_ptr[i] = 1.0 * static_cast<float>(i);
T
tangwei12 已提交
64 65 66 67 68
}

void GetDownpourDenseTableProto(
    ::paddle::distributed::TableParameter* dense_table_proto) {
  dense_table_proto->set_table_id(0);
69
  dense_table_proto->set_table_class("MemoryDenseTable");
T
tangwei12 已提交
70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101
  dense_table_proto->set_shard_num(256);
  dense_table_proto->set_type(::paddle::distributed::PS_DENSE_TABLE);
  ::paddle::distributed::TableAccessorParameter* accessor_proto =
      dense_table_proto->mutable_accessor();
  ::paddle::distributed::CommonAccessorParameter* common_proto =
      dense_table_proto->mutable_common();

  accessor_proto->set_accessor_class("CommMergeAccessor");
  accessor_proto->set_fea_dim(100);
  accessor_proto->set_embedx_dim(1);

  common_proto->set_name("sgd");
  common_proto->set_table_name("MergedDense");
  common_proto->set_trainer_num(1);
  common_proto->set_sync(false);
  common_proto->add_params("Param");
  common_proto->add_dims(100);
  common_proto->add_initializers("fill_constant&1.0");
  common_proto->add_params("LearningRate");
  common_proto->add_dims(1);
  common_proto->add_initializers("fill_constant&1.0");
}

::paddle::distributed::PSParameter GetServerProto() {
  // Generate server proto desc
  ::paddle::distributed::PSParameter server_fleet_desc;
  ::paddle::distributed::ServerParameter* server_proto =
      server_fleet_desc.mutable_server_param();
  ::paddle::distributed::DownpourServerParameter* downpour_server_proto =
      server_proto->mutable_downpour_server_param();
  ::paddle::distributed::ServerServiceParameter* server_service_proto =
      downpour_server_proto->mutable_service_param();
T
tangwei12 已提交
102
  server_service_proto->set_service_class("BrpcPsService");
T
tangwei12 已提交
103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131
  server_service_proto->set_server_class("BrpcPsServer");
  server_service_proto->set_client_class("BrpcPsClient");
  server_service_proto->set_start_server_port(0);
  server_service_proto->set_server_thread_num(12);

  ::paddle::distributed::TableParameter* dense_table_proto =
      downpour_server_proto->add_downpour_table_param();
  GetDownpourDenseTableProto(dense_table_proto);
  return server_fleet_desc;
}

::paddle::distributed::PSParameter GetWorkerProto() {
  ::paddle::distributed::PSParameter worker_fleet_desc;
  ::paddle::distributed::WorkerParameter* worker_proto =
      worker_fleet_desc.mutable_worker_param();

  ::paddle::distributed::DownpourWorkerParameter* downpour_worker_proto =
      worker_proto->mutable_downpour_worker_param();

  ::paddle::distributed::TableParameter* worker_dense_table_proto =
      downpour_worker_proto->add_downpour_table_param();
  GetDownpourDenseTableProto(worker_dense_table_proto);

  ::paddle::distributed::ServerParameter* server_proto =
      worker_fleet_desc.mutable_server_param();
  ::paddle::distributed::DownpourServerParameter* downpour_server_proto =
      server_proto->mutable_downpour_server_param();
  ::paddle::distributed::ServerServiceParameter* server_service_proto =
      downpour_server_proto->mutable_service_param();
T
tangwei12 已提交
132
  server_service_proto->set_service_class("BrpcPsService");
T
tangwei12 已提交
133 134 135 136 137 138 139 140 141 142 143 144 145 146
  server_service_proto->set_server_class("BrpcPsServer");
  server_service_proto->set_client_class("BrpcPsClient");
  server_service_proto->set_start_server_port(0);
  server_service_proto->set_server_thread_num(12);

  ::paddle::distributed::TableParameter* server_dense_table_proto =
      downpour_server_proto->add_downpour_table_param();
  GetDownpourDenseTableProto(server_dense_table_proto);

  return worker_fleet_desc;
}

/*-------------------------------------------------------------------------*/

147
const char* ip_ = "127.0.0.1";
T
tangwei12 已提交
148 149 150 151 152 153 154 155 156 157 158 159 160
uint32_t port_ = 4214;

std::vector<std::string> host_sign_list_;

std::shared_ptr<paddle::distributed::PSServer> pserver_ptr_;

std::shared_ptr<paddle::distributed::PSClient> worker_ptr_;

void RunServer() {
  ::paddle::distributed::PSParameter server_proto = GetServerProto();

  auto _ps_env = paddle::distributed::PaddlePSEnvironment();
  LOG(INFO) << "RUN set_ps_servers";
Z
zhaocaibei123 已提交
161
  _ps_env.SetPsServers(&host_sign_list_, 1);
T
tangwei12 已提交
162
  pserver_ptr_ = std::shared_ptr<paddle::distributed::PSServer>(
Z
zhaocaibei123 已提交
163
      paddle::distributed::PSServerFactory::Create(server_proto));
T
tangwei12 已提交
164
  LOG(INFO) << "RUN configure";
165 166 167
  std::vector<framework::ProgramDesc> empty_vec;
  framework::ProgramDesc empty_prog;
  empty_vec.push_back(empty_prog);
Z
zhaocaibei123 已提交
168
  pserver_ptr_->Configure(server_proto, _ps_env, 0, empty_vec);
T
tangwei12 已提交
169
  LOG(INFO) << "RUN start";
Z
zhaocaibei123 已提交
170
  pserver_ptr_->Start(ip_, port_);
T
tangwei12 已提交
171 172 173 174 175 176 177 178 179 180
  LOG(INFO) << "End start";
}

void RunClient(std::map<uint64_t, std::vector<paddle::distributed::Region>>&
                   dense_regions) {
  ::paddle::distributed::PSParameter worker_proto = GetWorkerProto();
  paddle::distributed::PaddlePSEnvironment _ps_env;
  auto servers_ = host_sign_list_.size();
  _ps_env = paddle::distributed::PaddlePSEnvironment();
  LOG(INFO) << "Run set_ps_servers";
Z
zhaocaibei123 已提交
181
  _ps_env.SetPsServers(&host_sign_list_, servers_);
T
tangwei12 已提交
182 183
  LOG(INFO) << "Run Create PSClient";
  worker_ptr_ = std::shared_ptr<paddle::distributed::PSClient>(
Z
zhaocaibei123 已提交
184
      paddle::distributed::PSClientFactory::Create(worker_proto));
T
tangwei12 已提交
185
  LOG(INFO) << "Run configure";
Z
zhaocaibei123 已提交
186
  worker_ptr_->Configure(worker_proto, dense_regions, _ps_env, 0);
T
tangwei12 已提交
187 188 189 190 191 192
}

void RunBrpcPushDense() {
  setenv("http_proxy", "", 1);
  setenv("https_proxy", "", 1);
  auto ph_host = paddle::distributed::PSHost(ip_, port_, 0);
Z
zhaocaibei123 已提交
193
  host_sign_list_.push_back(ph_host.SerializeToString());
T
tangwei12 已提交
194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223

  // Srart Server
  std::thread server_thread(RunServer);
  sleep(1);

  // Start Client
  LOG(INFO) << "Run InitTensorsOnClient";
  framework::Scope client_scope;
  platform::CPUPlace place;
  InitTensorsOnClient(&client_scope, &place, 100);
  std::map<uint64_t, std::vector<paddle::distributed::Region>> dense_regions;
  dense_regions.insert(
      std::pair<uint64_t, std::vector<paddle::distributed::Region>>(0, {}));
  auto regions = dense_regions[0];
  framework::Variable* var = client_scope.FindVar("x");
  framework::LoDTensor* tensor = var->GetMutable<framework::LoDTensor>();
  float* w = tensor->data<float>();
  paddle::distributed::Region reg(w, tensor->numel());
  regions.emplace_back(std::move(reg));

  LOG(INFO) << "Run RunClient";
  RunClient(dense_regions);

  /*-----------------------Test Server Init----------------------------------*/
  LOG(INFO) << "Run pull_dense_param";
  float* temp = new float[tensor->numel()]();
  std::vector<paddle::distributed::Region> temp_region;
  paddle::distributed::Region temp_reg(temp, tensor->numel());
  temp_region.emplace_back(std::move(temp_reg));
  auto pull_status =
Z
zhaocaibei123 已提交
224
      worker_ptr_->PullDense(temp_region.data(), temp_region.size(), 0);
T
tangwei12 已提交
225 226
  pull_status.wait();

227
  for (int64_t idx = 0; idx < tensor->numel(); ++idx) {
T
tangwei12 已提交
228 229 230 231 232 233 234
    EXPECT_FLOAT_EQ(temp[idx], 1.0);
  }

  /*-----------------------Test Push Param----------------------------------*/

  LOG(INFO) << "Run push_dense_param";
  auto push_status =
Z
zhaocaibei123 已提交
235
      worker_ptr_->PushDenseParam(regions.data(), regions.size(), 0);
T
tangwei12 已提交
236 237
  push_status.wait();

Z
zhaocaibei123 已提交
238
  pull_status = worker_ptr_->PullDense(regions.data(), regions.size(), 0);
T
tangwei12 已提交
239 240
  pull_status.wait();

241
  for (int64_t idx = 0; idx < tensor->numel(); ++idx) {
242
    EXPECT_FLOAT_EQ(w[idx], static_cast<float>(idx));
T
tangwei12 已提交
243 244 245 246 247 248 249 250 251
  }

  /*-----------------------Test Push Grad----------------------------------*/

  paddle::distributed::DownpourBrpcClosure* closure =
      new paddle::distributed::DownpourBrpcClosure(1, [&](void* done) {
        int ret = 0;
        auto* closure = (paddle::distributed::DownpourBrpcClosure*)done;
        for (size_t i = 0; i < 1; ++i) {
T
tangwei12 已提交
252 253
          if (closure->check_response(
                  i, paddle::distributed::PS_PUSH_DENSE_TABLE) != 0) {
T
tangwei12 已提交
254 255 256 257 258 259 260 261 262
            ret = -1;
            break;
          }
        }
        closure->set_promise_value(ret);
      });

  LOG(INFO) << "Run pull_dense_grad";
  auto push_grad_status =
Z
zhaocaibei123 已提交
263
      worker_ptr_->PushDenseRawGradient(0, temp, tensor->numel(), closure);
T
tangwei12 已提交
264 265 266
  push_grad_status.wait();

  auto pull_update_status =
Z
zhaocaibei123 已提交
267
      worker_ptr_->PullDense(regions.data(), regions.size(), 0);
T
tangwei12 已提交
268 269
  pull_update_status.wait();

270
  for (int64_t idx = 0; idx < tensor->numel(); ++idx) {
271
    EXPECT_FLOAT_EQ(w[idx], static_cast<float>(idx) - 1.0);
T
tangwei12 已提交
272 273 274
  }

  LOG(INFO) << "Run stop_server";
Z
zhaocaibei123 已提交
275
  worker_ptr_->StopServer();
T
tangwei12 已提交
276
  LOG(INFO) << "Run finalize_worker";
Z
zhaocaibei123 已提交
277
  worker_ptr_->FinalizeWorker();
T
tangwei12 已提交
278 279 280 281
  server_thread.join();
}

TEST(RunBrpcPushDense, Run) { RunBrpcPushDense(); }