request_handler_impl.cc 5.1 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
// Copyright (c) 2018 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 <iostream>
#include <string>
#include <vector>

#include "paddle/fluid/framework/data_type.h"
#include "paddle/fluid/framework/lod_tensor.h"
#include "paddle/fluid/framework/scope.h"
#include "paddle/fluid/framework/selected_rows.h"
23 24
#include "paddle/fluid/operators/distributed/request_handler_impl.h"
#include "paddle/fluid/operators/distributed/rpc_server.h"
T
tangwei12 已提交
25
#include "paddle/fluid/string/printf.h"
26 27 28

namespace paddle {
namespace operators {
29
namespace distributed {
30

T
tangwei12 已提交
31 32
// define LOOKUP_TABLE_PATH for checkpoint notify to save lookup table variables
// to directory specified.
T
tangwei12 已提交
33
constexpr char LOOKUP_TABLE_PATH[] = "kLookupTablePath";
T
tangwei12 已提交
34

35 36 37
bool RequestSendHandler::Handle(const std::string& varname,
                                framework::Scope* scope,
                                framework::Variable* invar,
Q
qiaolongfei 已提交
38 39
                                framework::Variable** outvar,
                                const std::string& out_var_name) {
40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55
  VLOG(4) << "RequestSendHandler:" << varname;

  // Async
  if (!sync_mode_) {
    try {
      executor_->RunPreparedContext((*grad_to_prepared_ctx_)[varname].get(),
                                    scope);
    } catch (std::exception& e) {
      LOG(ERROR) << "async: run sub program error " << e.what();
      return false;
    }
    return true;
  }

  // Sync
  if (varname == BATCH_BARRIER_MESSAGE) {
56
    VLOG(3) << "sync: recv BATCH_BARRIER_MESSAGE";
57
    rpc_server_->IncreaseBatchBarrier(kRequestSend);
Y
Yancey1989 已提交
58 59 60
  } else if (varname == COMPLETE_MESSAGE) {
    VLOG(3) << "sync: recv complete message";
    rpc_server_->Complete();
61 62
  } else {
    VLOG(3) << "sync: received var_name: " << varname;
Y
Yancey1989 已提交
63 64
    rpc_server_->WaitCond(kRequestSend);
    VLOG(3) << "sync: processing received var: " << varname;
65 66

    if (invar == nullptr) {
67
      LOG(FATAL) << "sync: Can not find server side var: " << varname;
68 69
      return false;
    }
Y
Yancey1989 已提交
70
    if (invar->IsType<framework::SelectedRows>()) {
Y
Yancey1989 已提交
71 72
      std::unique_lock<std::mutex> lock(mutex_sparse_vars_);
      sparse_vars_.push_back(invar);
Y
Yancey1989 已提交
73
    }
74 75 76 77
  }
  return true;
}

Y
Yancey1989 已提交
78 79 80 81 82 83 84 85
void RequestSendHandler::ResetSparseVarRecorder() {
  std::unique_lock<std::mutex> lock(mutex_sparse_vars_);
  for (auto* var : sparse_vars_) {
    var->GetMutable<framework::SelectedRows>()->mutable_rows()->clear();
  }
  sparse_vars_.clear();
}

86 87 88
bool RequestGetHandler::Handle(const std::string& varname,
                               framework::Scope* scope,
                               framework::Variable* invar,
Q
qiaolongfei 已提交
89 90
                               framework::Variable** outvar,
                               const std::string& out_var_name) {
91
  VLOG(4) << "RequestGetHandler:" << varname;
Y
Yancey1989 已提交
92 93 94 95 96
  if (sync_mode_) {
    if (varname == FETCH_BARRIER_MESSAGE) {
      VLOG(3) << "sync: recv fetch barrier message";
      rpc_server_->IncreaseBatchBarrier(kRequestGet);
    } else {
97
      rpc_server_->WaitCond(kRequestGet);
Y
Yancey1989 已提交
98 99 100
      *outvar = scope_->FindVar(varname);
    }
  } else {
Y
Yancey1989 已提交
101
    if (varname != FETCH_BARRIER_MESSAGE && varname != COMPLETE_MESSAGE) {
Y
Yancey1989 已提交
102
      *outvar = scope_->FindVar(varname);
103 104 105 106 107 108 109 110
    }
  }
  return true;
}

bool RequestPrefetchHandler::Handle(const std::string& varname,
                                    framework::Scope* scope,
                                    framework::Variable* invar,
Q
qiaolongfei 已提交
111 112
                                    framework::Variable** outvar,
                                    const std::string& out_var_name) {
113 114
  VLOG(4) << "RequestPrefetchHandler " << varname;

Q
qiaolongfei 已提交
115
  auto var_desc = program_->Block(0).FindVar(out_var_name);
116
  InitializeVariable(*outvar, var_desc->GetType());
117 118
  executor_->RunPreparedContext(
      (*prefetch_var_name_to_prepared_ctx_)[varname].get(), scope);
119 120 121 122

  return true;
}

T
tangwei12 已提交
123 124 125 126
bool RequestCheckpointHandler::Handle(const std::string& varname,
                                      framework::Scope* scope,
                                      framework::Variable* invar,
                                      framework::Variable** outvar,
T
bug fix  
tangwei12 已提交
127
                                      const std::string& out_var_name) {
128 129 130
  PADDLE_ENFORCE(
      checkpoint_notify_id != -1,
      "when checkpoint_notify_id = -1, there should be no RPC invoke.");
T
tangwei12 已提交
131

T
tangwei12 已提交
132
  auto* lt_var = scope->FindVar(LOOKUP_TABLE_PATH)->GetMutable<std::string>();
T
tangwei12 已提交
133 134
  lt_var->clear();
  lt_var->append(out_var_name);
T
tangwei12 已提交
135
  VLOG(4) << "RequestCheckpointHandler update var kLookupTablePath to: "
136
          << out_var_name;
T
bug fix  
tangwei12 已提交
137
  executor_->RunPreparedContext(checkpoint_prepared_ctx_.get(), scope);
T
bug fix  
tangwei12 已提交
138 139
  return true;
}
T
tangwei12 已提交
140

141
}  // namespace distributed
142 143
}  // namespace operators
}  // namespace paddle