|
|
|
|
@ -16,6 +16,12 @@
|
|
|
|
|
#include "paddle/fluid/framework/details/container_cast.h"
|
|
|
|
|
#include "paddle/fluid/framework/details/reduce_and_gather.h"
|
|
|
|
|
#include "paddle/fluid/framework/details/variable_visitor.h"
|
|
|
|
|
#if defined PADDLE_WITH_CUDA && defined PADDLE_WITH_DISTRIBUTE
|
|
|
|
|
#include "paddle/fluid/operators/distributed/collective_client.h"
|
|
|
|
|
#include "paddle/fluid/operators/distributed/collective_server.h"
|
|
|
|
|
#include "paddle/fluid/operators/distributed/request_handler.h"
|
|
|
|
|
#endif
|
|
|
|
|
#include "paddle/fluid/operators/math/selected_rows_functor.h"
|
|
|
|
|
#include "paddle/fluid/platform/profiler.h"
|
|
|
|
|
|
|
|
|
|
DEFINE_bool(
|
|
|
|
|
@ -26,6 +32,112 @@ namespace paddle {
|
|
|
|
|
namespace framework {
|
|
|
|
|
namespace details {
|
|
|
|
|
|
|
|
|
|
std::once_flag CollectiveContext::init_flag_;
|
|
|
|
|
std::unique_ptr<CollectiveContext> CollectiveContext::context_;
|
|
|
|
|
|
|
|
|
|
static inline std::string GetRemoteVarName(const std::string &var_name,
|
|
|
|
|
int trainer_id) {
|
|
|
|
|
return string::Sprintf("%s_merged_tmp@trainer_%d", var_name, trainer_id);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
void ReduceOpHandle::Wait(
|
|
|
|
|
const std::map<platform::Place, platform::DeviceContext *> &dev_ctxes) {
|
|
|
|
|
// TODO(gongwb): use event wait?
|
|
|
|
|
for (auto &dev_ctx : dev_ctxes) {
|
|
|
|
|
dev_ctx.second->Wait();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#if defined PADDLE_WITH_CUDA && defined PADDLE_WITH_DISTRIBUTE
|
|
|
|
|
template <typename DevCtx, typename DataType>
|
|
|
|
|
void ReduceOpHandle::GatherSelectedRows(
|
|
|
|
|
const std::vector<const SelectedRows *> &src_selected_rows,
|
|
|
|
|
const std::vector<platform::Place> &in_places,
|
|
|
|
|
const std::map<platform::Place, platform::DeviceContext *> &dev_ctxes,
|
|
|
|
|
VarHandle *out_var_handle, const platform::Place &out_place,
|
|
|
|
|
SelectedRows *dst_selected_rows) {
|
|
|
|
|
const CollectiveContext &collective_context =
|
|
|
|
|
*CollectiveContext::GetInstance();
|
|
|
|
|
|
|
|
|
|
// 1. gather local selected rows, merge them
|
|
|
|
|
std::string gathered_var_name = out_var_handle->name_ + "_gathered_tmp";
|
|
|
|
|
auto scope = local_scopes_.at(out_var_handle->scope_idx_);
|
|
|
|
|
auto gathered_var_mid = scope->Var(gathered_var_name);
|
|
|
|
|
auto gathered_select_rows =
|
|
|
|
|
gathered_var_mid->GetMutable<framework::SelectedRows>();
|
|
|
|
|
GatherLocalSelectedRows(src_selected_rows, in_places, dev_ctxes, out_place,
|
|
|
|
|
gathered_select_rows);
|
|
|
|
|
// FIXME(gongwb): remove this Wait.
|
|
|
|
|
Wait(dev_ctxes);
|
|
|
|
|
|
|
|
|
|
// merge them
|
|
|
|
|
auto merged_dev_ctx = dynamic_cast<DevCtx *>(dev_ctxes.at(out_place));
|
|
|
|
|
std::string merged_var_name =
|
|
|
|
|
GetRemoteVarName(out_var_handle->name_, collective_context.trainer_id_);
|
|
|
|
|
auto merged_select_rows =
|
|
|
|
|
scope->Var(merged_var_name)->GetMutable<SelectedRows>();
|
|
|
|
|
operators::math::scatter::MergeAdd<DevCtx, DataType> merge_func;
|
|
|
|
|
merge_func(*merged_dev_ctx, *gathered_select_rows, merged_select_rows);
|
|
|
|
|
|
|
|
|
|
// 2. start collective server if it doesn't exist
|
|
|
|
|
operators::distributed::CollectiveServer *server =
|
|
|
|
|
operators::distributed::CollectiveServer::GetInstance(
|
|
|
|
|
collective_context.endpoints_[collective_context.trainer_id_],
|
|
|
|
|
collective_context.endpoints_.size() - 1);
|
|
|
|
|
|
|
|
|
|
auto rpc_server = server->GetRPCServer();
|
|
|
|
|
rpc_server->RegisterVar(merged_var_name,
|
|
|
|
|
operators::distributed::kRequestGetMonomerVariable,
|
|
|
|
|
scope, merged_dev_ctx);
|
|
|
|
|
|
|
|
|
|
// 3. gather them from all remote nodes.
|
|
|
|
|
std::vector<const SelectedRows *> remote;
|
|
|
|
|
operators::distributed::CollectiveClient *client =
|
|
|
|
|
operators::distributed::CollectiveClient::GetInstance();
|
|
|
|
|
|
|
|
|
|
std::vector<operators::distributed::RemoteVar> vars;
|
|
|
|
|
for (unsigned int i = 0; i < collective_context.endpoints_.size(); i++) {
|
|
|
|
|
if (i == (unsigned)collective_context.trainer_id_) continue;
|
|
|
|
|
|
|
|
|
|
operators::distributed::RemoteVar var;
|
|
|
|
|
var.trainer_id_ = i;
|
|
|
|
|
var.var_name_ = GetRemoteVarName(out_var_handle->name_, i);
|
|
|
|
|
var.ep_ = collective_context.endpoints_[i];
|
|
|
|
|
|
|
|
|
|
vars.push_back(var);
|
|
|
|
|
VLOG(4) << "gather from:" << var.String();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// erase gathered vars
|
|
|
|
|
merged_dev_ctx->Wait();
|
|
|
|
|
scope->EraseVars(std::vector<std::string>{gathered_var_name});
|
|
|
|
|
|
|
|
|
|
PADDLE_ENFORCE(client->Gather(vars, &remote, *merged_dev_ctx, scope));
|
|
|
|
|
PADDLE_ENFORCE(remote.size() == vars.size());
|
|
|
|
|
|
|
|
|
|
// 4. merged local selected rows.
|
|
|
|
|
std::vector<const SelectedRows *> all;
|
|
|
|
|
all.resize(collective_context.endpoints_.size());
|
|
|
|
|
for (auto v : vars) {
|
|
|
|
|
all[v.trainer_id_] =
|
|
|
|
|
scope->FindVar(v.var_name_)->GetMutable<SelectedRows>();
|
|
|
|
|
}
|
|
|
|
|
all[collective_context.trainer_id_] = merged_select_rows;
|
|
|
|
|
|
|
|
|
|
merge_func(*merged_dev_ctx, all, dst_selected_rows);
|
|
|
|
|
|
|
|
|
|
rpc_server->WaitVarBarrier(merged_var_name);
|
|
|
|
|
rpc_server->ClearVar(merged_var_name);
|
|
|
|
|
|
|
|
|
|
// 5. clear mid vars
|
|
|
|
|
std::vector<std::string> tmp_vars{merged_var_name};
|
|
|
|
|
for (auto r : vars) {
|
|
|
|
|
tmp_vars.push_back(r.var_name_);
|
|
|
|
|
}
|
|
|
|
|
scope->EraseVars(tmp_vars);
|
|
|
|
|
}
|
|
|
|
|
#endif
|
|
|
|
|
|
|
|
|
|
void ReduceOpHandle::RunImpl() {
|
|
|
|
|
platform::RecordEvent record_event(Name(), dev_ctxes_.cbegin()->second);
|
|
|
|
|
|
|
|
|
|
@ -90,8 +202,36 @@ void ReduceOpHandle::RunImpl() {
|
|
|
|
|
this->RunAndRecordEvent([&] {
|
|
|
|
|
std::vector<const SelectedRows *> in_selected_rows =
|
|
|
|
|
GetInputValues<SelectedRows>(in_var_handles, var_scopes);
|
|
|
|
|
GatherSelectedRows(in_selected_rows, in_places, dev_ctxes_, t_out_p,
|
|
|
|
|
|
|
|
|
|
const CollectiveContext &collective_context =
|
|
|
|
|
*CollectiveContext::GetInstance();
|
|
|
|
|
VLOG(10) << "GatherSelectedRows CollectiveContext:"
|
|
|
|
|
<< collective_context.String();
|
|
|
|
|
|
|
|
|
|
// TODO(gongwb): add cpu support
|
|
|
|
|
if (collective_context.endpoints_.size() <= 1 ||
|
|
|
|
|
is_cpu_place(in_places[0]) || is_cpu_place(t_out_p)) {
|
|
|
|
|
GatherLocalSelectedRows(in_selected_rows, in_places, dev_ctxes_,
|
|
|
|
|
t_out_p,
|
|
|
|
|
out_var->GetMutable<framework::SelectedRows>());
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
#if defined PADDLE_WITH_CUDA && defined PADDLE_WITH_DISTRIBUTE
|
|
|
|
|
if (framework::IsType<const float>(in_selected_rows[0]->value().type())) {
|
|
|
|
|
GatherSelectedRows<platform::CUDADeviceContext, float>(
|
|
|
|
|
in_selected_rows, in_places, dev_ctxes_, out_var_handle, t_out_p,
|
|
|
|
|
out_var->GetMutable<framework::SelectedRows>());
|
|
|
|
|
} else if (framework::IsType<const double>(
|
|
|
|
|
in_selected_rows[0]->value().type())) {
|
|
|
|
|
GatherSelectedRows<platform::CUDADeviceContext, double>(
|
|
|
|
|
in_selected_rows, in_places, dev_ctxes_, out_var_handle, t_out_p,
|
|
|
|
|
out_var->GetMutable<framework::SelectedRows>());
|
|
|
|
|
} else {
|
|
|
|
|
PADDLE_ENFORCE(false,
|
|
|
|
|
"only support double or float when gahter SelectedRows");
|
|
|
|
|
}
|
|
|
|
|
#endif
|
|
|
|
|
});
|
|
|
|
|
} else {
|
|
|
|
|
std::vector<const LoDTensor *> lod_tensors =
|
|
|
|
|
|
tensor-tang
6 years ago