提交 dfbe06cc 编写于 作者: Y yuyang18

Merge branch 'develop' of https://github.com/PaddlePaddle/Paddle into feature/fix_ninja_build

...@@ -70,6 +70,12 @@ copy(glog_lib ...@@ -70,6 +70,12 @@ copy(glog_lib
DSTS ${dst_dir} ${dst_dir}/lib DSTS ${dst_dir} ${dst_dir}/lib
) )
set(dst_dir "${CMAKE_INSTALL_PREFIX}/third_party/boost/")
copy(boost_lib
SRCS ${BOOST_INCLUDE_DIR}/boost
DSTS ${dst_dir}
)
if(NOT PROTOBUF_FOUND) if(NOT PROTOBUF_FOUND)
set(dst_dir "${CMAKE_INSTALL_PREFIX}/third_party/install/protobuf") set(dst_dir "${CMAKE_INSTALL_PREFIX}/third_party/install/protobuf")
copy(protobuf_lib copy(protobuf_lib
......
...@@ -4,34 +4,37 @@ ...@@ -4,34 +4,37 @@
For the typical synchronous distributed training, some significant steps are as follows: For the typical synchronous distributed training, some significant steps are as follows:
1. A Trainer will compute the gradients and SEND them to the Parameter Server(PServer) nodes. 1. A trainer process will compute the gradients and **send** them to the parameter server (PS) nodes.
1. After the PServer node received gradients came from all the Trainers, It will aggregate the 1. After the PS node received gradients came from all the Trainers, It will aggregate the
gradient variables for the same parameter into one gradient variable and then apply the aggregated gradient variables for the same parameter into one gradient variable and then apply the aggregated
gradient to the respective parameter, finally using an optimize algorithms(SGD, Monument...) gradient to the respective parameter, finally using an optimize algorithms(SGD, Monument...)
to update the parameters. to update the parameters.
1. The Trainer would wait for the PServers finished the optimize stage, and GET the parameters from PServer, 1. The Trainer would wait for the PS finished the optimize stage, and GET the parameters from PS,
so all the Trainers would get the same parameters. so all the Trainers would get the same parameters.
In the synchronously distributed training, there should be a `Barrier` to synchronise the In Synchronous Distributed Training, there is a **barrier** on each PS to wait until all trainers processes
parameters after the optimizing stage. The performance of a distributed training job would have completed running current mini-batch. After that, all trainers can continue to run the next
depend on the slowest node if there were hundreds or thousands of training nodes in a mini-batch. So, we can find that the overall performance of Synchronous Distributed Training depends
Job, the performance of synchronously distributed training might be very poor because of on the slowest node.
the slow node. So this design doc would introduce an approach to implement
*asynchronously* distributed training in PaddlePaddle Fluid. In Asynchronous Distributed Training, we don't need to wait for a global mini-bach, the optimizer on
the PS will run immediately when the gradient is uploaded to the PS from one trainer. This mode would
train such models that achieve scaling, better throughput. In this design doc, we will introduce how to
implement the Asynchronous Distributed Training base on PaddlePaddle Fluid.
## Design ## Design
<img src="./src/async_update.png" width="600"/> <img src="./src/async_update.png" width="600"/>
As the figure above, we describe a global view of asynchronously update process and use As the figure above, we describe a global view of the asynchronous update process and use
the parameter `w1` as an example to introduce the steps: the parameter `w1` as an example to introduce the steps:
1. For each gradient variables, they may distribute on different GPU card and aggregate 1. For each gradient variables, they may distribute on different GPU card and aggregate
them while they are all calculated. them while they are all calculated.
1. Split the gradient variable into multiple blocks according to the number of PServer 1. Split the gradient variable into multiple blocks according to the number of PS
instances and then send them. instances and then send them.
1. PServer would run an `Optimize Block` using a specified optimize algorithm to update 1. PS would run an `Optimize Block` using a specified optimize algorithm to update
the specified parameter. the specified parameter.
1. The trainer will fetch latest parameter from PServer before running forward Op which depends 1. The trainer will fetch the latest parameter from PS before running forward Op which depends
on the specified parameter. on the specified parameter.
1. Broadcast the received variable into multiple GPU cards and continue to run the next 1. Broadcast the received variable into multiple GPU cards and continue to run the next
mini-batch. mini-batch.
...@@ -40,8 +43,8 @@ mini-batch. ...@@ -40,8 +43,8 @@ mini-batch.
- For the multiple devices distributed training, we need to aggregate the gradient - For the multiple devices distributed training, we need to aggregate the gradient
variables which placed on different devices firstly and then schedule a `SendVars` Operator to variables which placed on different devices firstly and then schedule a `SendVars` Operator to
send the gradient variables to the multiple PServer instances. send the gradient variables to the multiple PS instances.
- Schedule `FetchVars` operator to fetch the latest parameter from PServer before running - Schedule `FetchVars` operator to fetch the latest parameter from PS before running
the forward ops. the forward ops.
- There could be a large number of gradient variables to be sent, so we need to use another - There could be a large number of gradient variables to be sent, so we need to use another
thread pool(IO Threadpool) whose a number of the schedulable threads is larger than the thread pool(IO Threadpool) whose a number of the schedulable threads is larger than the
......
...@@ -19,8 +19,9 @@ ...@@ -19,8 +19,9 @@
---------------- ----------------
PaddlePaddle需要使用Docker环境完成编译,这样可以免去单独安装编译依赖的步骤,可选的不同编译环境Docker镜像 PaddlePaddle需要使用Docker环境完成编译,这样可以免去单独安装编译依赖的步骤,可选的不同编译环境Docker镜像
可以在 `这里 <https://hub.docker.com/r/paddlepaddle/paddle_manylinux_devel/tags/>`_ 找到。或者 可以在 `这里 <https://hub.docker.com/r/paddlepaddle/paddle_manylinux_devel/tags/>`_ 找到,您也可以
参考下述可选步骤,从源码中构建用于编译PaddlePaddle的Docker镜像。 在 `这里 <https://github.com/PaddlePaddle/Paddle/tree/develop/tools/manylinux1/>`_ 找到 paddle_manylinux_devel
镜像的编译以及使用方法。或者参考下述可选步骤,从源码中构建用于编译PaddlePaddle的Docker镜像。
如果您选择不使用Docker镜像,则需要在本机安装下面章节列出的 `编译依赖`_ 之后才能开始编译的步骤。 如果您选择不使用Docker镜像,则需要在本机安装下面章节列出的 `编译依赖`_ 之后才能开始编译的步骤。
......
...@@ -22,6 +22,8 @@ How To Build ...@@ -22,6 +22,8 @@ How To Build
You need to use Docker to build PaddlePaddle You need to use Docker to build PaddlePaddle
to avoid installing dependencies by yourself. We have several pre-built to avoid installing dependencies by yourself. We have several pre-built
Docker images `here <https://hub.docker.com/r/paddlepaddle/paddle_manylinux_devel/tags/>`_ , Docker images `here <https://hub.docker.com/r/paddlepaddle/paddle_manylinux_devel/tags/>`_ ,
you can also find how to build and use paddle_manylinux_devel Docker image from
`here <https://github.com/PaddlePaddle/Paddle/tree/develop/tools/manylinux1/>`_
Or you can build your own image from source as the optional step below: Or you can build your own image from source as the optional step below:
.. code-block:: bash .. code-block:: bash
......
...@@ -5,11 +5,11 @@ proto_library(framework_proto SRCS framework.proto) ...@@ -5,11 +5,11 @@ proto_library(framework_proto SRCS framework.proto)
cc_library(ddim SRCS ddim.cc DEPS eigen3 boost) cc_library(ddim SRCS ddim.cc DEPS eigen3 boost)
cc_test(ddim_test SRCS ddim_test.cc DEPS ddim) cc_test(ddim_test SRCS ddim_test.cc DEPS ddim)
nv_test(dim_test SRCS dim_test.cu DEPS ddim) nv_test(dim_test SRCS dim_test.cu DEPS ddim)
cc_library(data_type SRCS data_type.cc DEPS framework_proto ddim device_context)
if(WITH_GPU) if(WITH_GPU)
nv_library(tensor SRCS tensor.cc tensor_util.cu DEPS ddim place memory device_context framework_proto) nv_library(tensor SRCS tensor.cc tensor_util.cu DEPS place memory data_type)
else() else()
cc_library(tensor SRCS tensor.cc tensor_util.cc DEPS ddim place memory device_context framework_proto) cc_library(tensor SRCS tensor.cc tensor_util.cc DEPS place memory data_type)
endif() endif()
cc_test(tensor_test SRCS tensor_test.cc DEPS tensor) cc_test(tensor_test SRCS tensor_test.cc DEPS tensor)
......
// 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 "paddle/fluid/framework/data_type.h"
#include <stdint.h>
#include <string>
#include <unordered_map>
namespace paddle {
namespace framework {
struct DataTypeMap {
std::unordered_map<std::type_index, proto::VarType::Type> cpp_to_proto_;
std::unordered_map<int, std::type_index> proto_to_cpp_;
std::unordered_map<int, std::string> proto_to_str_;
std::unordered_map<std::type_index, size_t> cpp_to_size_;
};
static DataTypeMap* InitDataTypeMap();
static DataTypeMap& gDataTypeMap() {
static DataTypeMap* g_data_type_map_ = InitDataTypeMap();
return *g_data_type_map_;
}
template <typename T>
static inline void RegisterType(DataTypeMap* map,
proto::VarType::Type proto_type,
const std::string& name) {
map->proto_to_cpp_.emplace(static_cast<int>(proto_type), typeid(T));
map->cpp_to_proto_.emplace(typeid(T), proto_type);
map->proto_to_str_.emplace(static_cast<int>(proto_type), name);
map->cpp_to_size_.emplace(typeid(T), sizeof(T));
}
static DataTypeMap* InitDataTypeMap() {
auto retv = new DataTypeMap();
#define RegType(cc_type, proto_type) \
RegisterType<cc_type>(retv, proto_type, #cc_type)
// NOTE: Add your customize type here.
RegType(platform::float16, proto::VarType::FP16);
RegType(float, proto::VarType::FP32);
RegType(double, proto::VarType::FP64);
RegType(int, proto::VarType::INT32);
RegType(int64_t, proto::VarType::INT64);
RegType(bool, proto::VarType::BOOL);
RegType(size_t, proto::VarType::SIZE_T);
RegType(int16_t, proto::VarType::INT16);
#undef RegType
return retv;
}
proto::VarType::Type ToDataType(std::type_index type) {
auto it = gDataTypeMap().cpp_to_proto_.find(type);
if (it != gDataTypeMap().cpp_to_proto_.end()) {
return it->second;
}
PADDLE_THROW("Not support %s as tensor type", type.name());
}
std::type_index ToTypeIndex(proto::VarType::Type type) {
auto it = gDataTypeMap().proto_to_cpp_.find(static_cast<int>(type));
if (it != gDataTypeMap().proto_to_cpp_.end()) {
return it->second;
}
PADDLE_THROW("Not support proto::VarType::Type(%d) as tensor type",
static_cast<int>(type));
}
std::string DataTypeToString(const proto::VarType::Type type) {
auto it = gDataTypeMap().proto_to_str_.find(static_cast<int>(type));
if (it != gDataTypeMap().proto_to_str_.end()) {
return it->second;
}
PADDLE_THROW("Not support proto::VarType::Type(%d) as tensor type",
static_cast<int>(type));
}
size_t SizeOfType(std::type_index type) {
auto it = gDataTypeMap().cpp_to_size_.find(type);
if (it != gDataTypeMap().cpp_to_size_.end()) {
return it->second;
}
PADDLE_THROW("Not support %s as tensor type", type.name());
}
} // namespace framework
} // namespace paddle
...@@ -17,51 +17,14 @@ limitations under the License. */ ...@@ -17,51 +17,14 @@ limitations under the License. */
#include <typeindex> #include <typeindex>
#include "paddle/fluid/framework/framework.pb.h" #include "paddle/fluid/framework/framework.pb.h"
#include "paddle/fluid/platform/enforce.h" #include "paddle/fluid/platform/enforce.h"
#include "paddle/fluid/platform/float16.h" #include "paddle/fluid/platform/float16.h"
namespace paddle { namespace paddle {
namespace framework { namespace framework {
inline proto::VarType::Type ToDataType(std::type_index type) { extern proto::VarType::Type ToDataType(std::type_index type);
if (typeid(platform::float16).hash_code() == type.hash_code()) { extern std::type_index ToTypeIndex(proto::VarType::Type type);
return proto::VarType::FP16;
} else if (typeid(const float).hash_code() == type.hash_code()) {
// CPPLint complains Using C-style cast. Use static_cast<float>() instead
// One fix to this is to replace float with const float because
// typeid(T) == typeid(const T)
// http://en.cppreference.com/w/cpp/language/typeid
return proto::VarType::FP32;
} else if (typeid(const double).hash_code() == type.hash_code()) {
return proto::VarType::FP64;
} else if (typeid(const int).hash_code() == type.hash_code()) {
return proto::VarType::INT32;
} else if (typeid(const int64_t).hash_code() == type.hash_code()) {
return proto::VarType::INT64;
} else if (typeid(const bool).hash_code() == type.hash_code()) {
return proto::VarType::BOOL;
} else {
PADDLE_THROW("Not supported");
}
}
inline std::type_index ToTypeIndex(proto::VarType::Type type) {
switch (type) {
case proto::VarType::FP16:
return typeid(platform::float16);
case proto::VarType::FP32:
return typeid(float);
case proto::VarType::FP64:
return typeid(double);
case proto::VarType::INT32:
return typeid(int);
case proto::VarType::INT64:
return typeid(int64_t);
case proto::VarType::BOOL:
return typeid(bool);
default:
PADDLE_THROW("Not support type %d", type);
}
}
template <typename Visitor> template <typename Visitor>
inline void VisitDataType(proto::VarType::Type type, Visitor visitor) { inline void VisitDataType(proto::VarType::Type type, Visitor visitor) {
...@@ -89,32 +52,12 @@ inline void VisitDataType(proto::VarType::Type type, Visitor visitor) { ...@@ -89,32 +52,12 @@ inline void VisitDataType(proto::VarType::Type type, Visitor visitor) {
} }
} }
inline std::string DataTypeToString(const proto::VarType::Type type) { extern std::string DataTypeToString(const proto::VarType::Type type);
switch (type) { extern size_t SizeOfType(std::type_index type);
case proto::VarType::FP16:
return "float16";
case proto::VarType::FP32:
return "float32";
case proto::VarType::FP64:
return "float64";
case proto::VarType::INT16:
return "int16";
case proto::VarType::INT32:
return "int32";
case proto::VarType::INT64:
return "int64";
case proto::VarType::BOOL:
return "bool";
default:
PADDLE_THROW("Not support type %d", type);
}
}
inline std::ostream& operator<<(std::ostream& out, inline std::ostream& operator<<(std::ostream& out,
const proto::VarType::Type& type) { const proto::VarType::Type& type) {
out << DataTypeToString(type); out << DataTypeToString(type);
return out; return out;
} }
} // namespace framework } // namespace framework
} // namespace paddle } // namespace paddle
// 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.
#pragma once
namespace paddle {
namespace framework {
namespace details {
struct BuildStrategy {
enum class ReduceStrategy { kAllReduce = 0, kReduce = 1 };
enum class GradientScaleStrategy {
kCoeffNumDevice = 0,
kOne = 1,
kCustomized = 2,
};
ReduceStrategy reduce_{ReduceStrategy::kAllReduce};
GradientScaleStrategy gradient_scale_{GradientScaleStrategy::kCoeffNumDevice};
};
} // namespace details
} // namespace framework
} // namespace paddle
// 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.
#pragma once
namespace paddle {
namespace framework {
namespace details {
struct ExecutionStrategy {
size_t num_threads_{0};
bool use_event_{true};
bool allow_op_delay_{false};
};
} // namespace details
} // namespace framework
} // namespace paddle
...@@ -37,31 +37,26 @@ MultiDevSSAGraphBuilder::MultiDevSSAGraphBuilder( ...@@ -37,31 +37,26 @@ MultiDevSSAGraphBuilder::MultiDevSSAGraphBuilder(
const std::string &loss_var_name, const std::string &loss_var_name,
const std::unordered_set<std::string> &params, const std::unordered_set<std::string> &params,
const std::vector<Scope *> &local_scopes, const std::vector<Scope *> &local_scopes,
platform::NCCLContextMap *nccl_ctxs, bool use_default_grad_scale, platform::NCCLContextMap *nccl_ctxs, const BuildStrategy &strategy)
bool balance_parameter_opt_between_cards)
: loss_var_name_(loss_var_name), : loss_var_name_(loss_var_name),
places_(places), places_(places),
local_scopes_(local_scopes), local_scopes_(local_scopes),
nccl_ctxs_(nccl_ctxs), nccl_ctxs_(nccl_ctxs),
balance_parameter_opt_between_cards_( strategy_(strategy) {
balance_parameter_opt_between_cards) {
#else #else
MultiDevSSAGraphBuilder::MultiDevSSAGraphBuilder( MultiDevSSAGraphBuilder::MultiDevSSAGraphBuilder(
const std::vector<platform::Place> &places, const std::vector<platform::Place> &places,
const std::string &loss_var_name, const std::string &loss_var_name,
const std::unordered_set<std::string> &params, const std::unordered_set<std::string> &params,
const std::vector<Scope *> &local_scopes, bool use_default_grad_scale, const std::vector<Scope *> &local_scopes, const BuildStrategy &strategy)
bool balance_parameter_opt_between_cards)
: loss_var_name_(loss_var_name), : loss_var_name_(loss_var_name),
places_(places), places_(places),
local_scopes_(local_scopes), local_scopes_(local_scopes),
balance_parameter_opt_between_cards_( strategy_(strategy) {
balance_parameter_opt_between_cards) {
#endif #endif
for (auto &p : params) { for (auto &p : params) {
grad_names_.insert(GradVarName(p)); grad_names_.insert(GradVarName(p));
} }
use_default_grad_scale_ = use_default_grad_scale;
} }
void MultiDevSSAGraphBuilder::CreateOpHandleIOs(SSAGraph *result, void MultiDevSSAGraphBuilder::CreateOpHandleIOs(SSAGraph *result,
...@@ -146,7 +141,8 @@ std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build( ...@@ -146,7 +141,8 @@ std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build(
CreateComputationalOps(&result, *op, 1); CreateComputationalOps(&result, *op, 1);
} else if (IsScaleLossOp(*op)) { } else if (IsScaleLossOp(*op)) {
// user can customize loss@grad if not use_default_grad_scale_ // user can customize loss@grad if not use_default_grad_scale_
if (use_default_grad_scale_) { if (strategy_.gradient_scale_ !=
BuildStrategy::GradientScaleStrategy::kCustomized) {
CreateScaleLossGradOp(&result); CreateScaleLossGradOp(&result);
} }
is_forwarding = false; is_forwarding = false;
...@@ -165,19 +161,22 @@ std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build( ...@@ -165,19 +161,22 @@ std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build(
// broadcast, and each gradient is only broadcast once. // broadcast, and each gradient is only broadcast once.
for (auto &og : op->OutputArgumentNames()) { for (auto &og : op->OutputArgumentNames()) {
if (IsParameterGradientOnce(og, &og_has_been_broadcast)) { if (IsParameterGradientOnce(og, &og_has_been_broadcast)) {
if (balance_parameter_opt_between_cards_) { switch (strategy_.reduce_) {
CreateReduceOp(&result, og, cur_device_id); case BuildStrategy::ReduceStrategy::kReduce:
var_name_on_devices[cur_device_id].emplace(og); CreateReduceOp(&result, og, cur_device_id);
bcast_var_name_set[cur_device_id].emplace( var_name_on_devices[cur_device_id].emplace(og);
og.substr(0, og.size() - strlen(kGradVarSuffix))); bcast_var_name_set[cur_device_id].emplace(
cur_device_id = (cur_device_id + 1) % places_.size(); og.substr(0, og.size() - strlen(kGradVarSuffix)));
} else { cur_device_id = (cur_device_id + 1) % places_.size();
if (IsSparseGradient(var_types, og)) { break;
CreateReduceOp(&result, og, 0); case BuildStrategy::ReduceStrategy::kAllReduce:
CreateBroadcastOp(&result, og, 0); if (IsSparseGradient(var_types, og)) {
} else { CreateReduceOp(&result, og, 0);
InsertNCCLAllReduceOp(&result, og); CreateBroadcastOp(&result, og, 0);
} } else {
InsertNCCLAllReduceOp(&result, og);
}
break;
} }
} }
} }
...@@ -303,7 +302,7 @@ bool MultiDevSSAGraphBuilder::IsParameterGradientOnce( ...@@ -303,7 +302,7 @@ bool MultiDevSSAGraphBuilder::IsParameterGradientOnce(
int MultiDevSSAGraphBuilder::GetOpDeviceID( int MultiDevSSAGraphBuilder::GetOpDeviceID(
const std::vector<std::unordered_set<std::string>> &var_name_on_devices, const std::vector<std::unordered_set<std::string>> &var_name_on_devices,
const OpDesc &op) const { const OpDesc &op) const {
if (!balance_parameter_opt_between_cards_) { if (strategy_.reduce_ != BuildStrategy::ReduceStrategy::kReduce) {
return -1; return -1;
} }
......
...@@ -17,6 +17,7 @@ ...@@ -17,6 +17,7 @@
#include <utility> #include <utility>
#include <vector> #include <vector>
#include "paddle/fluid/framework/details/build_strategy.h"
#include "paddle/fluid/framework/details/ssa_graph_builder.h" #include "paddle/fluid/framework/details/ssa_graph_builder.h"
namespace paddle { namespace paddle {
...@@ -36,15 +37,13 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder { ...@@ -36,15 +37,13 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder {
const std::unordered_set<std::string> &params, const std::unordered_set<std::string> &params,
const std::vector<Scope *> &local_scopes, const std::vector<Scope *> &local_scopes,
platform::NCCLContextMap *nccl_ctxs, platform::NCCLContextMap *nccl_ctxs,
bool use_default_grad_scale, const BuildStrategy &strategy);
bool balance_parameter_opt_between_cards);
#else #else
MultiDevSSAGraphBuilder(const std::vector<platform::Place> &places, MultiDevSSAGraphBuilder(const std::vector<platform::Place> &places,
const std::string &loss_var_name, const std::string &loss_var_name,
const std::unordered_set<std::string> &params, const std::unordered_set<std::string> &params,
const std::vector<Scope *> &local_scopes, const std::vector<Scope *> &local_scopes,
bool use_default_grad_scale, const BuildStrategy &strategy);
bool balance_parameter_opt_between_cards);
#endif #endif
std::unique_ptr<SSAGraph> Build(const ProgramDesc &program) const override; std::unique_ptr<SSAGraph> Build(const ProgramDesc &program) const override;
...@@ -62,8 +61,6 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder { ...@@ -62,8 +61,6 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder {
#ifdef PADDLE_WITH_CUDA #ifdef PADDLE_WITH_CUDA
platform::NCCLContextMap *nccl_ctxs_; platform::NCCLContextMap *nccl_ctxs_;
#endif #endif
bool balance_parameter_opt_between_cards_;
bool use_default_grad_scale_;
bool IsScaleLossOp(const OpDesc &op) const; bool IsScaleLossOp(const OpDesc &op) const;
...@@ -105,6 +102,9 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder { ...@@ -105,6 +102,9 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder {
bool IsSparseGradient( bool IsSparseGradient(
const std::unordered_map<std::string, proto::VarType::Type> &var_types, const std::unordered_map<std::string, proto::VarType::Type> &var_types,
const std::string &og) const; const std::string &og) const;
private:
BuildStrategy strategy_;
}; };
} // namespace details } // namespace details
} // namespace framework } // namespace framework
......
...@@ -18,18 +18,17 @@ namespace paddle { ...@@ -18,18 +18,17 @@ namespace paddle {
namespace framework { namespace framework {
namespace details { namespace details {
ThreadedSSAGraphExecutor::ThreadedSSAGraphExecutor( ThreadedSSAGraphExecutor::ThreadedSSAGraphExecutor(
size_t num_threads, bool use_event, const ExecutionStrategy &strategy, const std::vector<Scope *> &local_scopes,
const std::vector<Scope *> &local_scopes,
const std::vector<platform::Place> &places, const std::vector<platform::Place> &places,
std::unique_ptr<SSAGraph> &&graph, bool allow_op_delay) std::unique_ptr<SSAGraph> &&graph)
: SSAGraphExecutor(std::move(graph)), : SSAGraphExecutor(std::move(graph)),
pool_(num_threads >= 2 ? new ::ThreadPool(num_threads) : nullptr), pool_(strategy.num_threads_ >= 2 ? new ::ThreadPool(strategy.num_threads_)
: nullptr),
local_scopes_(local_scopes), local_scopes_(local_scopes),
places_(places), places_(places),
fetch_ctxs_(places), fetch_ctxs_(places),
use_event_(use_event),
running_ops_(0), running_ops_(0),
allow_op_delay_(allow_op_delay) {} strategy_(strategy) {}
FeedFetchList ThreadedSSAGraphExecutor::Run( FeedFetchList ThreadedSSAGraphExecutor::Run(
const std::vector<std::string> &fetch_tensors) { const std::vector<std::string> &fetch_tensors) {
...@@ -86,7 +85,7 @@ FeedFetchList ThreadedSSAGraphExecutor::Run( ...@@ -86,7 +85,7 @@ FeedFetchList ThreadedSSAGraphExecutor::Run(
// //
// NOTE: DelayedOps have a lower priority. It will be scheduled after all // NOTE: DelayedOps have a lower priority. It will be scheduled after all
// ready_ops have been performed. // ready_ops have been performed.
if (ready_ops.empty() && allow_op_delay_ && running_ops_ == 0) { if (ready_ops.empty() && strategy_.allow_op_delay_ && running_ops_ == 0) {
run_all_ops(delayed_ops); run_all_ops(delayed_ops);
} else { } else {
run_all_ops(ready_ops); run_all_ops(ready_ops);
...@@ -113,7 +112,7 @@ FeedFetchList ThreadedSSAGraphExecutor::Run( ...@@ -113,7 +112,7 @@ FeedFetchList ThreadedSSAGraphExecutor::Run(
auto &deps = pending_ops[op]; auto &deps = pending_ops[op];
--deps; --deps;
if (deps == 0) { if (deps == 0) {
if (op->IsMultiDeviceTransfer() && allow_op_delay_) { if (op->IsMultiDeviceTransfer() && strategy_.allow_op_delay_) {
delayed_ops.insert(op); delayed_ops.insert(op);
} else { } else {
ready_ops.insert(op); ready_ops.insert(op);
...@@ -191,7 +190,7 @@ void ThreadedSSAGraphExecutor::RunOp( ...@@ -191,7 +190,7 @@ void ThreadedSSAGraphExecutor::RunOp(
auto op_run = [ready_var_q, op, this] { auto op_run = [ready_var_q, op, this] {
try { try {
VLOG(10) << op << " " << op->Name() << " : " << op->DebugString(); VLOG(10) << op << " " << op->Name() << " : " << op->DebugString();
op->Run(use_event_); op->Run(strategy_.use_event_);
VLOG(10) << op << " " << op->Name() << " Done "; VLOG(10) << op << " " << op->Name() << " Done ";
running_ops_--; running_ops_--;
ready_var_q->Extend(op->Outputs()); ready_var_q->Extend(op->Outputs());
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <functional> #include <functional>
#include "ThreadPool.h" // ThreadPool in thrird party #include "ThreadPool.h" // ThreadPool in thrird party
#include "paddle/fluid/framework/blocking_queue.h" #include "paddle/fluid/framework/blocking_queue.h"
#include "paddle/fluid/framework/details/execution_strategy.h"
#include "paddle/fluid/framework/details/fetch_op_handle.h" #include "paddle/fluid/framework/details/fetch_op_handle.h"
#include "paddle/fluid/framework/details/ssa_graph_executor.h" #include "paddle/fluid/framework/details/ssa_graph_executor.h"
...@@ -34,11 +35,10 @@ namespace details { ...@@ -34,11 +35,10 @@ namespace details {
class ThreadedSSAGraphExecutor : public SSAGraphExecutor { class ThreadedSSAGraphExecutor : public SSAGraphExecutor {
public: public:
ThreadedSSAGraphExecutor(size_t num_threads, bool use_event, ThreadedSSAGraphExecutor(const ExecutionStrategy &strategy,
const std::vector<Scope *> &local_scopes, const std::vector<Scope *> &local_scopes,
const std::vector<platform::Place> &places, const std::vector<platform::Place> &places,
std::unique_ptr<SSAGraph> &&graph, std::unique_ptr<SSAGraph> &&graph);
bool allow_op_delay);
// Run a SSAGraph by a thread pool // Run a SSAGraph by a thread pool
// Use topological sort algorithm // Use topological sort algorithm
...@@ -55,10 +55,8 @@ class ThreadedSSAGraphExecutor : public SSAGraphExecutor { ...@@ -55,10 +55,8 @@ class ThreadedSSAGraphExecutor : public SSAGraphExecutor {
std::vector<Scope *> local_scopes_; std::vector<Scope *> local_scopes_;
std::vector<platform::Place> places_; std::vector<platform::Place> places_;
platform::DeviceContextPool fetch_ctxs_; platform::DeviceContextPool fetch_ctxs_;
const bool use_event_;
std::unique_ptr<platform::EnforceNotMet> exception_; std::unique_ptr<platform::EnforceNotMet> exception_;
std::atomic<int> running_ops_; std::atomic<int> running_ops_;
bool allow_op_delay_;
void InsertPendingOp(std::unordered_map<OpHandleBase *, size_t> *pending_ops, void InsertPendingOp(std::unordered_map<OpHandleBase *, size_t> *pending_ops,
OpHandleBase *op_instance) const; OpHandleBase *op_instance) const;
...@@ -74,6 +72,9 @@ class ThreadedSSAGraphExecutor : public SSAGraphExecutor { ...@@ -74,6 +72,9 @@ class ThreadedSSAGraphExecutor : public SSAGraphExecutor {
std::unordered_map<OpHandleBase *, size_t> *pending_ops, std::unordered_map<OpHandleBase *, size_t> *pending_ops,
std::unordered_set<VarHandleBase *> *pending_vars, std::unordered_set<VarHandleBase *> *pending_vars,
BlockingQueue<VarHandleBase *> *ready_vars, FeedFetchList *fetch_data); BlockingQueue<VarHandleBase *> *ready_vars, FeedFetchList *fetch_data);
private:
ExecutionStrategy strategy_;
}; };
} // namespace details } // namespace details
......
...@@ -101,6 +101,8 @@ message VarType { ...@@ -101,6 +101,8 @@ message VarType {
FP16 = 4; FP16 = 4;
FP32 = 5; FP32 = 5;
FP64 = 6; FP64 = 6;
// Tensor<size_t> is used in C++.
SIZE_T = 19;
// Other types that may need additional descriptions // Other types that may need additional descriptions
LOD_TENSOR = 7; LOD_TENSOR = 7;
......
...@@ -27,7 +27,7 @@ TEST(OpKernelType, ToString) { ...@@ -27,7 +27,7 @@ TEST(OpKernelType, ToString) {
LibraryType::kCUDNN); LibraryType::kCUDNN);
ASSERT_EQ(paddle::framework::KernelTypeToString(op_kernel_type), ASSERT_EQ(paddle::framework::KernelTypeToString(op_kernel_type),
"data_type[float32]:data_layout[NCHW]:place[CPUPlace]:library_type[" "data_type[float]:data_layout[NCHW]:place[CPUPlace]:library_type["
"CUDNN]"); "CUDNN]");
} }
......
...@@ -52,13 +52,12 @@ std::vector<Scope *> &ParallelExecutor::GetLocalScopes() { ...@@ -52,13 +52,12 @@ std::vector<Scope *> &ParallelExecutor::GetLocalScopes() {
} }
ParallelExecutor::ParallelExecutor( ParallelExecutor::ParallelExecutor(
size_t num_threads, bool use_event,
const std::vector<platform::Place> &places, const std::vector<platform::Place> &places,
const std::unordered_set<std::string> &params, const std::unordered_set<std::string> &params,
const std::unordered_set<std::string> &bcast_vars, const std::unordered_set<std::string> &bcast_vars,
const ProgramDesc &main_program, const std::string &loss_var_name, const ProgramDesc &main_program, const std::string &loss_var_name,
Scope *scope, const std::vector<Scope *> &local_scopes, bool allow_op_delay, Scope *scope, const std::vector<Scope *> &local_scopes,
bool use_default_grad_scale, bool balance_parameter_opt_between_cards, const ExecutionStrategy &exec_strategy, const BuildStrategy &build_strategy,
size_t num_trainers, size_t trainer_id) size_t num_trainers, size_t trainer_id)
: member_(new ParallelExecutorPrivate(places)) { : member_(new ParallelExecutorPrivate(places)) {
member_->global_scope_ = scope; member_->global_scope_ = scope;
...@@ -100,18 +99,16 @@ ParallelExecutor::ParallelExecutor( ...@@ -100,18 +99,16 @@ ParallelExecutor::ParallelExecutor(
#ifdef PADDLE_WITH_CUDA #ifdef PADDLE_WITH_CUDA
details::MultiDevSSAGraphBuilder builder( details::MultiDevSSAGraphBuilder builder(
member_->places_, loss_var_name, params, member_->local_scopes_, member_->places_, loss_var_name, params, member_->local_scopes_,
member_->nccl_ctxs_.get(), use_default_grad_scale, member_->nccl_ctxs_.get(), build_strategy);
balance_parameter_opt_between_cards);
#else #else
details::MultiDevSSAGraphBuilder builder( details::MultiDevSSAGraphBuilder builder(member_->places_, loss_var_name,
member_->places_, loss_var_name, params, member_->local_scopes_, params, member_->local_scopes_,
use_default_grad_scale, balance_parameter_opt_between_cards); build_strategy);
#endif #endif
auto graph = builder.Build(main_program); auto graph = builder.Build(main_program);
member_->executor_.reset(new details::ThreadedSSAGraphExecutor( member_->executor_.reset(new details::ThreadedSSAGraphExecutor(
num_threads, use_event, member_->local_scopes_, places, std::move(graph), exec_strategy, member_->local_scopes_, places, std::move(graph)));
allow_op_delay));
// Step 3. Create vars in each scope; // Step 3. Create vars in each scope;
for (auto *var : main_program.Block(0).AllVars()) { for (auto *var : main_program.Block(0).AllVars()) {
......
...@@ -14,57 +14,60 @@ limitations under the License. */ ...@@ -14,57 +14,60 @@ limitations under the License. */
#pragma once #pragma once
#include <paddle/fluid/framework/details/build_strategy.h>
#include <string> #include <string>
#include <unordered_set> #include <unordered_set>
#include <vector> #include <vector>
#include "paddle/fluid/framework/details/execution_strategy.h"
#include "paddle/fluid/framework/executor.h" #include "paddle/fluid/framework/executor.h"
#include "paddle/fluid/framework/op_info.h" #include "paddle/fluid/framework/op_info.h"
#include "paddle/fluid/framework/program_desc.h" #include "paddle/fluid/framework/program_desc.h"
#include "paddle/fluid/framework/scope.h" #include "paddle/fluid/framework/scope.h"
#include "paddle/fluid/framework/tensor.h" #include "paddle/fluid/framework/tensor.h"
#include "paddle/fluid/platform/device_context.h" #include "paddle/fluid/platform/device_context.h"
namespace paddle { namespace paddle {
namespace framework { namespace framework {
class ParallelExecutorPrivate; class ParallelExecutorPrivate;
using details::BuildStrategy;
using details::ExecutionStrategy;
class ParallelExecutor { class ParallelExecutor {
DISABLE_COPY_AND_ASSIGN(ParallelExecutor); DISABLE_COPY_AND_ASSIGN(ParallelExecutor);
public: public:
explicit ParallelExecutor(size_t num_threads, bool use_event, explicit ParallelExecutor(const std::vector<platform::Place> &places,
const std::vector<platform::Place>& places, const std::unordered_set<std::string> &params,
const std::unordered_set<std::string>& params, const std::unordered_set<std::string> &bcast_vars,
const std::unordered_set<std::string>& bcast_vars, const ProgramDesc &main_program,
const ProgramDesc& main_program, const std::string &loss_var_name, Scope *scope,
const std::string& loss_var_name, Scope* scope, const std::vector<Scope *> &local_scopes,
const std::vector<Scope*>& local_scopes, const ExecutionStrategy &exec_strategy,
bool allow_op_delay, bool use_default_grad_scale, const BuildStrategy &build_strategy,
bool balance_parameter_opt_between_cards,
size_t num_trainers = 1, size_t trainer_id = 0); size_t num_trainers = 1, size_t trainer_id = 0);
~ParallelExecutor(); ~ParallelExecutor();
std::vector<Scope*>& GetLocalScopes(); std::vector<Scope *> &GetLocalScopes();
/** /**
* Feed tensors to local scopes. The size of tensors should be equal to the * Feed tensors to local scopes. The size of tensors should be equal to the
* size of local scopes. * size of local scopes.
*/ */
void FeedTensorsIntoLocalScopes( void FeedTensorsIntoLocalScopes(
const std::vector<std::unordered_map<std::string, LoDTensor>>& tensors); const std::vector<std::unordered_map<std::string, LoDTensor>> &tensors);
void FeedAndSplitTensorIntoLocalScopes( void FeedAndSplitTensorIntoLocalScopes(
const std::unordered_map<std::string, LoDTensor>& tensors); const std::unordered_map<std::string, LoDTensor> &tensors);
void Run(const std::vector<std::string>& fetch_tensors, void Run(const std::vector<std::string> &fetch_tensors,
const std::string& fetched_var_name); const std::string &fetched_var_name);
void BCastParamsToGPUs(const std::unordered_set<std::string>& vars) const; void BCastParamsToGPUs(const std::unordered_set<std::string> &vars) const;
private: private:
ParallelExecutorPrivate* member_; ParallelExecutorPrivate *member_;
}; };
} // namespace framework } // namespace framework
......
...@@ -13,54 +13,14 @@ See the License for the specific language governing permissions and ...@@ -13,54 +13,14 @@ See the License for the specific language governing permissions and
limitations under the License. */ limitations under the License. */
#pragma once #pragma once
#include "paddle/fluid/framework/data_type.h"
#include "paddle/fluid/memory/memcpy.h" #include "paddle/fluid/memory/memcpy.h"
#include "paddle/fluid/platform/enforce.h" #include "paddle/fluid/platform/enforce.h"
#include "paddle/fluid/platform/float16.h" #include "paddle/fluid/platform/float16.h"
namespace paddle { namespace paddle {
namespace framework { namespace framework {
extern size_t SizeOfType(std::type_index type);
template <typename... T>
struct SizeOfTypeFunctor;
template <typename T>
struct SizeOfTypeFunctor<T> {
size_t operator()(std::type_index type) const {
if (typeid(T).hash_code() == type.hash_code()) {
return sizeof(T);
} else {
return 0UL;
}
}
};
template <>
struct SizeOfTypeFunctor<> {
size_t operator()(std::type_index type) const { return 0UL; }
};
template <typename HEAD, typename... TAIL>
struct SizeOfTypeFunctor<HEAD, TAIL...> {
size_t operator()(std::type_index type) const {
SizeOfTypeFunctor<HEAD> head;
size_t head_size = head(type);
if (head_size != 0) {
return head_size;
}
SizeOfTypeFunctor<TAIL...> tail;
return tail(type);
}
};
static inline size_t SizeOfType(std::type_index type) {
SizeOfTypeFunctor<int, float, double, int16_t, int64_t, bool, size_t,
platform::float16>
functor;
size_t size = functor(type);
PADDLE_ENFORCE(size != 0UL, "Cannot get size of type %s", type.name());
return size;
}
inline void Tensor::check_memory_size() const { inline void Tensor::check_memory_size() const {
PADDLE_ENFORCE_NOT_NULL( PADDLE_ENFORCE_NOT_NULL(
holder_, "Tensor holds no memory. Call Tensor::mutable_data first."); holder_, "Tensor holds no memory. Call Tensor::mutable_data first.");
......
...@@ -11,6 +11,7 @@ distributed under the License is distributed on an "AS IS" BASIS, ...@@ -11,6 +11,7 @@ distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and See the License for the specific language governing permissions and
limitations under the License. */ limitations under the License. */
#pragma once
namespace paddle { namespace paddle {
namespace inference { namespace inference {
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#include <glog/logging.h> #include <glog/logging.h>
#include <sstream> #include <sstream>
#include <string>
#include <unordered_map> #include <unordered_map>
#include <vector> #include <vector>
......
...@@ -19,6 +19,7 @@ limitations under the License. */ ...@@ -19,6 +19,7 @@ limitations under the License. */
*/ */
#pragma once #pragma once
#include <limits>
#include <memory> #include <memory>
#include <string> #include <string>
#include <unordered_map> #include <unordered_map>
......
...@@ -14,6 +14,7 @@ limitations under the License. */ ...@@ -14,6 +14,7 @@ limitations under the License. */
#pragma once #pragma once
#include <string>
#include "paddle/fluid/framework/framework.pb.h" #include "paddle/fluid/framework/framework.pb.h"
namespace paddle { namespace paddle {
...@@ -58,8 +59,8 @@ class EngineBase { ...@@ -58,8 +59,8 @@ class EngineBase {
struct Buffer { struct Buffer {
void* buffer{nullptr}; // buffer should be allocated only once. void* buffer{nullptr}; // buffer should be allocated only once.
int max_size; // buffer allocated space. size_t max_size; // buffer allocated space.
int size; // data size. size_t size; // data size.
DeviceType device{DeviceType::UNK}; // tells which device this buffer is on. DeviceType device{DeviceType::UNK}; // tells which device this buffer is on.
}; };
......
nv_library(tensorrt_engine SRCS engine.cc DEPS framework_proto) nv_library(tensorrt_engine SRCS engine.cc DEPS framework_proto)
nv_test(test_tensorrt SRCS test_tensorrt.cc DEPS dynload_cuda device_context dynamic_loader) nv_test(test_tensorrt SRCS test_tensorrt.cc DEPS dynload_cuda device_context dynamic_loader)
nv_test(test_tensorrt_engine SRCS test_engine.cc DEPS dynload_cuda tensorrt_engine) nv_test(test_tensorrt_engine SRCS test_engine.cc DEPS dynload_cuda tensorrt_engine)
add_subdirectory(convert) add_subdirectory(convert)
nv_test(test_op_converter SRCS test_op_converter.cc mul_op.cc conv2d_op.cc op_converter.h DEPS ${FLUID_CORE_MODULES}) nv_test(test_op_converter SRCS test_op_converter.cc mul_op.cc conv2d_op.cc DEPS ${FLUID_CORE_MODULES})
nv_test(test_trt_activation_op SRCS test_activation_op.cc activation_op.cc nv_test(test_trt_activation_op SRCS test_activation_op.cc activation_op.cc io_converter.cc
DEPS ${FLUID_CORE_MODULES} activation_op tensorrt_engine) DEPS ${FLUID_CORE_MODULES} activation_op tensorrt_engine)
nv_test(test_io_converter SRCS test_io_converter.cc io_converter.cc DEPS dynload_cuda dynamic_loader lod_tensor) nv_test(test_io_converter SRCS test_io_converter.cc io_converter.cc DEPS dynload_cuda dynamic_loader lod_tensor)
...@@ -23,26 +23,42 @@ namespace tensorrt { ...@@ -23,26 +23,42 @@ namespace tensorrt {
using platform::is_gpu_place; using platform::is_gpu_place;
using platform::is_cpu_place; using platform::is_cpu_place;
class DefaultInputConverter : public EngineInputConverter { class DefaultIOConverter : public EngineIOConverter {
public: public:
DefaultInputConverter() {} DefaultIOConverter() {}
// NOTE out is GPU memory. // NOTE out is GPU memory.
virtual void operator()(const LoDTensor& in, void* out, virtual void operator()(const LoDTensor& in, void* out,
size_t max_size) override { size_t max_size) override {
PADDLE_ENFORCE(out != nullptr); PADDLE_ENFORCE(out != nullptr);
PADDLE_ENFORCE_LE(in.memory_size(), max_size); PADDLE_ENFORCE(stream_ != nullptr);
const auto& place = in.place(); const auto& place = in.place();
size_t size = in.memory_size();
PADDLE_ENFORCE_LE(size, max_size);
if (is_cpu_place(place)) { if (is_cpu_place(place)) {
PADDLE_ENFORCE(stream_ != nullptr); PADDLE_ENFORCE_EQ(0, cudaMemcpyAsync(out, in.data<float>(), size,
PADDLE_ENFORCE_EQ(0, cudaMemcpyHostToDevice, *stream_));
cudaMemcpyAsync(out, in.data<float>(), in.memory_size(),
cudaMemcpyHostToDevice, *stream_));
} else if (is_gpu_place(place)) { } else if (is_gpu_place(place)) {
PADDLE_ENFORCE_EQ(0, PADDLE_ENFORCE_EQ(0, cudaMemcpyAsync(out, in.data<float>(), size,
cudaMemcpyAsync(out, in.data<float>(), in.memory_size(), cudaMemcpyDeviceToDevice, *stream_));
cudaMemcpyHostToHost, *stream_)); } else {
PADDLE_THROW("Unknown device for converter");
}
cudaStreamSynchronize(*stream_);
}
// NOTE in is GPU memory.
virtual void operator()(const void* in, LoDTensor* out,
size_t max_size) override {
PADDLE_ENFORCE(in != nullptr);
PADDLE_ENFORCE(stream_ != nullptr);
const auto& place = out->place();
size_t size = out->memory_size();
PADDLE_ENFORCE_LE(size, max_size);
if (is_cpu_place(place)) {
PADDLE_ENFORCE_EQ(0, cudaMemcpyAsync(out->data<float>(), in, size,
cudaMemcpyDeviceToHost, *stream_));
} else if (is_gpu_place(place)) {
PADDLE_ENFORCE_EQ(0, cudaMemcpyAsync(out->data<float>(), in, size,
cudaMemcpyDeviceToDevice, *stream_));
} else { } else {
PADDLE_THROW("Unknown device for converter"); PADDLE_THROW("Unknown device for converter");
} }
...@@ -50,7 +66,8 @@ class DefaultInputConverter : public EngineInputConverter { ...@@ -50,7 +66,8 @@ class DefaultInputConverter : public EngineInputConverter {
} }
}; };
REGISTER_TENSORRT_INPUT_CONVERTER(default, DefaultInputConverter); // fluid LodTensor <-> tensorrt ITensor
REGISTER_TENSORRT_IO_CONVERTER(default, DefaultIOConverter);
} // namespace tensorrt } // namespace tensorrt
} // namespace inference } // namespace inference
......
...@@ -14,6 +14,7 @@ limitations under the License. */ ...@@ -14,6 +14,7 @@ limitations under the License. */
#pragma once #pragma once
#include <string>
#include <unordered_map> #include <unordered_map>
#include "paddle/fluid/framework/lod_tensor.h" #include "paddle/fluid/framework/lod_tensor.h"
#include "paddle/fluid/inference/utils/singleton.h" #include "paddle/fluid/inference/utils/singleton.h"
...@@ -25,43 +26,57 @@ namespace tensorrt { ...@@ -25,43 +26,57 @@ namespace tensorrt {
using framework::LoDTensor; using framework::LoDTensor;
/* /*
* Convert Input from Fluid to an Engine. * Convert Input from Fluid to TensorRT Engine.
* TensorRT's ITensor follows row major, NCHW. Fluid is also row major, so in * Convert Output from TensorRT Engine to Fluid.
* most cases just need to copy the data. *
* Note that TensorRT's ITensor follows row major, NCHW. Fluid is also row
* major,
* so in the default case just need to copy the data.
*/ */
class EngineInputConverter { class EngineIOConverter {
public: public:
EngineInputConverter() {} EngineIOConverter() {}
virtual void operator()(const LoDTensor& in, void* out, size_t max_size) {} virtual void operator()(const LoDTensor& in, void* out, size_t max_size) {}
virtual void operator()(const void* in, LoDTensor* out, size_t max_size) {}
void SetStream(cudaStream_t* stream) { stream_ = stream; } void SetStream(cudaStream_t* stream) { stream_ = stream; }
static void Run(const std::string& in_op_type, const LoDTensor& in, void* out, static void ConvertInput(const std::string& op_type, const LoDTensor& in,
size_t max_size, cudaStream_t* stream) { void* out, size_t max_size, cudaStream_t* stream) {
PADDLE_ENFORCE(stream != nullptr); PADDLE_ENFORCE(stream != nullptr);
auto* converter = Registry<EngineInputConverter>::Lookup( auto* converter = Registry<EngineIOConverter>::Lookup(
in_op_type, "default" /* default_type */); op_type, "default" /* default_type */);
PADDLE_ENFORCE_NOT_NULL(converter); PADDLE_ENFORCE_NOT_NULL(converter);
converter->SetStream(stream); converter->SetStream(stream);
(*converter)(in, out, max_size); (*converter)(in, out, max_size);
} }
virtual ~EngineInputConverter() {} static void ConvertOutput(const std::string& op_type, const void* in,
LoDTensor* out, size_t max_size,
cudaStream_t* stream) {
PADDLE_ENFORCE(stream != nullptr);
auto* converter = Registry<EngineIOConverter>::Lookup(
op_type, "default" /* default_type */);
PADDLE_ENFORCE_NOT_NULL(converter);
converter->SetStream(stream);
(*converter)(in, out, max_size);
}
virtual ~EngineIOConverter() {}
protected: protected:
cudaStream_t* stream_{nullptr}; cudaStream_t* stream_{nullptr};
}; };
#define REGISTER_TENSORRT_IO_CONVERTER(op_type__, Converter__) \
struct trt_io_##op_type__##_converter { \
trt_io_##op_type__##_converter() { \
Registry<EngineIOConverter>::Register<Converter__>(#op_type__); \
} \
}; \
trt_io_##op_type__##_converter trt_io_##op_type__##_converter__;
} // namespace tensorrt } // namespace tensorrt
} // namespace inference } // namespace inference
} // namespace paddle } // namespace paddle
#define REGISTER_TENSORRT_INPUT_CONVERTER(in_op_type__, Converter__) \
struct trt_input_##in_op_type__##_converter { \
trt_input_##in_op_type__##_converter() { \
::paddle::inference::Registry<EngineInputConverter>::Register< \
Converter__>(#in_op_type__); \
} \
}; \
trt_input_##in_op_type__##_converter trt_input_##in_op_type__##_converter__;
...@@ -16,6 +16,7 @@ limitations under the License. */ ...@@ -16,6 +16,7 @@ limitations under the License. */
#include "paddle/fluid/framework/lod_tensor.h" #include "paddle/fluid/framework/lod_tensor.h"
#include "paddle/fluid/framework/op_registry.h" #include "paddle/fluid/framework/op_registry.h"
#include "paddle/fluid/framework/program_desc.h" #include "paddle/fluid/framework/program_desc.h"
#include "paddle/fluid/inference/tensorrt/convert/io_converter.h"
#include "paddle/fluid/inference/tensorrt/convert/op_converter.h" #include "paddle/fluid/inference/tensorrt/convert/op_converter.h"
#include "paddle/fluid/platform/device_context.h" #include "paddle/fluid/platform/device_context.h"
#include "paddle/fluid/platform/place.h" #include "paddle/fluid/platform/place.h"
...@@ -26,7 +27,7 @@ namespace paddle { ...@@ -26,7 +27,7 @@ namespace paddle {
namespace inference { namespace inference {
namespace tensorrt { namespace tensorrt {
void Compare(float input, float expect) { void Compare(const std::string op_type, float input, float expect) {
framework::Scope scope; framework::Scope scope;
platform::CUDAPlace place; platform::CUDAPlace place;
platform::CUDADeviceContext ctx(place); platform::CUDADeviceContext ctx(place);
...@@ -35,6 +36,7 @@ void Compare(float input, float expect) { ...@@ -35,6 +36,7 @@ void Compare(float input, float expect) {
auto x_var = scope.Var("X"); auto x_var = scope.Var("X");
auto x_tensor = x_var->GetMutable<framework::LoDTensor>(); auto x_tensor = x_var->GetMutable<framework::LoDTensor>();
x_tensor->Resize({1, 1}); x_tensor->Resize({1, 1});
x_tensor->mutable_data<float>(place);
std::vector<float> init; std::vector<float> init;
init.push_back(input); init.push_back(input);
framework::TensorFromVector(init, ctx, x_tensor); framework::TensorFromVector(init, ctx, x_tensor);
...@@ -45,14 +47,15 @@ void Compare(float input, float expect) { ...@@ -45,14 +47,15 @@ void Compare(float input, float expect) {
out_tensor->mutable_data<float>(place); out_tensor->mutable_data<float>(place);
framework::OpDesc op_desc; framework::OpDesc op_desc;
op_desc.SetType("relu"); op_desc.SetType(op_type);
op_desc.SetInput("X", {"X"}); op_desc.SetInput("X", {"X"});
op_desc.SetOutput("Out", {"Out"}); op_desc.SetOutput("Out", {"Out"});
auto relu_op = framework::OpRegistry::CreateOp(*op_desc.Proto()); auto op = framework::OpRegistry::CreateOp(*op_desc.Proto());
// run fluid op // run fluid op
relu_op->Run(scope, place); op->Run(scope, place);
// get fluid output
std::vector<float> out1; std::vector<float> out1;
framework::TensorToVector(*out_tensor, ctx, &out1); framework::TensorToVector(*out_tensor, ctx, &out1);
...@@ -63,21 +66,28 @@ void Compare(float input, float expect) { ...@@ -63,21 +66,28 @@ void Compare(float input, float expect) {
engine->InitNetwork(); engine->InitNetwork();
engine->DeclareInput("X", nvinfer1::DataType::kFLOAT, engine->DeclareInput("X", nvinfer1::DataType::kFLOAT,
nvinfer1::DimsCHW{1, 1, 1}); nvinfer1::DimsCHW{1, 1, 1});
// convert op
OpConverter op_converter; OpConverter op_converter;
op_converter.ConvertOp(*op_desc.Proto(), engine); op_converter.ConvertOp(*op_desc.Proto(), engine);
engine->DeclareOutput("Out"); engine->DeclareOutput("Out");
engine->FreezeNetwork(); engine->FreezeNetwork();
engine->SetInputFromCPU("X", &input, 1 * sizeof(float));
// run tensorrt op // convert LoDTensor to ITensor
size_t size = x_tensor->memory_size();
EngineIOConverter::ConvertInput(op_type, *x_tensor,
engine->buffer("X").buffer, size, &stream);
// run tensorrt Outp
engine->Execute(1); engine->Execute(1);
// convert ITensor to LoDTensor
float out2; EngineIOConverter::ConvertOutput(op_type, engine->buffer("Out").buffer,
engine->GetOutputInCPU("Out", &out2, 1 * sizeof(float)); out_tensor, size, &stream);
// get tensorrt output
ASSERT_EQ(out1[0], out2); std::vector<float> out2;
framework::TensorToVector(*out_tensor, ctx, &out2);
// compare
ASSERT_EQ(out1[0], out2[0]);
ASSERT_EQ(out1[0], expect); ASSERT_EQ(out1[0], expect);
delete engine; delete engine;
...@@ -85,8 +95,8 @@ void Compare(float input, float expect) { ...@@ -85,8 +95,8 @@ void Compare(float input, float expect) {
} }
TEST(OpConverter, ConvertRelu) { TEST(OpConverter, ConvertRelu) {
Compare(1, 1); // relu(1) = 1 Compare("relu", 1, 1); // relu(1) = 1
Compare(-5, 0); // relu(-5) = 0 Compare("relu", -5, 0); // relu(-5) = 0
} }
} // namespace tensorrt } // namespace tensorrt
......
...@@ -12,40 +12,63 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. ...@@ -12,40 +12,63 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and See the License for the specific language governing permissions and
limitations under the License. */ limitations under the License. */
#include <gtest/gtest.h>
#include "paddle/fluid/framework/lod_tensor.h" #include "paddle/fluid/framework/lod_tensor.h"
#include "paddle/fluid/inference/tensorrt/convert/io_converter.h" #include "paddle/fluid/inference/tensorrt/convert/io_converter.h"
#include <gtest/gtest.h>
namespace paddle { namespace paddle {
namespace inference { namespace inference {
namespace tensorrt { namespace tensorrt {
class EngineInputConverterTester : public ::testing::Test { void IOConverterTester(const platform::DeviceContext& ctx) {
public: cudaStream_t stream;
void SetUp() override { tensor.Resize({10, 10}); } ASSERT_EQ(0, cudaStreamCreate(&stream));
framework::LoDTensor tensor; // init fluid in_tensor
}; framework::LoDTensor in_tensor;
in_tensor.Resize({10, 10});
auto place = ctx.GetPlace();
in_tensor.mutable_data<float>(place);
std::vector<float> init;
for (int64_t i = 0; i < 10 * 10; ++i) {
init.push_back(i);
}
framework::TensorFromVector(init, ctx, &in_tensor);
TEST_F(EngineInputConverterTester, DefaultCPU) { // init tensorrt buffer
void* buffer; void* buffer;
tensor.mutable_data<float>(platform::CPUPlace()); size_t size = in_tensor.memory_size();
ASSERT_EQ(cudaMalloc(&buffer, tensor.memory_size()), 0); ASSERT_EQ(cudaMalloc(&buffer, size), 0);
cudaStream_t stream; // convert fluid in_tensor to tensorrt buffer
EngineInputConverter::Run("test", tensor, buffer, tensor.memory_size(), EngineIOConverter::ConvertInput("test", in_tensor, buffer, size, &stream);
&stream);
// convert tensorrt buffer to fluid out_tensor
framework::LoDTensor out_tensor;
out_tensor.Resize({10, 10});
out_tensor.mutable_data<float>(place);
EngineIOConverter::ConvertOutput("test", buffer, &out_tensor, size, &stream);
// compare in_tensor and out_tensor
std::vector<float> result;
framework::TensorToVector(out_tensor, ctx, &result);
EXPECT_EQ(init.size(), result.size());
for (size_t i = 0; i < init.size(); i++) {
EXPECT_EQ(init[i], result[i]);
}
cudaStreamDestroy(stream);
} }
TEST_F(EngineInputConverterTester, DefaultGPU) { TEST(EngineIOConverterTester, DefaultCPU) {
void* buffer; platform::CPUPlace place;
tensor.mutable_data<float>(platform::CUDAPlace()); platform::CPUDeviceContext ctx(place);
ASSERT_EQ(cudaMalloc(&buffer, tensor.memory_size()), 0); IOConverterTester(ctx);
}
cudaStream_t stream; TEST(EngineIOConverterTester, DefaultGPU) {
EngineInputConverter::Run("test", tensor, buffer, tensor.memory_size(), platform::CUDAPlace place;
&stream); platform::CUDADeviceContext ctx(place);
IOConverterTester(ctx);
} }
} // namespace tensorrt } // namespace tensorrt
......
...@@ -306,7 +306,7 @@ void AsyncGRPCServer::TryToRegisterNewPrefetchOne() { ...@@ -306,7 +306,7 @@ void AsyncGRPCServer::TryToRegisterNewPrefetchOne() {
} }
RequestPrefetch* prefetch = RequestPrefetch* prefetch =
new RequestPrefetch(&service_, cq_prefetch_.get(), sync_mode_, scope_, new RequestPrefetch(&service_, cq_prefetch_.get(), sync_mode_, scope_,
dev_ctx_, executor_, program_, prefetch_ctx_); dev_ctx_, executor_, program_, prefetch_ctx_.get());
VLOG(4) << "Create RequestPrefetch status:" << prefetch->Status(); VLOG(4) << "Create RequestPrefetch status:" << prefetch->Status();
} }
......
...@@ -64,8 +64,9 @@ class AsyncGRPCServer final { ...@@ -64,8 +64,9 @@ class AsyncGRPCServer final {
void SetExecutor(framework::Executor *executor) { executor_ = executor; } void SetExecutor(framework::Executor *executor) { executor_ = executor; }
void SetPrefetchPreparedCtx(framework::ExecutorPrepareContext *prepared) { void SetPrefetchPreparedCtx(
prefetch_ctx_ = prepared; std::unique_ptr<framework::ExecutorPrepareContext> prepared) {
prefetch_ctx_.reset(prepared.release());
} }
int GetSelectedPort() const { return selected_port_; } int GetSelectedPort() const { return selected_port_; }
...@@ -116,7 +117,7 @@ class AsyncGRPCServer final { ...@@ -116,7 +117,7 @@ class AsyncGRPCServer final {
std::unique_ptr<std::thread> t_get_; std::unique_ptr<std::thread> t_get_;
std::unique_ptr<std::thread> t_prefetch_; std::unique_ptr<std::thread> t_prefetch_;
framework::ExecutorPrepareContext *prefetch_ctx_; std::unique_ptr<framework::ExecutorPrepareContext> prefetch_ctx_;
framework::ProgramDesc *program_; framework::ProgramDesc *program_;
framework::Executor *executor_; framework::Executor *executor_;
int selected_port_; int selected_port_;
......
...@@ -100,7 +100,7 @@ void StartServer(const std::string& endpoint) { ...@@ -100,7 +100,7 @@ void StartServer(const std::string& endpoint) {
InitTensorsOnServer(&scope, &place, 10); InitTensorsOnServer(&scope, &place, 10);
rpc_service_->SetProgram(&program); rpc_service_->SetProgram(&program);
rpc_service_->SetPrefetchPreparedCtx(prepared.get()); rpc_service_->SetPrefetchPreparedCtx(std::move(prepared));
rpc_service_->SetDevCtx(&ctx); rpc_service_->SetDevCtx(&ctx);
rpc_service_->SetScope(&scope); rpc_service_->SetScope(&scope);
rpc_service_->SetExecutor(&exe); rpc_service_->SetExecutor(&exe);
......
...@@ -322,8 +322,7 @@ void ListenAndServOp::RunImpl(const framework::Scope &scope, ...@@ -322,8 +322,7 @@ void ListenAndServOp::RunImpl(const framework::Scope &scope,
// prepare for prefetch // prepare for prefetch
VLOG(3) << "prefetch block id is " << prefetch_block->ID(); VLOG(3) << "prefetch block id is " << prefetch_block->ID();
auto prefetch_prepared = executor.Prepare(*program, prefetch_block->ID()); auto prefetch_prepared = executor.Prepare(*program, prefetch_block->ID());
rpc_service_->SetPrefetchPreparedCtx(prefetch_prepared.get()); rpc_service_->SetPrefetchPreparedCtx(std::move(prefetch_prepared));
prefetch_prepared.release();
// start the server listening after all member initialized. // start the server listening after all member initialized.
server_thread_.reset(new std::thread(RunServer, rpc_service_)); server_thread_.reset(new std::thread(RunServer, rpc_service_));
......
...@@ -12,7 +12,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. ...@@ -12,7 +12,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and See the License for the specific language governing permissions and
limitations under the License. */ limitations under the License. */
#include <fstream> #include <fstream>
#include "paddle/fluid/framework/data_type_transform.h"
#include "paddle/fluid/framework/op_registry.h" #include "paddle/fluid/framework/op_registry.h"
#include "paddle/fluid/platform/device_context.h" #include "paddle/fluid/platform/device_context.h"
...@@ -31,6 +31,7 @@ class LoadCombineOp : public framework::OperatorBase { ...@@ -31,6 +31,7 @@ class LoadCombineOp : public framework::OperatorBase {
void RunImpl(const framework::Scope &scope, void RunImpl(const framework::Scope &scope,
const platform::Place &place) const override { const platform::Place &place) const override {
auto filename = Attr<std::string>("file_path"); auto filename = Attr<std::string>("file_path");
auto load_as_fp16 = Attr<bool>("load_as_fp16");
std::ifstream fin(filename); std::ifstream fin(filename);
PADDLE_ENFORCE(static_cast<bool>(fin), PADDLE_ENFORCE(static_cast<bool>(fin),
...@@ -59,17 +60,25 @@ class LoadCombineOp : public framework::OperatorBase { ...@@ -59,17 +60,25 @@ class LoadCombineOp : public framework::OperatorBase {
// Get data from fin to tensor // Get data from fin to tensor
DeserializeFromStream(fin, tensor, dev_ctx); DeserializeFromStream(fin, tensor, dev_ctx);
if (platform::is_gpu_place(place)) { auto in_dtype = framework::ToDataType(tensor->type());
// copy CPU to GPU auto out_dtype =
framework::LoDTensor cpu_tensor; load_as_fp16 ? framework::proto::VarType::FP16 : in_dtype;
cpu_tensor.ShareDataWith(*tensor);
cpu_tensor.set_lod(tensor->lod()); if (in_dtype != out_dtype) {
// convert to float16 tensor
// reset tensor auto in_kernel_type = framework::OpKernelType(in_dtype, place);
auto out_kernel_type = framework::OpKernelType(out_dtype, place);
framework::LoDTensor fp16_tensor;
// copy LoD info to the new tensor
fp16_tensor.set_lod(tensor->lod());
framework::TransDataType(in_kernel_type, out_kernel_type, *tensor,
&fp16_tensor);
// reset output tensor
out_var->Clear(); out_var->Clear();
tensor = out_var->GetMutable<framework::LoDTensor>(); tensor = out_var->GetMutable<framework::LoDTensor>();
tensor->set_lod(cpu_tensor.lod()); tensor->set_lod(fp16_tensor.lod());
TensorCopy(cpu_tensor, place, dev_ctx, tensor); tensor->ShareDataWith(fp16_tensor);
} }
} }
} }
...@@ -82,6 +91,13 @@ class LoadCombineOpProtoMaker : public framework::OpProtoAndCheckerMaker { ...@@ -82,6 +91,13 @@ class LoadCombineOpProtoMaker : public framework::OpProtoAndCheckerMaker {
"Out", "Out",
"(vector) The output LoDTensors that will be read from the input file.") "(vector) The output LoDTensors that will be read from the input file.")
.AsDuplicable(); .AsDuplicable();
AddAttr<bool>(
"load_as_fp16",
"(boolean, default false)"
"If true, the tensor will be first loaded and then "
"converted to float16 data type. Otherwise, the tensor will be "
"directly loaded without data type conversion.")
.SetDefault(false);
AddAttr<std::string>("file_path", AddAttr<std::string>("file_path",
"(string) " "(string) "
"LoDTensors will be loaded from \"file_path\".") "LoDTensors will be loaded from \"file_path\".")
......
...@@ -139,8 +139,9 @@ TEST(SaveLoadCombineOp, CPU) { ...@@ -139,8 +139,9 @@ TEST(SaveLoadCombineOp, CPU) {
CheckValues<int, int>(expect4, actual4, expect_lod4, actual_lod4, numel4); CheckValues<int, int>(expect4, actual4, expect_lod4, actual_lod4, numel4);
} }
// FP16 version of SaveLoadCombineOp Test // FP16 version of SaveLoadCombineOp Test, only altering the saving aspect
TEST(SaveLoadCombineFP16Op, CPU) { // to save as FP16.
TEST(SaveCombineFP16Op, CPU) {
paddle::framework::Scope scope; paddle::framework::Scope scope;
paddle::platform::CPUPlace place; paddle::platform::CPUPlace place;
...@@ -169,7 +170,7 @@ TEST(SaveLoadCombineFP16Op, CPU) { ...@@ -169,7 +170,7 @@ TEST(SaveLoadCombineFP16Op, CPU) {
20, 50, lod4, "test_var4", place, &scope, &expect_lod4); 20, 50, lod4, "test_var4", place, &scope, &expect_lod4);
// Set attributes // Set attributes
std::string filename = "check_tensor_fp16.ls"; std::string filename = "check_tensor_fp16_save.ls";
paddle::framework::AttributeMap attrs; paddle::framework::AttributeMap attrs;
attrs.insert({"file_path", std::string(filename)}); attrs.insert({"file_path", std::string(filename)});
attrs.insert({"save_as_fp16", true}); attrs.insert({"save_as_fp16", true});
...@@ -216,6 +217,89 @@ TEST(SaveLoadCombineFP16Op, CPU) { ...@@ -216,6 +217,89 @@ TEST(SaveLoadCombineFP16Op, CPU) {
actual_lod4, numel4); actual_lod4, numel4);
} }
// FP16 version of SaveLoadCombineOp Test, only altering the loading aspect
// to load tensors with FP16 precision.
TEST(LoadCombineFP16Op, CPU) {
paddle::framework::Scope scope;
paddle::platform::CPUPlace place;
std::vector<int> lod1 = {0, 1, 2, 3, 10};
int numel1 = 100;
paddle::framework::LoD expect_lod1;
float* expect1 = CreateForSaveCombineOp<float, paddle::platform::float16>(
10, 10, lod1, "test_var1", place, &scope, &expect_lod1);
std::vector<int> lod2 = {0, 2, 5, 10};
int numel2 = 200;
paddle::framework::LoD expect_lod2;
float* expect2 = CreateForSaveCombineOp<float, paddle::platform::float16>(
10, 20, lod2, "test_var2", place, &scope, &expect_lod2);
std::vector<int> lod3 = {0, 20};
int numel3 = 4000;
paddle::framework::LoD expect_lod3;
float* expect3 = CreateForSaveCombineOp<float, paddle::platform::float16>(
20, 200, lod3, "test_var3", place, &scope, &expect_lod3);
std::vector<int> lod4 = {0, 1, 20};
int numel4 = 1000;
paddle::framework::LoD expect_lod4;
float* expect4 = CreateForSaveCombineOp<float, paddle::platform::float16>(
20, 50, lod4, "test_var4", place, &scope, &expect_lod4);
// Set attributes
std::string filename = "check_tensor_fp16_load.ls";
paddle::framework::AttributeMap attrs;
attrs.insert({"file_path", std::string(filename)});
// Run the save_combine_op
auto save_combine_op = paddle::framework::OpRegistry::CreateOp(
"save_combine",
{{"X", {"test_var1", "test_var2", "test_var3", "test_var4"}}}, {}, attrs);
save_combine_op->Run(scope, place);
// Set up output vars
auto load_var1 = scope.Var("out_var1");
auto load_var2 = scope.Var("out_var2");
auto load_var3 = scope.Var("out_var3");
auto load_var4 = scope.Var("out_var4");
attrs.insert({"load_as_fp16", true});
// Run the load_combine_op
auto load_combine_op = paddle::framework::OpRegistry::CreateOp(
"load_combine", {},
{{"Out", {"out_var1", "out_var2", "out_var3", "out_var4"}}}, attrs);
load_combine_op->Run(scope, place);
auto* target1 = load_var1->GetMutable<paddle::framework::LoDTensor>();
auto* target2 = load_var2->GetMutable<paddle::framework::LoDTensor>();
auto* target3 = load_var3->GetMutable<paddle::framework::LoDTensor>();
auto* target4 = load_var4->GetMutable<paddle::framework::LoDTensor>();
paddle::framework::LoD actual_lod1, actual_lod2, actual_lod3, actual_lod4;
paddle::platform::float16* actual1 =
GetValuesAfterLoadCombineOp<paddle::platform::float16>(target1, scope,
&actual_lod1);
paddle::platform::float16* actual2 =
GetValuesAfterLoadCombineOp<paddle::platform::float16>(target2, scope,
&actual_lod2);
paddle::platform::float16* actual3 =
GetValuesAfterLoadCombineOp<paddle::platform::float16>(target3, scope,
&actual_lod3);
paddle::platform::float16* actual4 =
GetValuesAfterLoadCombineOp<paddle::platform::float16>(target4, scope,
&actual_lod4);
CheckValues<float, paddle::platform::float16>(expect1, actual1, expect_lod1,
actual_lod1, numel1);
CheckValues<float, paddle::platform::float16>(expect2, actual2, expect_lod2,
actual_lod2, numel2);
CheckValues<float, paddle::platform::float16>(expect3, actual3, expect_lod3,
actual_lod3, numel3);
CheckValues<float, paddle::platform::float16>(expect4, actual4, expect_lod4,
actual_lod4, numel4);
}
// Test with original SaveLoadTest // Test with original SaveLoadTest
TEST(SaveLoadTestWithCombineOp, CPU) { TEST(SaveLoadTestWithCombineOp, CPU) {
paddle::framework::Scope scope; paddle::framework::Scope scope;
......
...@@ -53,7 +53,7 @@ class NCCLGroupGuard { ...@@ -53,7 +53,7 @@ class NCCLGroupGuard {
} }
inline ~NCCLGroupGuard() { inline ~NCCLGroupGuard() {
PADDLE_ENFORCE(dynload::ncclGroupEnd()); CHECK_EQ(dynload::ncclGroupEnd(), ncclSuccess);
NCCLMutex().unlock(); NCCLMutex().unlock();
} }
}; };
......
...@@ -494,23 +494,61 @@ All parameter, weight, gradient are variables in Paddle. ...@@ -494,23 +494,61 @@ All parameter, weight, gradient are variables in Paddle.
m.def("disable_profiler", platform::DisableProfiler); m.def("disable_profiler", platform::DisableProfiler);
m.def("reset_profiler", platform::ResetProfiler); m.def("reset_profiler", platform::ResetProfiler);
py::class_<ParallelExecutor>(m, "ParallelExecutor") // -- python binds for parallel executor.
.def("__init__", py::class_<ParallelExecutor> pe(m, "ParallelExecutor");
[](ParallelExecutor &self, size_t num_threads, bool use_event, py::class_<ExecutionStrategy>(pe, "ExecutionStrategy")
const std::vector<platform::Place> &places, .def(py::init())
const std::unordered_set<std::string> &params, .def_property(
const std::unordered_set<std::string> &bcast_vars, "num_threads",
const ProgramDesc &main_program, const std::string &loss_var_name, [](const ExecutionStrategy &self) { return self.num_threads_; },
Scope *scope, std::vector<Scope *> &local_scopes, [](ExecutionStrategy &self, size_t num_threads) {
bool allow_op_delay, bool use_default_grad_scale, self.num_threads_ = num_threads;
bool balance_parameter_opt_between_cards, size_t num_trainers, })
size_t trainer_id) { .def_property(
new (&self) ParallelExecutor( "use_event",
num_threads, use_event, places, params, bcast_vars, [](const ExecutionStrategy &self) { return self.use_event_; },
main_program, loss_var_name, scope, local_scopes, [](ExecutionStrategy &self, bool use_event) {
allow_op_delay, use_default_grad_scale, self.use_event_ = use_event;
balance_parameter_opt_between_cards, num_trainers, trainer_id); })
}) .def_property(
"allow_op_delay",
[](const ExecutionStrategy &self) { return self.allow_op_delay_; },
[](ExecutionStrategy &self, bool allow_op_delay) {
self.allow_op_delay_ = allow_op_delay;
});
py::class_<BuildStrategy> build_strategy(pe, "BuildStrategy");
py::enum_<BuildStrategy::ReduceStrategy>(build_strategy, "ReduceStrategy")
.value("Reduce", BuildStrategy::ReduceStrategy::kReduce)
.value("AllReduce", BuildStrategy::ReduceStrategy::kAllReduce);
py::enum_<BuildStrategy::GradientScaleStrategy>(build_strategy,
"GradientScaleStrategy")
.value("CoeffNumDevice",
BuildStrategy::GradientScaleStrategy::kCoeffNumDevice)
.value("One", BuildStrategy::GradientScaleStrategy::kOne)
.value("Customized", BuildStrategy::GradientScaleStrategy::kCustomized);
build_strategy.def(py::init())
.def_property(
"reduce_strategy",
[](const BuildStrategy &self) { return self.reduce_; },
[](BuildStrategy &self, BuildStrategy::ReduceStrategy strategy) {
self.reduce_ = strategy;
})
.def_property(
"gradient_scale_strategy",
[](const BuildStrategy &self) { return self.gradient_scale_; },
[](BuildStrategy &self,
BuildStrategy::GradientScaleStrategy strategy) {
self.gradient_scale_ = strategy;
});
pe.def(py::init<const std::vector<platform::Place> &,
const std::unordered_set<std::string> &,
const std::unordered_set<std::string> &, const ProgramDesc &,
const std::string &, Scope *, std::vector<Scope *> &,
const ExecutionStrategy &, const BuildStrategy &, size_t,
size_t>())
.def("bcast_params", &ParallelExecutor::BCastParamsToGPUs) .def("bcast_params", &ParallelExecutor::BCastParamsToGPUs)
// NOTE: even we return a vec<Scope*>* to Python use reference policy. // NOTE: even we return a vec<Scope*>* to Python use reference policy.
// We still cannot get local_scope from this vector, since the element // We still cannot get local_scope from this vector, since the element
......
...@@ -44,42 +44,44 @@ import transpiler ...@@ -44,42 +44,44 @@ import transpiler
from param_attr import ParamAttr, WeightNormParamAttr from param_attr import ParamAttr, WeightNormParamAttr
from data_feeder import DataFeeder from data_feeder import DataFeeder
from core import LoDTensor, CPUPlace, CUDAPlace, CUDAPinnedPlace from core import LoDTensor, CPUPlace, CUDAPlace, CUDAPinnedPlace
from transpiler import DistributeTranspiler, SimpleDistributeTranspiler, InferenceTranspiler, memory_optimize, release_memory from transpiler import DistributeTranspiler, SimpleDistributeTranspiler, \
InferenceTranspiler, memory_optimize, release_memory
from concurrency import (Go, make_channel, channel_send, channel_recv, from concurrency import (Go, make_channel, channel_send, channel_recv,
channel_close, Select) channel_close, Select)
import clip import clip
import profiler import profiler
import unique_name import unique_name
import recordio_writer import recordio_writer
from parallel_executor import ParallelExecutor import parallel_executor
from parallel_executor import *
Tensor = LoDTensor Tensor = LoDTensor
__all__ = framework.__all__ + executor.__all__ + concurrency.__all__ +\ __all__ = framework.__all__ + executor.__all__ + concurrency.__all__ + \
trainer.__all__ + inferencer.__all__ + transpiler.__all__ + [ trainer.__all__ + inferencer.__all__ + transpiler.__all__ + \
'io', parallel_executor.__all__ + [
'initializer', 'io',
'layers', 'initializer',
'transpiler' 'layers',
'nets', 'transpiler'
'optimizer', 'nets',
'learning_rate_decay', 'optimizer',
'backward', 'learning_rate_decay',
'regularizer', 'backward',
'LoDTensor', 'regularizer',
'CPUPlace', 'LoDTensor',
'CUDAPlace', 'CPUPlace',
'CUDAPinnedPlace', 'CUDAPlace',
'Tensor', 'CUDAPinnedPlace',
'ParamAttr', 'Tensor',
'WeightNormParamAttr', 'ParamAttr',
'DataFeeder', 'WeightNormParamAttr',
'clip', 'DataFeeder',
'profiler', 'clip',
'unique_name', 'profiler',
'recordio_writer', 'unique_name',
'ParallelExecutor', 'recordio_writer',
] ]
def __bootstrap__(): def __bootstrap__():
......
...@@ -480,6 +480,8 @@ def append_backward(loss, parameter_list=None, no_grad_set=None, ...@@ -480,6 +480,8 @@ def append_backward(loss, parameter_list=None, no_grad_set=None,
program.current_block_idx = current_block_idx program.current_block_idx = current_block_idx
program.sync_with_cpp() program.sync_with_cpp()
# FIXME(zcd): prevent loss.grad optimized by mem_opt.
loss.block.var(_append_grad_suffix_(loss.name)).persistable = True
if parameter_list is not None: if parameter_list is not None:
parameters = parameter_list parameters = parameter_list
......
...@@ -13,29 +13,35 @@ ...@@ -13,29 +13,35 @@
# limitations under the License. # limitations under the License.
import core import core
import framework
import executor import executor
import framework
import io import io
import unique_name
from trainer import check_and_get_place from trainer import check_and_get_place
__all__ = ['Inferencer', ] __all__ = ['Inferencer', ]
class Inferencer(object): class Inferencer(object):
def __init__(self, param_path, place=None): def __init__(self, infer_func, param_path, place=None):
""" """
:param param_path: the path where the inference model is saved by fluid.io.save_inference_model :param infer_func: a function that will return predict Variable
:param param_path: the path where the inference model is saved by fluid.io.save_params
:param place: place to do the inference :param place: place to do the inference
""" """
self.param_path = param_path self.param_path = param_path
self.scope = core.Scope() self.scope = core.Scope()
self.inference_program = framework.Program()
with framework.program_guard(self.inference_program):
with unique_name.guard():
self.predict_var = infer_func()
self.exe = executor.Executor(check_and_get_place(place)) self.exe = executor.Executor(check_and_get_place(place))
with executor.scope_guard(self.scope): with executor.scope_guard(self.scope):
# load params from param_path into scope # load params from param_path into scope
[self.inference_program, _, io.load_params(self.exe, param_path, self.inference_program)
self.fetch_targets] = io.load_inference_model(
executor=self.exe, dirname=param_path)
def infer(self, inputs, return_numpy=True): def infer(self, inputs, return_numpy=True):
""" """
...@@ -51,7 +57,7 @@ class Inferencer(object): ...@@ -51,7 +57,7 @@ class Inferencer(object):
with executor.scope_guard(self.scope): with executor.scope_guard(self.scope):
results = self.exe.run(self.inference_program, results = self.exe.run(self.inference_program,
feed=inputs, feed=inputs,
fetch_list=self.fetch_targets, fetch_list=[self.predict_var],
return_numpy=return_numpy) return_numpy=return_numpy)
return results return results
...@@ -19,7 +19,10 @@ import executor ...@@ -19,7 +19,10 @@ import executor
import warnings import warnings
import sys import sys
__all__ = ['ParallelExecutor'] __all__ = ['ParallelExecutor', 'ExecutionStrategy', 'BuildStrategy']
ExecutionStrategy = core.ParallelExecutor.ExecutionStrategy
BuildStrategy = core.ParallelExecutor.BuildStrategy
class ParallelExecutor(object): class ParallelExecutor(object):
...@@ -27,13 +30,12 @@ class ParallelExecutor(object): ...@@ -27,13 +30,12 @@ class ParallelExecutor(object):
use_cuda, use_cuda,
loss_name=None, loss_name=None,
main_program=None, main_program=None,
num_threads=None,
allow_op_delay=False,
share_vars_from=None, share_vars_from=None,
use_default_grad_scale=True, exec_strategy=None,
balance_parameter_opt_between_cards=False, build_strategy=None,
num_trainers=1, num_trainers=1,
trainer_id=0): trainer_id=0,
**kwargs):
""" """
ParallelExecutor can run program in parallel. ParallelExecutor can run program in parallel.
...@@ -42,21 +44,8 @@ class ParallelExecutor(object): ...@@ -42,21 +44,8 @@ class ParallelExecutor(object):
loss_name(str, default None): The loss name must set in training. loss_name(str, default None): The loss name must set in training.
main_program(Program, default None): The program that need to run, main_program(Program, default None): The program that need to run,
if not provided, then default_main_program will be used. if not provided, then default_main_program will be used.
num_threads(int, default None): How many threads are used for
training.
allow_op_delay(bool, default False): Whether to delay and buffer
some operators together for scheduling or not, which may
improve performance in some cases, default False.
share_vars_from(ParallelExecutor, default None): If provied, share_vars_from(ParallelExecutor, default None): If provied,
it will share variables from the specified ParallelExecutor. it will share variables from the specified ParallelExecutor.
use_default_grad_scale(bool, default True): If set True, a default
scale value equal to `1./device_count` would be multiplied to
gradients of each device and scaled gradients would be
aggregated. Otherwise, a customized scale value should be fed
to the network.
balance_parameter_opt_between_cards(bool, default True): Whether
updating different gradients on different cards. Currently, it
is not recommended.
num_trainers(int, default 1): If greater than 1, NCCL will be num_trainers(int, default 1): If greater than 1, NCCL will be
initialized with multpile rank of nodes, each node should have initialized with multpile rank of nodes, each node should have
same number of GPUs. Distributed training will be enabled then. same number of GPUs. Distributed training will be enabled then.
...@@ -83,6 +72,25 @@ class ParallelExecutor(object): ...@@ -83,6 +72,25 @@ class ParallelExecutor(object):
train_loss, = train_exe.run([loss.name], feed=feed_dict) train_loss, = train_exe.run([loss.name], feed=feed_dict)
test_loss, = test_exe.run([loss.name], feed=feed_dict) test_loss, = test_exe.run([loss.name], feed=feed_dict)
""" """
if len(kwargs) != 0:
err_msg = ""
for key in kwargs:
if key in dir(ExecutionStrategy):
err_msg += \
"Setting {0} by constructor is deprecated. Use " \
"strategy=ExecutionStrategy(); strategy.{0}=xxx; " \
"pe=ParallelExecutor(exec_strategy=strategy) " \
"instead.\n ".format(key)
elif key in dir(BuildStrategy):
err_msg += \
"Setting {0} by constructor is deprecated. Use " \
"strategy=BuildStrategy(); See help(" \
"paddle.fluid.ParallelExecutor.BuildStrategy) \n".format(
key)
else:
err_msg += "Setting {0} by constructor is deprecated. Use strategy.\n".format(
key)
raise ValueError(err_msg)
self._places = [] self._places = []
self._act_places = [] self._act_places = []
...@@ -100,15 +108,25 @@ class ParallelExecutor(object): ...@@ -100,15 +108,25 @@ class ParallelExecutor(object):
self._places.append(p) self._places.append(p)
assert self._places, "no place for execution" assert self._places, "no place for execution"
if num_threads is None: if exec_strategy is None:
exec_strategy = ExecutionStrategy()
if use_cuda:
exec_strategy.use_event = True
else:
exec_strategy.use_event = False
if exec_strategy.num_threads == 0:
if use_cuda: if use_cuda:
# Experiments on se-resnext shows that too many threads hurt # Experiments on se-resnext shows that too many threads hurt
# performance. Worth tunning for other models in the future. # performance. Worth tunning for other models in the future.
num_threads = len(self._places) * 2 exec_strategy.num_threads = len(self._places) * 2
else: else:
num_threads = min( exec_strategy.num_threads = min(
len(self._places) * 2, multiprocessing.cpu_count()) len(self._places) * 2, multiprocessing.cpu_count())
if build_strategy is None:
build_strategy = BuildStrategy()
main = main_program main = main_program
main = main if main else framework.default_main_program() main = main if main else framework.default_main_program()
scope = executor.global_scope() scope = executor.global_scope()
...@@ -127,23 +145,14 @@ class ParallelExecutor(object): ...@@ -127,23 +145,14 @@ class ParallelExecutor(object):
] ]
self.executor = core.ParallelExecutor( self.executor = core.ParallelExecutor(
num_threads,
True if use_cuda else False, # use_event
self._places, self._places,
set([ set([
p.name for p in main.global_block().iter_parameters() p.name for p in main.global_block().iter_parameters()
if not p.stop_gradient if not p.stop_gradient
]), ]),
set(self.persistable_vars), set(self.persistable_vars), main.desc, loss_name
main.desc, if loss_name else '', scope, local_scopes, exec_strategy,
loss_name if loss_name else '', build_strategy, num_trainers, trainer_id)
scope,
local_scopes,
allow_op_delay,
use_default_grad_scale,
balance_parameter_opt_between_cards,
num_trainers,
trainer_id)
self.scope = scope self.scope = scope
def run(self, fetch_list, feed=None, feed_dict=None): def run(self, fetch_list, feed=None, feed_dict=None):
......
...@@ -48,12 +48,11 @@ def linear(): ...@@ -48,12 +48,11 @@ def linear():
return avg_loss return avg_loss
def train(use_cuda, save_dirname): def train(use_cuda, train_program, save_dirname):
place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace() place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
trainer = fluid.Trainer( trainer = fluid.Trainer(
train_func=linear, train_func=train_program,
infer_func=inference_program,
place=place, place=place,
optimizer=fluid.optimizer.SGD(learning_rate=0.001)) optimizer=fluid.optimizer.SGD(learning_rate=0.001))
...@@ -72,11 +71,7 @@ def train(use_cuda, save_dirname): ...@@ -72,11 +71,7 @@ def train(use_cuda, save_dirname):
''' '''
if float(test_metrics[0]) < 20.0: if float(test_metrics[0]) < 20.0:
if save_dirname is not None: if save_dirname is not None:
# NOT clear yet trainer.save_params(save_dirname)
# fluid.io.save_inference_model(save_dirname, ['x'], [y_predict])
# trainer.save_params(save_dirname)
# https://github.com/PaddlePaddle/Paddle/pull/10445
trainer.save_inference_model(save_dirname)
return return
trainer.train( trainer.train(
...@@ -87,12 +82,13 @@ def train(use_cuda, save_dirname): ...@@ -87,12 +82,13 @@ def train(use_cuda, save_dirname):
# infer # infer
def infer(use_cuda, save_dirname=None): def infer(use_cuda, inference_program, save_dirname=None):
if save_dirname is None: if save_dirname is None:
return return
place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace() place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
inferencer = fluid.Inferencer(param_path=save_dirname, place=place) inferencer = fluid.Inferencer(
infer_func=inference_program, param_path=save_dirname, place=place)
batch_size = 10 batch_size = 10
tensor_x = numpy.random.uniform(0, 10, [batch_size, 13]).astype("float32") tensor_x = numpy.random.uniform(0, 10, [batch_size, 13]).astype("float32")
...@@ -108,8 +104,8 @@ def main(use_cuda): ...@@ -108,8 +104,8 @@ def main(use_cuda):
# Directory for saving the trained model # Directory for saving the trained model
save_dirname = "fit_a_line.inference.model" save_dirname = "fit_a_line.inference.model"
train(use_cuda, save_dirname) train(use_cuda, linear, save_dirname)
infer(use_cuda, save_dirname) infer(use_cuda, inference_program, save_dirname)
class TestFitALine(unittest.TestCase): class TestFitALine(unittest.TestCase):
......
...@@ -53,48 +53,40 @@ def train_program(): ...@@ -53,48 +53,40 @@ def train_program():
predict = inference_program() predict = inference_program()
cost = fluid.layers.cross_entropy(input=predict, label=label) cost = fluid.layers.cross_entropy(input=predict, label=label)
avg_cost = fluid.layers.mean(cost) avg_cost = fluid.layers.mean(cost)
# acc = fluid.layers.accuracy(input=predict, label=label) acc = fluid.layers.accuracy(input=predict, label=label)
# return avg_cost, acc return [avg_cost, acc]
return avg_cost
def train(use_cuda, save_dirname): def train(use_cuda, train_program, save_dirname):
place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace() place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
optimizer = fluid.optimizer.Adam(learning_rate=0.001) optimizer = fluid.optimizer.Adam(learning_rate=0.001)
trainer = fluid.Trainer( trainer = fluid.Trainer(
train_func=train_program, train_func=train_program, place=place, optimizer=optimizer)
infer_func=inference_program,
place=place,
optimizer=optimizer)
def event_handler(event): def event_handler(event):
if isinstance(event, fluid.EndEpochEvent): if isinstance(event, fluid.EndEpochEvent):
# if (event.epoch + 1) % 10 == 0: test_reader = paddle.batch(
# trainer.save_params(save_dirname) paddle.dataset.mnist.test(), batch_size=BATCH_SIZE)
trainer.save_inference_model(save_dirname) test_metrics = trainer.test(
reader=test_reader, feed_order=['img', 'label'])
# TODO: Uncomment this part once we are sure that .train is working avg_cost_set = test_metrics[0]
# test_reader = paddle.batch( acc_set = test_metrics[1]
# paddle.dataset.mnist.test(), batch_size=BATCH_SIZE)
# test_metrics = trainer.test(reader=test_reader) # get test acc and loss
# avg_cost_set = test_metrics[0] acc = numpy.array(acc_set).mean()
# acc_set = test_metrics[1] avg_cost = numpy.array(avg_cost_set).mean()
#
# # get test acc and loss print("avg_cost: %s" % avg_cost)
# acc = numpy.array(acc_set).mean() print("acc : %s" % acc)
# avg_cost = numpy.array(avg_cost_set).mean()
# if float(acc) > 0.2: # Smaller value to increase CI speed
# print("avg_cost: %s" % avg_cost) trainer.save_params(save_dirname)
# print("acc : %s" % acc) else:
# print('BatchID {0}, Test Loss {1:0.2}, Acc {2:0.2}'.format(
# if float(acc) > 0.2: # Smaller value to increase CI speed event.epoch + 1, float(avg_cost), float(acc)))
# trainer.save_params(save_dirname) if math.isnan(float(avg_cost)):
# else: sys.exit("got NaN loss, training failed.")
# print('BatchID {0}, Test Loss {1:0.2}, Acc {2:0.2}'.format(
# event.epoch + 1, float(avg_cost), float(acc)))
# if math.isnan(float(avg_cost)):
# sys.exit("got NaN loss, training failed.")
train_reader = paddle.batch( train_reader = paddle.batch(
paddle.reader.shuffle( paddle.reader.shuffle(
...@@ -108,10 +100,11 @@ def train(use_cuda, save_dirname): ...@@ -108,10 +100,11 @@ def train(use_cuda, save_dirname):
feed_order=['img', 'label']) feed_order=['img', 'label'])
def infer(use_cuda, save_dirname=None): def infer(use_cuda, inference_program, save_dirname=None):
place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace() place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
inferencer = fluid.Inferencer(param_path=save_dirname, place=place) inferencer = fluid.Inferencer(
infer_func=inference_program, param_path=save_dirname, place=place)
batch_size = 1 batch_size = 1
tensor_img = numpy.random.uniform(-1.0, 1.0, tensor_img = numpy.random.uniform(-1.0, 1.0,
...@@ -126,8 +119,14 @@ def main(use_cuda): ...@@ -126,8 +119,14 @@ def main(use_cuda):
save_dirname = "recognize_digits_conv.inference.model" save_dirname = "recognize_digits_conv.inference.model"
# call train() with is_local argument to run distributed train # call train() with is_local argument to run distributed train
train(use_cuda=use_cuda, save_dirname=save_dirname) train(
infer(use_cuda=use_cuda, save_dirname=save_dirname) use_cuda=use_cuda,
train_program=train_program,
save_dirname=save_dirname)
infer(
use_cuda=use_cuda,
inference_program=inference_program,
save_dirname=save_dirname)
if __name__ == '__main__': if __name__ == '__main__':
......
...@@ -40,47 +40,40 @@ def train_program(): ...@@ -40,47 +40,40 @@ def train_program():
predict = inference_program() predict = inference_program()
cost = fluid.layers.cross_entropy(input=predict, label=label) cost = fluid.layers.cross_entropy(input=predict, label=label)
avg_cost = fluid.layers.mean(cost) avg_cost = fluid.layers.mean(cost)
# acc = fluid.layers.accuracy(input=predict, label=label) acc = fluid.layers.accuracy(input=predict, label=label)
# return avg_cost, acc return [avg_cost, acc]
return avg_cost
def train(use_cuda, save_dirname): def train(use_cuda, train_program, save_dirname):
place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace() place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
optimizer = fluid.optimizer.Adam(learning_rate=0.001) optimizer = fluid.optimizer.Adam(learning_rate=0.001)
trainer = fluid.Trainer( trainer = fluid.Trainer(
train_func=train_program, train_func=train_program, place=place, optimizer=optimizer)
infer_func=inference_program,
place=place,
optimizer=optimizer)
def event_handler(event): def event_handler(event):
if isinstance(event, fluid.EndEpochEvent): if isinstance(event, fluid.EndEpochEvent):
# if (event.epoch + 1) % 10 == 0: test_reader = paddle.batch(
trainer.save_inference_model(save_dirname) paddle.dataset.mnist.test(), batch_size=BATCH_SIZE)
test_metrics = trainer.test(
# TODO: Uncomment this part once we are sure that .train is working reader=test_reader, feed_order=['img', 'label'])
# test_reader = paddle.batch( avg_cost_set = test_metrics[0]
# paddle.dataset.mnist.test(), batch_size=BATCH_SIZE) acc_set = test_metrics[1]
# test_metrics = trainer.test(reader=test_reader)
# avg_cost_set = test_metrics[0] # get test acc and loss
# acc_set = test_metrics[1] acc = numpy.array(acc_set).mean()
# avg_cost = numpy.array(avg_cost_set).mean()
# # get test acc and loss
# acc = numpy.array(acc_set).mean() print("avg_cost: %s" % avg_cost)
# avg_cost = numpy.array(avg_cost_set).mean() print("acc : %s" % acc)
#
# print("avg_cost: %s" % avg_cost) if float(acc) > 0.2: # Smaller value to increase CI speed
# print("acc : %s" % acc) trainer.save_params(save_dirname)
# else:
# if float(acc) > 0.2: # Smaller value to increase CI speed print('BatchID {0}, Test Loss {1:0.2}, Acc {2:0.2}'.format(
# trainer.save_params(save_dirname) event.epoch + 1, float(avg_cost), float(acc)))
# else: if math.isnan(float(avg_cost)):
# print('BatchID {0}, Test Loss {1:0.2}, Acc {2:0.2}'.format( sys.exit("got NaN loss, training failed.")
# event.epoch + 1, float(avg_cost), float(acc)))
# if math.isnan(float(avg_cost)):
# sys.exit("got NaN loss, training failed.")
train_reader = paddle.batch( train_reader = paddle.batch(
paddle.reader.shuffle( paddle.reader.shuffle(
...@@ -94,10 +87,11 @@ def train(use_cuda, save_dirname): ...@@ -94,10 +87,11 @@ def train(use_cuda, save_dirname):
feed_order=['img', 'label']) feed_order=['img', 'label'])
def infer(use_cuda, save_dirname=None): def infer(use_cuda, inference_program, save_dirname=None):
place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace() place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
inferencer = fluid.Inferencer(param_path=save_dirname, place=place) inferencer = fluid.Inferencer(
infer_func=inference_program, param_path=save_dirname, place=place)
batch_size = 1 batch_size = 1
tensor_img = numpy.random.uniform(-1.0, 1.0, tensor_img = numpy.random.uniform(-1.0, 1.0,
...@@ -112,8 +106,14 @@ def main(use_cuda): ...@@ -112,8 +106,14 @@ def main(use_cuda):
save_dirname = "recognize_digits_mlp.inference.model" save_dirname = "recognize_digits_mlp.inference.model"
# call train() with is_local argument to run distributed train # call train() with is_local argument to run distributed train
train(use_cuda=use_cuda, save_dirname=save_dirname) train(
infer(use_cuda=use_cuda, save_dirname=save_dirname) use_cuda=use_cuda,
train_program=train_program,
save_dirname=save_dirname)
infer(
use_cuda=use_cuda,
inference_program=inference_program,
save_dirname=save_dirname)
if __name__ == '__main__': if __name__ == '__main__':
......
...@@ -90,7 +90,7 @@ def train_program(is_sparse): ...@@ -90,7 +90,7 @@ def train_program(is_sparse):
return avg_cost return avg_cost
def train(use_cuda, is_sparse, save_path): def train(use_cuda, train_program, save_path):
train_reader = paddle.batch( train_reader = paddle.batch(
paddle.dataset.imikolov.train(word_dict, N), BATCH_SIZE) paddle.dataset.imikolov.train(word_dict, N), BATCH_SIZE)
test_reader = paddle.batch( test_reader = paddle.batch(
...@@ -105,23 +105,21 @@ def train(use_cuda, is_sparse, save_path): ...@@ -105,23 +105,21 @@ def train(use_cuda, is_sparse, save_path):
print("loss= ", avg_cost) print("loss= ", avg_cost)
if avg_cost < 5.0: if avg_cost < 5.0:
trainer.save_inference_model(save_path) trainer.save_params(save_path)
return return
if math.isnan(avg_cost): if math.isnan(avg_cost):
sys.exit("got NaN loss, training failed.") sys.exit("got NaN loss, training failed.")
trainer = fluid.Trainer( trainer = fluid.Trainer(
partial(train_program, is_sparse), train_program, fluid.optimizer.SGD(learning_rate=0.001), place=place)
partial(inference_program, is_sparse),
fluid.optimizer.SGD(learning_rate=0.001),
place=place)
trainer.train( trainer.train(
reader=train_reader, num_epochs=1, event_handler=event_handler) reader=train_reader, num_epochs=1, event_handler=event_handler)
def infer(use_cuda, is_sparse, save_path): def infer(use_cuda, inference_program, save_path):
place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace() place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
inferencer = fluid.Inferencer(param_path=save_path, place=place) inferencer = fluid.Inferencer(
infer_func=inference_program, param_path=save_path, place=place)
lod = [0, 1] lod = [0, 1]
first_word = create_random_lodtensor(lod, place, low=0, high=dict_size - 1) first_word = create_random_lodtensor(lod, place, low=0, high=dict_size - 1)
...@@ -144,9 +142,9 @@ def main(use_cuda, is_sparse): ...@@ -144,9 +142,9 @@ def main(use_cuda, is_sparse):
if use_cuda and not fluid.core.is_compiled_with_cuda(): if use_cuda and not fluid.core.is_compiled_with_cuda():
return return
save_path = "word2vec.inference.model" save_path = "word2vec.params"
train(use_cuda, is_sparse, save_path) train(use_cuda, partial(train_program, is_sparse), save_path)
infer(use_cuda, is_sparse, save_path) infer(use_cuda, partial(inference_program, is_sparse), save_path)
if __name__ == '__main__': if __name__ == '__main__':
......
...@@ -170,7 +170,7 @@ def train(word_dict, ...@@ -170,7 +170,7 @@ def train(word_dict,
assert save_dirname is None assert save_dirname is None
adagrad = fluid.optimizer.Adagrad(learning_rate=0.002) adagrad = fluid.optimizer.Adagrad(learning_rate=0.002)
optimize_ops, params_grads = adagrad.minimize(cost) adagrad.minimize(cost)
train_data = paddle.batch( train_data = paddle.batch(
paddle.reader.shuffle( paddle.reader.shuffle(
......
...@@ -33,7 +33,7 @@ def train(use_cuda, save_dirname, is_local): ...@@ -33,7 +33,7 @@ def train(use_cuda, save_dirname, is_local):
avg_cost = fluid.layers.mean(cost) avg_cost = fluid.layers.mean(cost)
sgd_optimizer = fluid.optimizer.SGD(learning_rate=0.001) sgd_optimizer = fluid.optimizer.SGD(learning_rate=0.001)
optimize_ops, params_grads = sgd_optimizer.minimize(avg_cost) sgd_optimizer.minimize(avg_cost)
BATCH_SIZE = 20 BATCH_SIZE = 20
......
...@@ -125,7 +125,7 @@ def train(net_type, use_cuda, save_dirname, is_local): ...@@ -125,7 +125,7 @@ def train(net_type, use_cuda, save_dirname, is_local):
test_program = fluid.default_main_program().clone(for_test=True) test_program = fluid.default_main_program().clone(for_test=True)
optimizer = fluid.optimizer.Adam(learning_rate=0.001) optimizer = fluid.optimizer.Adam(learning_rate=0.001)
optimize_ops, params_grads = optimizer.minimize(avg_cost) optimizer.minimize(avg_cost)
BATCH_SIZE = 128 BATCH_SIZE = 128
PASS_NUM = 1 PASS_NUM = 1
......
...@@ -175,7 +175,7 @@ def train(use_cuda, save_dirname=None, is_local=True): ...@@ -175,7 +175,7 @@ def train(use_cuda, save_dirname=None, is_local=True):
decay_steps=100000, decay_steps=100000,
decay_rate=0.5, decay_rate=0.5,
staircase=True)) staircase=True))
optimize_ops, params_grads = sgd_optimizer.minimize(avg_cost) sgd_optimizer.minimize(avg_cost)
# TODO(qiao) # TODO(qiao)
# add dependency track and move this config before optimizer # add dependency track and move this config before optimizer
......
...@@ -185,7 +185,7 @@ def train_main(use_cuda, is_sparse, is_local=True): ...@@ -185,7 +185,7 @@ def train_main(use_cuda, is_sparse, is_local=True):
learning_rate=1e-4, learning_rate=1e-4,
regularization=fluid.regularizer.L2DecayRegularizer( regularization=fluid.regularizer.L2DecayRegularizer(
regularization_coeff=0.1)) regularization_coeff=0.1))
optimize_ops, params_grads = optimizer.minimize(avg_cost) optimizer.minimize(avg_cost)
train_data = paddle.batch( train_data = paddle.batch(
paddle.reader.shuffle( paddle.reader.shuffle(
......
...@@ -95,7 +95,7 @@ def train(nn_type, ...@@ -95,7 +95,7 @@ def train(nn_type,
test_program = fluid.default_main_program().clone(for_test=True) test_program = fluid.default_main_program().clone(for_test=True)
optimizer = fluid.optimizer.Adam(learning_rate=0.001) optimizer = fluid.optimizer.Adam(learning_rate=0.001)
optimize_ops, params_grads = optimizer.minimize(avg_loss) optimizer.minimize(avg_loss)
place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace() place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
......
...@@ -160,7 +160,7 @@ def train(use_cuda, save_dirname, is_local=True): ...@@ -160,7 +160,7 @@ def train(use_cuda, save_dirname, is_local=True):
test_program = fluid.default_main_program().clone(for_test=True) test_program = fluid.default_main_program().clone(for_test=True)
sgd_optimizer = SGDOptimizer(learning_rate=0.2) sgd_optimizer = SGDOptimizer(learning_rate=0.2)
optimize_ops, params_grads = sgd_optimizer.minimize(avg_cost) sgd_optimizer.minimize(avg_cost)
place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace() place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
......
...@@ -101,7 +101,7 @@ def train(use_cuda, is_sparse, is_parallel, save_dirname, is_local=True): ...@@ -101,7 +101,7 @@ def train(use_cuda, is_sparse, is_parallel, save_dirname, is_local=True):
avg_cost = fluid.layers.mean(pd()) avg_cost = fluid.layers.mean(pd())
sgd_optimizer = fluid.optimizer.SGD(learning_rate=0.001) sgd_optimizer = fluid.optimizer.SGD(learning_rate=0.001)
optimize_ops, params_grads = sgd_optimizer.minimize(avg_cost) sgd_optimizer.minimize(avg_cost)
train_reader = paddle.batch( train_reader = paddle.batch(
paddle.dataset.imikolov.train(word_dict, N), BATCH_SIZE) paddle.dataset.imikolov.train(word_dict, N), BATCH_SIZE)
......
# 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.
import unittest
import paddle.fluid as fluid
import paddle.fluid.core as core
import paddle.fluid.layers as layers
from paddle.fluid.transpiler.distribute_transpiler import delete_ops
import numpy
class TestDistTranspiler(unittest.TestCase):
def setUp(self):
self.trainer_id = 0
self.trainers = 2
self.pservers = 2
self.pserver_eps = "127.0.0.1:6174,127.0.0.1:6175"
self.current_pserver_ep = "127.0.0.1:6174"
def net_conf(self):
x = fluid.layers.data(name='x', shape=[1000], dtype='float32')
y_predict = fluid.layers.fc(input=x,
size=1000,
act=None,
param_attr=fluid.ParamAttr(name='fc_w'))
y = fluid.layers.data(name='y', shape=[1], dtype='float32')
cost = fluid.layers.square_error_cost(input=y_predict, label=y)
avg_cost = fluid.layers.mean(cost)
sgd_optimizer = fluid.optimizer.SGD(learning_rate=0.1)
optimize_ops, params_grads = sgd_optimizer.minimize(avg_cost)
return optimize_ops, params_grads
def test_transpiler(self):
trainer = self.get_trainer()
pserver, startup = self.get_pserver(self.current_pserver_ep)
self.assertEqual([op.type for op in trainer.global_block().ops],
self.get_expect_trainer_ops())
self.assertEqual(len(pserver.blocks), 3)
# block0: listen_and_serv
self.assertEqual([op.type for op in pserver.blocks[0].ops],
["listen_and_serv"])
# block2: optimize pass
self.assertEqual([op.type for op in pserver.blocks[1].ops],
["sum", "scale", "sgd"])
# confirm startup program
self.assertEqual([op.type for op in startup.global_block().ops], [
"fill_constant", "fill_constant", "uniform_random", "uniform_random"
])
# the variable #fc_w will be split into two blocks
fc_w_var = startup.global_block().var("fc_w.block1")
self.assertEqual(fc_w_var.shape, (500, 1000))
def get_main_program(self):
main = fluid.Program()
with fluid.program_guard(main):
self.net_conf()
return main
def get_expect_trainer_ops(self):
trainer = fluid.Program()
with fluid.program_guard(trainer):
optimize_ops, params_grads = self.net_conf()
delete_ops(trainer.global_block(), optimize_ops)
return [op.type for op in trainer.global_block().ops
] + ["split_byref", "send", "concat"]
def get_trainer(self):
return self._transpiler_instance().get_trainer_program()
def get_pserver(self, ep):
t = self._transpiler_instance()
pserver = t.get_pserver_program(ep)
startup = t.get_startup_program(ep, pserver)
return pserver, startup
def _transpiler_instance(self):
main = self.get_main_program()
t = fluid.DistributeTranspiler()
t.transpile(
self.trainer_id,
program=main,
pservers=self.pserver_eps,
trainers=self.trainers)
return t
if __name__ == "__main__":
unittest.main()
...@@ -232,14 +232,18 @@ class TestParallelExecutorBase(unittest.TestCase): ...@@ -232,14 +232,18 @@ class TestParallelExecutorBase(unittest.TestCase):
place = fluid.CUDAPlace(0) place = fluid.CUDAPlace(0)
startup_exe = fluid.Executor(place) startup_exe = fluid.Executor(place)
startup_exe.run(startup) startup_exe.run(startup)
exec_strategy = fluid.ExecutionStrategy()
exec_strategy.allow_op_delay = allow_op_delay
build_strategy = fluid.BuildStrategy()
build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.Reduce if balance_parameter_opt_between_cards else fluid.BuildStrategy.ReduceStrategy.AllReduce
if use_parallel_executor: if use_parallel_executor:
exe = fluid.ParallelExecutor( exe = fluid.ParallelExecutor(
True, True,
loss_name=loss.name, loss_name=loss.name,
allow_op_delay=allow_op_delay, exec_strategy=exec_strategy,
balance_parameter_opt_between_cards=balance_parameter_opt_between_cards build_strategy=build_strategy)
)
else: else:
exe = fluid.Executor(place=place) exe = fluid.Executor(place=place)
...@@ -548,7 +552,7 @@ class TestTransformer(TestParallelExecutorBase): ...@@ -548,7 +552,7 @@ class TestTransformer(TestParallelExecutorBase):
class ParallelExecutorTestingDuringTraining(unittest.TestCase): class ParallelExecutorTestingDuringTraining(unittest.TestCase):
def check_network_convergence(self, balance_parameter_opt_between_cards): def check_network_convergence(self, build_strategy=None):
main = fluid.Program() main = fluid.Program()
startup = fluid.Program() startup = fluid.Program()
with fluid.program_guard(main, startup): with fluid.program_guard(main, startup):
...@@ -571,15 +575,13 @@ class ParallelExecutorTestingDuringTraining(unittest.TestCase): ...@@ -571,15 +575,13 @@ class ParallelExecutorTestingDuringTraining(unittest.TestCase):
use_cuda=True, use_cuda=True,
loss_name=loss.name, loss_name=loss.name,
main_program=main, main_program=main,
balance_parameter_opt_between_cards=balance_parameter_opt_between_cards build_strategy=build_strategy)
)
test_exe = fluid.ParallelExecutor( test_exe = fluid.ParallelExecutor(
use_cuda=True, use_cuda=True,
main_program=test_program, main_program=test_program,
share_vars_from=train_exe, share_vars_from=train_exe,
balance_parameter_opt_between_cards=balance_parameter_opt_between_cards build_strategy=build_strategy)
)
for i in xrange(5): for i in xrange(5):
test_loss, = test_exe.run([loss.name], feed=feed_dict) test_loss, = test_exe.run([loss.name], feed=feed_dict)
...@@ -594,10 +596,14 @@ class ParallelExecutorTestingDuringTraining(unittest.TestCase): ...@@ -594,10 +596,14 @@ class ParallelExecutorTestingDuringTraining(unittest.TestCase):
str(test_loss)) str(test_loss))
def test_parallel_testing(self): def test_parallel_testing(self):
self.check_network_convergence(False) build_strategy = fluid.BuildStrategy()
build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.AllReduce
self.check_network_convergence(build_strategy)
def test_parallel_testing_with_new_strategy(self): def test_parallel_testing_with_new_strategy(self):
self.check_network_convergence(True) build_strategy = fluid.BuildStrategy()
build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.Reduce
self.check_network_convergence(build_strategy)
import paddle.dataset.conll05 as conll05 import paddle.dataset.conll05 as conll05
...@@ -617,7 +623,7 @@ embedding_name = 'emb' ...@@ -617,7 +623,7 @@ embedding_name = 'emb'
def db_lstm(word, predicate, ctx_n2, ctx_n1, ctx_0, ctx_p1, ctx_p2, mark, def db_lstm(word, predicate, ctx_n2, ctx_n1, ctx_0, ctx_p1, ctx_p2, mark,
is_sparse, balance_parameter_opt_between_cards, **ignored): is_sparse, **ignored):
# 8 features # 8 features
predicate_embedding = fluid.layers.embedding( predicate_embedding = fluid.layers.embedding(
input=predicate, input=predicate,
...@@ -686,9 +692,7 @@ def db_lstm(word, predicate, ctx_n2, ctx_n1, ctx_0, ctx_p1, ctx_p2, mark, ...@@ -686,9 +692,7 @@ def db_lstm(word, predicate, ctx_n2, ctx_n1, ctx_0, ctx_p1, ctx_p2, mark,
class TestCRFModel(unittest.TestCase): class TestCRFModel(unittest.TestCase):
def check_network_convergence(self, def check_network_convergence(self, is_sparse, build_strategy=None):
is_sparse,
balance_parameter_opt_between_cards=False):
main = fluid.Program() main = fluid.Program()
startup = fluid.Program() startup = fluid.Program()
with fluid.program_guard(main, startup): with fluid.program_guard(main, startup):
...@@ -739,8 +743,7 @@ class TestCRFModel(unittest.TestCase): ...@@ -739,8 +743,7 @@ class TestCRFModel(unittest.TestCase):
pe = fluid.ParallelExecutor( pe = fluid.ParallelExecutor(
use_cuda=True, use_cuda=True,
loss_name=avg_cost.name, loss_name=avg_cost.name,
balance_parameter_opt_between_cards=balance_parameter_opt_between_cards build_strategy=build_strategy)
)
feeder = fluid.DataFeeder( feeder = fluid.DataFeeder(
feed_list=[ feed_list=[
...@@ -756,19 +759,29 @@ class TestCRFModel(unittest.TestCase): ...@@ -756,19 +759,29 @@ class TestCRFModel(unittest.TestCase):
pe.run(feed=feeder.feed(cur_batch), pe.run(feed=feeder.feed(cur_batch),
fetch_list=[avg_cost.name]))[0] fetch_list=[avg_cost.name]))[0]
def test_update_sparse_parameter(self): def test_update_sparse_parameter_all_reduce(self):
self.check_network_convergence(is_sparse=True) build_strategy = fluid.BuildStrategy()
build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.AllReduce
self.check_network_convergence(
is_sparse=True, build_strategy=build_strategy)
def test_update_dense_parameter(self): def test_update_dense_parameter_all_reduce(self):
self.check_network_convergence(is_sparse=False) build_strategy = fluid.BuildStrategy()
build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.AllReduce
self.check_network_convergence(
is_sparse=False, build_strategy=build_strategy)
def test_update_sparse_parameter_with_new_strategy(self): def test_update_sparse_parameter_reduce(self):
build_strategy = fluid.BuildStrategy()
build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.Reduce
self.check_network_convergence( self.check_network_convergence(
is_sparse=False, balance_parameter_opt_between_cards=True) is_sparse=False, build_strategy=build_strategy)
def test_update_dense_parameter_with_new_strategy(self): def test_update_dense_parameter_reduce(self):
build_strategy = fluid.BuildStrategy()
build_strategy.reduce_strategy = fluid.BuildStrategy.ReduceStrategy.Reduce
self.check_network_convergence( self.check_network_convergence(
is_sparse=False, balance_parameter_opt_between_cards=True) is_sparse=False, build_strategy=build_strategy)
# test fetch all the variables of global_block # test fetch all the variables of global_block
...@@ -836,7 +849,8 @@ class TestFetchOp(unittest.TestCase): ...@@ -836,7 +849,8 @@ class TestFetchOp(unittest.TestCase):
assert not math.isnan(np.sum(ret[i])) and \ assert not math.isnan(np.sum(ret[i])) and \
not math.isinf(np.sum(ret[i])) not math.isinf(np.sum(ret[i]))
def test_update_sparse_parameter(self): @unittest.skip("this test is buggy")
def test_feed(self):
tst_reader = paddle.batch(flowers.test(use_xmap=False), batch_size=16) tst_reader = paddle.batch(flowers.test(use_xmap=False), batch_size=16)
tst_reader_iter = tst_reader() tst_reader_iter = tst_reader()
......
...@@ -21,15 +21,7 @@ import random ...@@ -21,15 +21,7 @@ import random
class TestSplitVar(unittest.TestCase): class TestSplitVar(unittest.TestCase):
def test_check_output(self): def check_split_output(self, shapes, expected_sizes, min_size):
# split below shapes to 10 servers
shapes = [[3, 5], [1024], [28, 784], [8, 1020], [800, 10]]
expected_sizes = [
[15], [1024],
[2352, 2352, 2352, 2352, 2352, 2352, 2352, 2352, 2352, 784],
[2040, 2040, 2040, 2040],
[1150, 1150, 1150, 1150, 1150, 1150, 1100]
]
var_list = [] var_list = []
program = fluid.Program() program = fluid.Program()
for shape in shapes: for shape in shapes:
...@@ -39,7 +31,7 @@ class TestSplitVar(unittest.TestCase): ...@@ -39,7 +31,7 @@ class TestSplitVar(unittest.TestCase):
# dtype=core.VarDesc.VarType.LOD_TENSOR, # dtype=core.VarDesc.VarType.LOD_TENSOR,
shape=shape) shape=shape)
var_list.append(var) var_list.append(var)
blocks = split_dense_variable(var_list, 10) blocks = split_dense_variable(var_list, 10, min_size)
all_sizes = [] all_sizes = []
for s in expected_sizes: for s in expected_sizes:
for s2 in s: for s2 in s:
...@@ -48,6 +40,25 @@ class TestSplitVar(unittest.TestCase): ...@@ -48,6 +40,25 @@ class TestSplitVar(unittest.TestCase):
varname, block_id, size = block_str.split(":") varname, block_id, size = block_str.split(":")
self.assertEqual(int(size), all_sizes[i]) self.assertEqual(int(size), all_sizes[i])
def test_1k(self):
shapes = [[3, 5], [1024], [28, 784], [8, 1020], [800, 10]]
expected_sizes = [
[15], [1024],
[2352, 2352, 2352, 2352, 2352, 2352, 2352, 2352, 2352, 784],
[2040, 2040, 2040, 2040],
[1150, 1150, 1150, 1150, 1150, 1150, 1100]
]
self.check_split_output(shapes, expected_sizes, 1024)
def test_check_output_8k(self):
shapes = [[3, 5], [1024], [28, 784], [8, 1020], [800, 10],
[6, 33, 33, 33]]
expected_sizes = [[15], [1024], [10976, 10976], [8160], [8000],
[35937, 35937, 35937, 35937, 35937, 35937]]
self.check_split_output(shapes, expected_sizes, 8192)
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()
...@@ -92,19 +92,13 @@ class Trainer(object): ...@@ -92,19 +92,13 @@ class Trainer(object):
place: The device place of this trainer. place: The device place of this trainer.
""" """
def __init__(self, def __init__(self, train_func, optimizer, param_path=None, place=None):
train_func,
infer_func,
optimizer,
param_path=None,
place=None):
# 1. we need to generate a framework.Program by calling # 1. we need to generate a framework.Program by calling
# program_func. Reference: fluid.program_guard in # program_func. Reference: fluid.program_guard in
# test_word2vec.py # test_word2vec.py
if not isinstance(optimizer, opt_module.Optimizer): if not isinstance(optimizer, opt_module.Optimizer):
raise TypeError("The optimizer should be an instance of Optimizer") raise TypeError("The optimizer should be an instance of Optimizer")
self.infer_func = infer_func
self.scope = core.Scope() self.scope = core.Scope()
self.startup_program = framework.Program() self.startup_program = framework.Program()
...@@ -178,9 +172,9 @@ class Trainer(object): ...@@ -178,9 +172,9 @@ class Trainer(object):
def train(self, def train(self,
num_epochs, num_epochs,
event_handler, event_handler,
reader=None, reader,
parallel=False, feed_order,
feed_order=None): parallel=False):
""" """
Train the model. Train the model.
...@@ -208,7 +202,7 @@ class Trainer(object): ...@@ -208,7 +202,7 @@ class Trainer(object):
self._train_by_executor(num_epochs, event_handler, reader, feed_order) self._train_by_executor(num_epochs, event_handler, reader, feed_order)
def test(self, reader, feed_order=None): def test(self, reader, feed_order):
""" """
Test the model on given test data Test the model on given test data
...@@ -226,15 +220,6 @@ class Trainer(object): ...@@ -226,15 +220,6 @@ class Trainer(object):
exe = executor.Executor(self.place) exe = executor.Executor(self.place)
io.save_persistables(exe, dirname=param_path) io.save_persistables(exe, dirname=param_path)
def save_inference_model(self, model_path):
inference_program = framework.Program()
with framework.program_guard(inference_program):
with unique_name.guard():
predict_var = self.infer_func()
predict_var = self.train_program.block(0).var(predict_var.name)
exe = executor.Executor(self.place)
io.save_inference_model(model_path, [], [predict_var], exe)
@contextlib.contextmanager @contextlib.contextmanager
def _prog_and_scope_guard(self): def _prog_and_scope_guard(self):
with framework.program_guard( with framework.program_guard(
...@@ -291,12 +276,7 @@ def build_feed_var_list(program, feed_order): ...@@ -291,12 +276,7 @@ def build_feed_var_list(program, feed_order):
if not isinstance(program, framework.Program): if not isinstance(program, framework.Program):
raise TypeError("The 'program' should be an object of Program") raise TypeError("The 'program' should be an object of Program")
if feed_order is None: if isinstance(feed_order, list):
feed_var_list = [
var for var in program.global_block().vars.itervalues()
if var.is_data
]
elif isinstance(feed_order, list):
feed_var_list = [ feed_var_list = [
program.global_block().var(var_name) for var_name in feed_order program.global_block().var(var_name) for var_name in feed_order
] ]
......
...@@ -11,6 +11,7 @@ ...@@ -11,6 +11,7 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
from distribute_transpiler import DistributeTranspiler from distribute_transpiler import DistributeTranspiler
from inference_transpiler import InferenceTranspiler from inference_transpiler import InferenceTranspiler
from memory_optimization_transpiler import memory_optimize, release_memory from memory_optimization_transpiler import memory_optimize, release_memory
......
...@@ -17,7 +17,7 @@ from __future__ import print_function ...@@ -17,7 +17,7 @@ from __future__ import print_function
import math import math
import distributed_splitter as splitter import distributed_splitter as splitter
from .. import core from .. import core, framework
from ..framework import Program, default_main_program, \ from ..framework import Program, default_main_program, \
default_startup_program, \ default_startup_program, \
Variable, Parameter, grad_var_name Variable, Parameter, grad_var_name
...@@ -93,30 +93,33 @@ def same_or_split_var(p_name, var_name): ...@@ -93,30 +93,33 @@ def same_or_split_var(p_name, var_name):
return p_name == var_name or p_name.startswith(var_name + ".block") return p_name == var_name or p_name.startswith(var_name + ".block")
def split_dense_variable(var_list, def split_dense_variable(var_list, service_count, min_block_size=8192):
pserver_count,
min_block_size=1024,
max_block_size=1048576):
""" """
We may need to split dense tensor to one or more blocks and put We may need to split dense tensor to one or more blocks and put
them equally onto parameter server. One block is a sub-tensor them equally onto parameter server. One block is a sub-tensor
aligned by dim[0] of the tensor. aligned by dim[0] of the tensor.
We need to have a minimal block size so that the calculations in We need to have a minimal block size so that the calculations in
the parameter server side can gain better performance. By default the parameter server side can gain better performance. By default
minimum block size is 1024. The max block size is used to prevent minimum block size 8K elements (maybe 16bit or 32bit or 64bit).
very large blocks that may cause send error.
:return: A list of VarBlocks. Each VarBlock specifies a shard of Args:
the var. var_list (list): List of variables.
service_count (int): Numel of pserver services. A pserver may have two
or more listening ports.
min_block_size (int): Minimum splitted block size.
Returns:
blocks (list[(varname, block_id, current_block_size)]): A list
of VarBlocks. Each VarBlock specifies a shard of the var.
""" """
blocks = [] blocks = []
for var in var_list: for var in var_list:
split_count = pserver_count split_count = service_count
var_numel = reduce(lambda x, y: x * y, var.shape) var_numel = reduce(lambda x, y: x * y, var.shape)
max_pserver_count = int(math.floor(var_numel / float(min_block_size))) max_pserver_count = int(math.floor(var_numel / float(min_block_size)))
if max_pserver_count == 0: if max_pserver_count == 0:
max_pserver_count = 1 max_pserver_count = 1
if max_pserver_count < pserver_count: if max_pserver_count < service_count:
split_count = max_pserver_count split_count = max_pserver_count
block_size = int(math.ceil(var_numel / float(split_count))) block_size = int(math.ceil(var_numel / float(split_count)))
...@@ -270,6 +273,7 @@ class DistributeTranspiler: ...@@ -270,6 +273,7 @@ class DistributeTranspiler:
grad_var_mapping = self._append_split_op(program, grad_blocks) grad_var_mapping = self._append_split_op(program, grad_blocks)
param_var_mapping = self._create_vars_from_blocklist(program, param_var_mapping = self._create_vars_from_blocklist(program,
param_blocks) param_blocks)
# step3: Add gradients as send op inputs and parameters as send # step3: Add gradients as send op inputs and parameters as send
# op outputs. # op outputs.
send_inputs = [] send_inputs = []
...@@ -277,9 +281,11 @@ class DistributeTranspiler: ...@@ -277,9 +281,11 @@ class DistributeTranspiler:
for b in grad_blocks: # append by order for b in grad_blocks: # append by order
varname, block_id, _ = b.split(":") varname, block_id, _ = b.split(":")
send_inputs.append(grad_var_mapping[varname][int(block_id)]) send_inputs.append(grad_var_mapping[varname][int(block_id)])
for b in param_blocks: for b in param_blocks:
varname, block_id, _ = b.split(":") varname, block_id, _ = b.split(":")
send_outputs.append(param_var_mapping[varname][int(block_id)]) send_outputs.append(param_var_mapping[varname][int(block_id)])
# let send_op know which endpoint to send which var to, eplist has the same # let send_op know which endpoint to send which var to, eplist has the same
# order as send_inputs. # order as send_inputs.
eplist = split_method(send_inputs, pserver_endpoints) eplist = split_method(send_inputs, pserver_endpoints)
...@@ -417,7 +423,7 @@ class DistributeTranspiler: ...@@ -417,7 +423,7 @@ class DistributeTranspiler:
def __append_optimize_op__(op, block, grad_to_block_id): def __append_optimize_op__(op, block, grad_to_block_id):
if self._is_opt_op(op): if self._is_opt_op(op):
self._append_pserver_ops(block, op, endpoint, grad_to_block_id, self._append_pserver_ops(block, op, endpoint, grad_to_block_id,
default_main_program()) self.origin_program)
else: else:
self._append_pserver_non_opt_ops(block, op) self._append_pserver_non_opt_ops(block, op)
...@@ -751,9 +757,18 @@ class DistributeTranspiler: ...@@ -751,9 +757,18 @@ class DistributeTranspiler:
Create vars for each split. Create vars for each split.
NOTE: only grads need to be named for different trainers, use NOTE: only grads need to be named for different trainers, use
add_trainer_suffix to rename the grad vars. add_trainer_suffix to rename the grad vars.
:return: A dict mapping from original var name to each var split. Args:
program (ProgramDesc): ProgramDesc which gradients blong.
block_list (list[(varname, block_id, block_size)]): List of gradient blocks.
add_trainer_suffix (Bool): Add trainer suffix to new variable's name if set True.
Returns:
var_mapping (dict(varname->[new_varname_variable])):A dict mapping
from original var name to each var split.
""" """
# varname->[(block_id, current_block_size)]
block_map = dict() block_map = dict()
var_mapping = dict() var_mapping = dict()
for block_str in block_list: for block_str in block_list:
varname, offset, size = block_str.split(":") varname, offset, size = block_str.split(":")
...@@ -824,7 +839,16 @@ class DistributeTranspiler: ...@@ -824,7 +839,16 @@ class DistributeTranspiler:
persistable=persistable) persistable=persistable)
def _append_split_op(self, program, gradblocks): def _append_split_op(self, program, gradblocks):
# Split variables that need to be split and append respective ops """
Split variables that need to be split and append respective ops
Args:
program (ProgramDesc): ProgramDesc that gradients blong.
gradblocks (list[(varname, block_id, block_size)]): List of gradient blocks.
Returns:
var_mapping (dict(varname->[new_splitted_variable])):A dict mapping
from original var name to each var split.
"""
add_suffix = False add_suffix = False
if self.trainer_num > 1: if self.trainer_num > 1:
add_suffix = True add_suffix = True
...@@ -1148,6 +1172,12 @@ class DistributeTranspiler: ...@@ -1148,6 +1172,12 @@ class DistributeTranspiler:
return lr_ops return lr_ops
def _get_optimize_pass(self): def _get_optimize_pass(self):
"""
Get optimizer operators, paramters and gradients from origin_program
Returns:
opt_ops (list): optimize operators.
params_grads (dict): paramter->gradient.
"""
block = self.origin_program.global_block() block = self.origin_program.global_block()
opt_ops = [] opt_ops = []
params_grads = [] params_grads = []
......
...@@ -28,3 +28,38 @@ git clone https://github.com/paddlepaddle/paddle ...@@ -28,3 +28,38 @@ git clone https://github.com/paddlepaddle/paddle
cd paddle/tools/manylinux1 cd paddle/tools/manylinux1
REPO=[yourrepo] ./build_all.sh REPO=[yourrepo] ./build_all.sh
``` ```
## Build PaddlePaddle for the different Python ABIs
Choose one of the following Python ABI and set the correct environment variables.
- cp27-cp27m
```bash
export LD_LIBRARY_PATH=/opt/_internal/cpython-2.7.11-ucs2/lib:${LD_LIBRARY_PATH#/opt/_internal/cpython-2.7.11-ucs4/lib:}
export PATH=/opt/python/cp27-cp27m/bin/:${PATH}
export PYTHON_FLAGS="-DPYTHON_EXECUTABLE:FILEPATH=/opt/python/cp27-cp27m/bin/python
-DPYTHON_INCLUDE_DIR:PATH=/opt/python/cp27-cp27m/include/python2.7
-DPYTHON_LIBRARIES:FILEPATH=/opt/_internal/cpython-2.7.11-ucs2/lib/libpython2.7.so"
```
- cp27-cp27mu
```bash
export LD_LIBRARY_PATH=/opt/_internal/cpython-2.7.11-ucs4/lib:${LD_LIBRARY_PATH#/opt/_internal/cpython-2.7.11-ucs2/lib:}
export PATH=/opt/python/cp27-cp27mu/bin/:${PATH}
export PYTHON_FLAGS="-DPYTHON_EXECUTABLE:FILEPATH=/opt/python/cp27-cp27mu/bin/python
-DPYTHON_INCLUDE_DIR:PATH=/opt/python/cp27-cp27mu/include/python2.7
-DPYTHON_LIBRARIES:FILEPATH=/opt/_internal/cpython-2.7.11-ucs4/lib/libpython2.7.so"
```
And then add the `PYTHON_FLAGS` as your cmake flags:
```bash
cmake ..
${PYTHON_FLAGS} \
-DWITH_GPU=OFF \
...
```
You can find more details about cmake flags at [here](http://www.paddlepaddle.org/docs/develop/documentation/fluid/en/build_and_install/build_from_source_en.html#appendix-build-options)
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册