Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
OPTHREE
Paddle
提交
9707aa6b
P
Paddle
项目概览
OPTHREE
/
Paddle
与 Fork 源项目一致
Fork自
PaddlePaddle / Paddle
通知
1
Star
1
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
P
Paddle
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
9707aa6b
编写于
5月 16, 2018
作者:
Y
Yu Yang
提交者:
GitHub
5月 16, 2018
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #10619 from reyoung/feature/exec_strategy
Clean code & add execution strategy
上级
1c4bb5c8
999d0fdb
变更
15
隐藏空白更改
内联
并排
Showing
15 changed file
with
304 addition
and
175 deletion
+304
-175
paddle/fluid/framework/details/build_strategy.h
paddle/fluid/framework/details/build_strategy.h
+36
-0
paddle/fluid/framework/details/execution_strategy.h
paddle/fluid/framework/details/execution_strategy.h
+29
-0
paddle/fluid/framework/details/multi_devices_graph_builder.cc
...le/fluid/framework/details/multi_devices_graph_builder.cc
+23
-24
paddle/fluid/framework/details/multi_devices_graph_builder.h
paddle/fluid/framework/details/multi_devices_graph_builder.h
+6
-6
paddle/fluid/framework/details/threaded_ssa_graph_executor.cc
...le/fluid/framework/details/threaded_ssa_graph_executor.cc
+8
-9
paddle/fluid/framework/details/threaded_ssa_graph_executor.h
paddle/fluid/framework/details/threaded_ssa_graph_executor.h
+6
-5
paddle/fluid/framework/parallel_executor.cc
paddle/fluid/framework/parallel_executor.cc
+7
-10
paddle/fluid/framework/parallel_executor.h
paddle/fluid/framework/parallel_executor.h
+20
-17
paddle/fluid/inference/analysis/device.h
paddle/fluid/inference/analysis/device.h
+1
-0
paddle/fluid/inference/analysis/node.h
paddle/fluid/inference/analysis/node.h
+1
-0
paddle/fluid/platform/nccl_helper.h
paddle/fluid/platform/nccl_helper.h
+1
-1
paddle/fluid/pybind/pybind.cc
paddle/fluid/pybind/pybind.cc
+55
-17
python/paddle/fluid/__init__.py
python/paddle/fluid/__init__.py
+29
-27
python/paddle/fluid/parallel_executor.py
python/paddle/fluid/parallel_executor.py
+43
-34
python/paddle/fluid/tests/unittests/test_parallel_executor.py
...on/paddle/fluid/tests/unittests/test_parallel_executor.py
+39
-25
未找到文件。
paddle/fluid/framework/details/build_strategy.h
0 → 100644
浏览文件 @
9707aa6b
// 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
paddle/fluid/framework/details/execution_strategy.h
0 → 100644
浏览文件 @
9707aa6b
// 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
paddle/fluid/framework/details/multi_devices_graph_builder.cc
浏览文件 @
9707aa6b
...
...
@@ -37,31 +37,26 @@ MultiDevSSAGraphBuilder::MultiDevSSAGraphBuilder(
const
std
::
string
&
loss_var_name
,
const
std
::
unordered_set
<
std
::
string
>
&
params
,
const
std
::
vector
<
Scope
*>
&
local_scopes
,
platform
::
NCCLContextMap
*
nccl_ctxs
,
bool
use_default_grad_scale
,
bool
balance_parameter_opt_between_cards
)
platform
::
NCCLContextMap
*
nccl_ctxs
,
const
BuildStrategy
&
strategy
)
:
loss_var_name_
(
loss_var_name
),
places_
(
places
),
local_scopes_
(
local_scopes
),
nccl_ctxs_
(
nccl_ctxs
),
balance_parameter_opt_between_cards_
(
balance_parameter_opt_between_cards
)
{
strategy_
(
strategy
)
{
#else
MultiDevSSAGraphBuilder
::
MultiDevSSAGraphBuilder
(
const
std
::
vector
<
platform
::
Place
>
&
places
,
const
std
::
string
&
loss_var_name
,
const
std
::
unordered_set
<
std
::
string
>
&
params
,
const
std
::
vector
<
Scope
*>
&
local_scopes
,
bool
use_default_grad_scale
,
bool
balance_parameter_opt_between_cards
)
const
std
::
vector
<
Scope
*>
&
local_scopes
,
const
BuildStrategy
&
strategy
)
:
loss_var_name_
(
loss_var_name
),
places_
(
places
),
local_scopes_
(
local_scopes
),
balance_parameter_opt_between_cards_
(
balance_parameter_opt_between_cards
)
{
strategy_
(
strategy
)
{
#endif
for
(
auto
&
p
:
params
)
{
grad_names_
.
insert
(
GradVarName
(
p
));
}
use_default_grad_scale_
=
use_default_grad_scale
;
}
void
MultiDevSSAGraphBuilder
::
CreateOpHandleIOs
(
SSAGraph
*
result
,
...
...
@@ -146,7 +141,8 @@ std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build(
CreateComputationalOps
(
&
result
,
*
op
,
1
);
}
else
if
(
IsScaleLossOp
(
*
op
))
{
// 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
);
}
is_forwarding
=
false
;
...
...
@@ -165,19 +161,22 @@ std::unique_ptr<SSAGraph> MultiDevSSAGraphBuilder::Build(
// broadcast, and each gradient is only broadcast once.
for
(
auto
&
og
:
op
->
OutputArgumentNames
())
{
if
(
IsParameterGradientOnce
(
og
,
&
og_has_been_broadcast
))
{
if
(
balance_parameter_opt_between_cards_
)
{
CreateReduceOp
(
&
result
,
og
,
cur_device_id
);
var_name_on_devices
[
cur_device_id
].
emplace
(
og
);
bcast_var_name_set
[
cur_device_id
].
emplace
(
og
.
substr
(
0
,
og
.
size
()
-
strlen
(
kGradVarSuffix
)));
cur_device_id
=
(
cur_device_id
+
1
)
%
places_
.
size
();
}
else
{
if
(
IsSparseGradient
(
var_types
,
og
))
{
CreateReduceOp
(
&
result
,
og
,
0
);
CreateBroadcastOp
(
&
result
,
og
,
0
);
}
else
{
InsertNCCLAllReduceOp
(
&
result
,
og
);
}
switch
(
strategy_
.
reduce_
)
{
case
BuildStrategy
::
ReduceStrategy
::
kReduce
:
CreateReduceOp
(
&
result
,
og
,
cur_device_id
);
var_name_on_devices
[
cur_device_id
].
emplace
(
og
);
bcast_var_name_set
[
cur_device_id
].
emplace
(
og
.
substr
(
0
,
og
.
size
()
-
strlen
(
kGradVarSuffix
)));
cur_device_id
=
(
cur_device_id
+
1
)
%
places_
.
size
();
break
;
case
BuildStrategy
::
ReduceStrategy
::
kAllReduce
:
if
(
IsSparseGradient
(
var_types
,
og
))
{
CreateReduceOp
(
&
result
,
og
,
0
);
CreateBroadcastOp
(
&
result
,
og
,
0
);
}
else
{
InsertNCCLAllReduceOp
(
&
result
,
og
);
}
break
;
}
}
}
...
...
@@ -303,7 +302,7 @@ bool MultiDevSSAGraphBuilder::IsParameterGradientOnce(
int
MultiDevSSAGraphBuilder
::
GetOpDeviceID
(
const
std
::
vector
<
std
::
unordered_set
<
std
::
string
>>
&
var_name_on_devices
,
const
OpDesc
&
op
)
const
{
if
(
!
balance_parameter_opt_between_cards_
)
{
if
(
strategy_
.
reduce_
!=
BuildStrategy
::
ReduceStrategy
::
kReduce
)
{
return
-
1
;
}
...
...
paddle/fluid/framework/details/multi_devices_graph_builder.h
浏览文件 @
9707aa6b
...
...
@@ -17,6 +17,7 @@
#include <utility>
#include <vector>
#include "paddle/fluid/framework/details/build_strategy.h"
#include "paddle/fluid/framework/details/ssa_graph_builder.h"
namespace
paddle
{
...
...
@@ -36,15 +37,13 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder {
const
std
::
unordered_set
<
std
::
string
>
&
params
,
const
std
::
vector
<
Scope
*>
&
local_scopes
,
platform
::
NCCLContextMap
*
nccl_ctxs
,
bool
use_default_grad_scale
,
bool
balance_parameter_opt_between_cards
);
const
BuildStrategy
&
strategy
);
#else
MultiDevSSAGraphBuilder
(
const
std
::
vector
<
platform
::
Place
>
&
places
,
const
std
::
string
&
loss_var_name
,
const
std
::
unordered_set
<
std
::
string
>
&
params
,
const
std
::
vector
<
Scope
*>
&
local_scopes
,
bool
use_default_grad_scale
,
bool
balance_parameter_opt_between_cards
);
const
BuildStrategy
&
strategy
);
#endif
std
::
unique_ptr
<
SSAGraph
>
Build
(
const
ProgramDesc
&
program
)
const
override
;
...
...
@@ -62,8 +61,6 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder {
#ifdef PADDLE_WITH_CUDA
platform
::
NCCLContextMap
*
nccl_ctxs_
;
#endif
bool
balance_parameter_opt_between_cards_
;
bool
use_default_grad_scale_
;
bool
IsScaleLossOp
(
const
OpDesc
&
op
)
const
;
...
...
@@ -105,6 +102,9 @@ class MultiDevSSAGraphBuilder : public SSAGraphBuilder {
bool
IsSparseGradient
(
const
std
::
unordered_map
<
std
::
string
,
proto
::
VarType
::
Type
>
&
var_types
,
const
std
::
string
&
og
)
const
;
private:
BuildStrategy
strategy_
;
};
}
// namespace details
}
// namespace framework
...
...
paddle/fluid/framework/details/threaded_ssa_graph_executor.cc
浏览文件 @
9707aa6b
...
...
@@ -18,18 +18,17 @@ namespace paddle {
namespace
framework
{
namespace
details
{
ThreadedSSAGraphExecutor
::
ThreadedSSAGraphExecutor
(
size_t
num_threads
,
bool
use_event
,
const
std
::
vector
<
Scope
*>
&
local_scopes
,
const
ExecutionStrategy
&
strategy
,
const
std
::
vector
<
Scope
*>
&
local_scopes
,
const
std
::
vector
<
platform
::
Place
>
&
places
,
std
::
unique_ptr
<
SSAGraph
>
&&
graph
,
bool
allow_op_delay
)
std
::
unique_ptr
<
SSAGraph
>
&&
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
),
places_
(
places
),
fetch_ctxs_
(
places
),
use_event_
(
use_event
),
running_ops_
(
0
),
allow_op_delay_
(
allow_op_dela
y
)
{}
strategy_
(
strateg
y
)
{}
FeedFetchList
ThreadedSSAGraphExecutor
::
Run
(
const
std
::
vector
<
std
::
string
>
&
fetch_tensors
)
{
...
...
@@ -86,7 +85,7 @@ FeedFetchList ThreadedSSAGraphExecutor::Run(
//
// NOTE: DelayedOps have a lower priority. It will be scheduled after all
// 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
);
}
else
{
run_all_ops
(
ready_ops
);
...
...
@@ -113,7 +112,7 @@ FeedFetchList ThreadedSSAGraphExecutor::Run(
auto
&
deps
=
pending_ops
[
op
];
--
deps
;
if
(
deps
==
0
)
{
if
(
op
->
IsMultiDeviceTransfer
()
&&
allow_op_delay_
)
{
if
(
op
->
IsMultiDeviceTransfer
()
&&
strategy_
.
allow_op_delay_
)
{
delayed_ops
.
insert
(
op
);
}
else
{
ready_ops
.
insert
(
op
);
...
...
@@ -191,7 +190,7 @@ void ThreadedSSAGraphExecutor::RunOp(
auto
op_run
=
[
ready_var_q
,
op
,
this
]
{
try
{
VLOG
(
10
)
<<
op
<<
" "
<<
op
->
Name
()
<<
" : "
<<
op
->
DebugString
();
op
->
Run
(
use_event_
);
op
->
Run
(
strategy_
.
use_event_
);
VLOG
(
10
)
<<
op
<<
" "
<<
op
->
Name
()
<<
" Done "
;
running_ops_
--
;
ready_var_q
->
Extend
(
op
->
Outputs
());
...
...
paddle/fluid/framework/details/threaded_ssa_graph_executor.h
浏览文件 @
9707aa6b
...
...
@@ -23,6 +23,7 @@
#include <functional>
#include "ThreadPool.h" // ThreadPool in thrird party
#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/ssa_graph_executor.h"
...
...
@@ -34,11 +35,10 @@ namespace details {
class
ThreadedSSAGraphExecutor
:
public
SSAGraphExecutor
{
public:
ThreadedSSAGraphExecutor
(
size_t
num_threads
,
bool
use_event
,
ThreadedSSAGraphExecutor
(
const
ExecutionStrategy
&
strategy
,
const
std
::
vector
<
Scope
*>
&
local_scopes
,
const
std
::
vector
<
platform
::
Place
>
&
places
,
std
::
unique_ptr
<
SSAGraph
>
&&
graph
,
bool
allow_op_delay
);
std
::
unique_ptr
<
SSAGraph
>
&&
graph
);
// Run a SSAGraph by a thread pool
// Use topological sort algorithm
...
...
@@ -55,10 +55,8 @@ class ThreadedSSAGraphExecutor : public SSAGraphExecutor {
std
::
vector
<
Scope
*>
local_scopes_
;
std
::
vector
<
platform
::
Place
>
places_
;
platform
::
DeviceContextPool
fetch_ctxs_
;
const
bool
use_event_
;
std
::
unique_ptr
<
platform
::
EnforceNotMet
>
exception_
;
std
::
atomic
<
int
>
running_ops_
;
bool
allow_op_delay_
;
void
InsertPendingOp
(
std
::
unordered_map
<
OpHandleBase
*
,
size_t
>
*
pending_ops
,
OpHandleBase
*
op_instance
)
const
;
...
...
@@ -74,6 +72,9 @@ class ThreadedSSAGraphExecutor : public SSAGraphExecutor {
std
::
unordered_map
<
OpHandleBase
*
,
size_t
>
*
pending_ops
,
std
::
unordered_set
<
VarHandleBase
*>
*
pending_vars
,
BlockingQueue
<
VarHandleBase
*>
*
ready_vars
,
FeedFetchList
*
fetch_data
);
private:
ExecutionStrategy
strategy_
;
};
}
// namespace details
...
...
paddle/fluid/framework/parallel_executor.cc
浏览文件 @
9707aa6b
...
...
@@ -52,13 +52,12 @@ std::vector<Scope *> &ParallelExecutor::GetLocalScopes() {
}
ParallelExecutor
::
ParallelExecutor
(
size_t
num_threads
,
bool
use_event
,
const
std
::
vector
<
platform
::
Place
>
&
places
,
const
std
::
unordered_set
<
std
::
string
>
&
params
,
const
std
::
unordered_set
<
std
::
string
>
&
bcast_vars
,
const
ProgramDesc
&
main_program
,
const
std
::
string
&
loss_var_name
,
Scope
*
scope
,
const
std
::
vector
<
Scope
*>
&
local_scopes
,
bool
allow_op_delay
,
bool
use_default_grad_scale
,
bool
balance_parameter_opt_between_cards
,
Scope
*
scope
,
const
std
::
vector
<
Scope
*>
&
local_scopes
,
const
ExecutionStrategy
&
exec_strategy
,
const
BuildStrategy
&
build_strategy
,
size_t
num_trainers
,
size_t
trainer_id
)
:
member_
(
new
ParallelExecutorPrivate
(
places
))
{
member_
->
global_scope_
=
scope
;
...
...
@@ -100,18 +99,16 @@ ParallelExecutor::ParallelExecutor(
#ifdef PADDLE_WITH_CUDA
details
::
MultiDevSSAGraphBuilder
builder
(
member_
->
places_
,
loss_var_name
,
params
,
member_
->
local_scopes_
,
member_
->
nccl_ctxs_
.
get
(),
use_default_grad_scale
,
balance_parameter_opt_between_cards
);
member_
->
nccl_ctxs_
.
get
(),
build_strategy
);
#else
details
::
MultiDevSSAGraphBuilder
builder
(
member_
->
places_
,
loss_var_name
,
params
,
member_
->
local_scopes_
,
use_default_grad_scale
,
balance_parameter_opt_between_cards
);
details
::
MultiDevSSAGraphBuilder
builder
(
member_
->
places_
,
loss_var_name
,
params
,
member_
->
local_scopes_
,
build_strategy
);
#endif
auto
graph
=
builder
.
Build
(
main_program
);
member_
->
executor_
.
reset
(
new
details
::
ThreadedSSAGraphExecutor
(
num_threads
,
use_event
,
member_
->
local_scopes_
,
places
,
std
::
move
(
graph
),
allow_op_delay
));
exec_strategy
,
member_
->
local_scopes_
,
places
,
std
::
move
(
graph
)));
// Step 3. Create vars in each scope;
for
(
auto
*
var
:
main_program
.
Block
(
0
).
AllVars
())
{
...
...
paddle/fluid/framework/parallel_executor.h
浏览文件 @
9707aa6b
...
...
@@ -14,57 +14,60 @@ limitations under the License. */
#pragma once
#include <paddle/fluid/framework/details/build_strategy.h>
#include <string>
#include <unordered_set>
#include <vector>
#include "paddle/fluid/framework/details/execution_strategy.h"
#include "paddle/fluid/framework/executor.h"
#include "paddle/fluid/framework/op_info.h"
#include "paddle/fluid/framework/program_desc.h"
#include "paddle/fluid/framework/scope.h"
#include "paddle/fluid/framework/tensor.h"
#include "paddle/fluid/platform/device_context.h"
namespace
paddle
{
namespace
framework
{
class
ParallelExecutorPrivate
;
using
details
::
BuildStrategy
;
using
details
::
ExecutionStrategy
;
class
ParallelExecutor
{
DISABLE_COPY_AND_ASSIGN
(
ParallelExecutor
);
public:
explicit
ParallelExecutor
(
size_t
num_threads
,
bool
use_event
,
const
std
::
vector
<
platform
::
Place
>&
places
,
const
std
::
unordered_set
<
std
::
string
>&
params
,
const
std
::
unordered_set
<
std
::
string
>&
bcast_vars
,
const
ProgramDesc
&
main_program
,
const
std
::
string
&
loss_var_name
,
Scope
*
scope
,
const
std
::
vector
<
Scope
*>&
local_scopes
,
bool
allow_op_delay
,
bool
use_default_grad_scale
,
bool
balance_parameter_opt_between_cards
,
explicit
ParallelExecutor
(
const
std
::
vector
<
platform
::
Place
>
&
places
,
const
std
::
unordered_set
<
std
::
string
>
&
params
,
const
std
::
unordered_set
<
std
::
string
>
&
bcast_vars
,
const
ProgramDesc
&
main_program
,
const
std
::
string
&
loss_var_name
,
Scope
*
scope
,
const
std
::
vector
<
Scope
*>
&
local_scopes
,
const
ExecutionStrategy
&
exec_strategy
,
const
BuildStrategy
&
build_strategy
,
size_t
num_trainers
=
1
,
size_t
trainer_id
=
0
);
~
ParallelExecutor
();
std
::
vector
<
Scope
*>&
GetLocalScopes
();
std
::
vector
<
Scope
*>
&
GetLocalScopes
();
/**
* Feed tensors to local scopes. The size of tensors should be equal to the
* size of local scopes.
*/
void
FeedTensorsIntoLocalScopes
(
const
std
::
vector
<
std
::
unordered_map
<
std
::
string
,
LoDTensor
>>
&
tensors
);
const
std
::
vector
<
std
::
unordered_map
<
std
::
string
,
LoDTensor
>>
&
tensors
);
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
,
const
std
::
string
&
fetched_var_name
);
void
Run
(
const
std
::
vector
<
std
::
string
>
&
fetch_tensors
,
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:
ParallelExecutorPrivate
*
member_
;
ParallelExecutorPrivate
*
member_
;
};
}
// namespace framework
...
...
paddle/fluid/inference/analysis/device.h
浏览文件 @
9707aa6b
...
...
@@ -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.
See the License for the specific language governing permissions and
limitations under the License. */
#pragma once
namespace
paddle
{
namespace
inference
{
...
...
paddle/fluid/inference/analysis/node.h
浏览文件 @
9707aa6b
...
...
@@ -19,6 +19,7 @@ limitations under the License. */
*/
#pragma once
#include <limits>
#include <memory>
#include <string>
#include <unordered_map>
...
...
paddle/fluid/platform/nccl_helper.h
浏览文件 @
9707aa6b
...
...
@@ -53,7 +53,7 @@ class NCCLGroupGuard {
}
inline
~
NCCLGroupGuard
()
{
PADDLE_ENFORCE
(
dynload
::
ncclGroupEnd
()
);
CHECK_EQ
(
dynload
::
ncclGroupEnd
(),
ncclSuccess
);
NCCLMutex
().
unlock
();
}
};
...
...
paddle/fluid/pybind/pybind.cc
浏览文件 @
9707aa6b
...
...
@@ -494,23 +494,61 @@ All parameter, weight, gradient are variables in Paddle.
m
.
def
(
"disable_profiler"
,
platform
::
DisableProfiler
);
m
.
def
(
"reset_profiler"
,
platform
::
ResetProfiler
);
py
::
class_
<
ParallelExecutor
>
(
m
,
"ParallelExecutor"
)
.
def
(
"__init__"
,
[](
ParallelExecutor
&
self
,
size_t
num_threads
,
bool
use_event
,
const
std
::
vector
<
platform
::
Place
>
&
places
,
const
std
::
unordered_set
<
std
::
string
>
&
params
,
const
std
::
unordered_set
<
std
::
string
>
&
bcast_vars
,
const
ProgramDesc
&
main_program
,
const
std
::
string
&
loss_var_name
,
Scope
*
scope
,
std
::
vector
<
Scope
*>
&
local_scopes
,
bool
allow_op_delay
,
bool
use_default_grad_scale
,
bool
balance_parameter_opt_between_cards
,
size_t
num_trainers
,
size_t
trainer_id
)
{
new
(
&
self
)
ParallelExecutor
(
num_threads
,
use_event
,
places
,
params
,
bcast_vars
,
main_program
,
loss_var_name
,
scope
,
local_scopes
,
allow_op_delay
,
use_default_grad_scale
,
balance_parameter_opt_between_cards
,
num_trainers
,
trainer_id
);
})
// -- python binds for parallel executor.
py
::
class_
<
ParallelExecutor
>
pe
(
m
,
"ParallelExecutor"
);
py
::
class_
<
ExecutionStrategy
>
(
pe
,
"ExecutionStrategy"
)
.
def
(
py
::
init
())
.
def_property
(
"num_threads"
,
[](
const
ExecutionStrategy
&
self
)
{
return
self
.
num_threads_
;
},
[](
ExecutionStrategy
&
self
,
size_t
num_threads
)
{
self
.
num_threads_
=
num_threads
;
})
.
def_property
(
"use_event"
,
[](
const
ExecutionStrategy
&
self
)
{
return
self
.
use_event_
;
},
[](
ExecutionStrategy
&
self
,
bool
use_event
)
{
self
.
use_event_
=
use_event
;
})
.
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
)
// NOTE: even we return a vec<Scope*>* to Python use reference policy.
// We still cannot get local_scope from this vector, since the element
...
...
python/paddle/fluid/__init__.py
浏览文件 @
9707aa6b
...
...
@@ -44,42 +44,44 @@ import transpiler
from
param_attr
import
ParamAttr
,
WeightNormParamAttr
from
data_feeder
import
DataFeeder
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
,
channel_close
,
Select
)
import
clip
import
profiler
import
unique_name
import
recordio_writer
from
parallel_executor
import
ParallelExecutor
import
parallel_executor
from
parallel_executor
import
*
Tensor
=
LoDTensor
__all__
=
framework
.
__all__
+
executor
.
__all__
+
concurrency
.
__all__
+
\
trainer
.
__all__
+
inferencer
.
__all__
+
transpiler
.
__all__
+
[
'io'
,
'initializer
'
,
'layers
'
,
'transpiler'
'nets'
,
'optimizer
'
,
'learning_rate_decay
'
,
'backward
'
,
'regularizer
'
,
'LoDTenso
r'
,
'CPUPlace
'
,
'CUDA
Place'
,
'CUDAPinned
Place'
,
'Tensor
'
,
'ParamAtt
r'
,
'WeightNorm
ParamAttr'
,
'DataFeede
r'
,
'clip
'
,
'profiler
'
,
'unique_name
'
,
'recordio_writer
'
,
'ParallelExecuto
r'
,
]
__all__
=
framework
.
__all__
+
executor
.
__all__
+
concurrency
.
__all__
+
\
trainer
.
__all__
+
inferencer
.
__all__
+
transpiler
.
__all__
+
\
parallel_executor
.
__all__
+
[
'io
'
,
'initializer
'
,
'layers'
,
'transpiler'
'nets
'
,
'optimizer
'
,
'learning_rate_decay
'
,
'backward
'
,
'regularize
r'
,
'LoDTensor
'
,
'CPU
Place'
,
'CUDA
Place'
,
'CUDAPinnedPlace
'
,
'Tenso
r'
,
'
ParamAttr'
,
'WeightNormParamAtt
r'
,
'DataFeeder
'
,
'clip
'
,
'profiler
'
,
'unique_name
'
,
'recordio_write
r'
,
]
def
__bootstrap__
():
...
...
python/paddle/fluid/parallel_executor.py
浏览文件 @
9707aa6b
...
...
@@ -19,7 +19,10 @@ import executor
import
warnings
import
sys
__all__
=
[
'ParallelExecutor'
]
__all__
=
[
'ParallelExecutor'
,
'ExecutionStrategy'
,
'BuildStrategy'
]
ExecutionStrategy
=
core
.
ParallelExecutor
.
ExecutionStrategy
BuildStrategy
=
core
.
ParallelExecutor
.
BuildStrategy
class
ParallelExecutor
(
object
):
...
...
@@ -27,13 +30,12 @@ class ParallelExecutor(object):
use_cuda
,
loss_name
=
None
,
main_program
=
None
,
num_threads
=
None
,
allow_op_delay
=
False
,
share_vars_from
=
None
,
use_default_grad_scale
=
Tru
e
,
b
alance_parameter_opt_between_cards
=
Fals
e
,
exec_strategy
=
Non
e
,
b
uild_strategy
=
Non
e
,
num_trainers
=
1
,
trainer_id
=
0
):
trainer_id
=
0
,
**
kwargs
):
"""
ParallelExecutor can run program in parallel.
...
...
@@ -42,21 +44,8 @@ class ParallelExecutor(object):
loss_name(str, default None): The loss name must set in training.
main_program(Program, default None): The program that need to run,
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,
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
initialized with multpile rank of nodes, each node should have
same number of GPUs. Distributed training will be enabled then.
...
...
@@ -83,6 +72,25 @@ class ParallelExecutor(object):
train_loss, = train_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
.
_act_places
=
[]
...
...
@@ -100,15 +108,25 @@ class ParallelExecutor(object):
self
.
_places
.
append
(
p
)
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
:
# Experiments on se-resnext shows that too many threads hurt
# performance. Worth tunning for other models in the future.
num_threads
=
len
(
self
.
_places
)
*
2
exec_strategy
.
num_threads
=
len
(
self
.
_places
)
*
2
else
:
num_threads
=
min
(
exec_strategy
.
num_threads
=
min
(
len
(
self
.
_places
)
*
2
,
multiprocessing
.
cpu_count
())
if
build_strategy
is
None
:
build_strategy
=
BuildStrategy
()
main
=
main_program
main
=
main
if
main
else
framework
.
default_main_program
()
scope
=
executor
.
global_scope
()
...
...
@@ -127,23 +145,14 @@ class ParallelExecutor(object):
]
self
.
executor
=
core
.
ParallelExecutor
(
num_threads
,
True
if
use_cuda
else
False
,
# use_event
self
.
_places
,
set
([
p
.
name
for
p
in
main
.
global_block
().
iter_parameters
()
if
not
p
.
stop_gradient
]),
set
(
self
.
persistable_vars
),
main
.
desc
,
loss_name
if
loss_name
else
''
,
scope
,
local_scopes
,
allow_op_delay
,
use_default_grad_scale
,
balance_parameter_opt_between_cards
,
num_trainers
,
trainer_id
)
set
(
self
.
persistable_vars
),
main
.
desc
,
loss_name
if
loss_name
else
''
,
scope
,
local_scopes
,
exec_strategy
,
build_strategy
,
num_trainers
,
trainer_id
)
self
.
scope
=
scope
def
run
(
self
,
fetch_list
,
feed
=
None
,
feed_dict
=
None
):
...
...
python/paddle/fluid/tests/unittests/test_parallel_executor.py
浏览文件 @
9707aa6b
...
...
@@ -232,14 +232,18 @@ class TestParallelExecutorBase(unittest.TestCase):
place
=
fluid
.
CUDAPlace
(
0
)
startup_exe
=
fluid
.
Executor
(
place
)
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
:
exe
=
fluid
.
ParallelExecutor
(
True
,
loss_name
=
loss
.
name
,
allow_op_delay
=
allow_op_delay
,
balance_parameter_opt_between_cards
=
balance_parameter_opt_between_cards
)
exec_strategy
=
exec_strategy
,
build_strategy
=
build_strategy
)
else
:
exe
=
fluid
.
Executor
(
place
=
place
)
...
...
@@ -548,7 +552,7 @@ class TestTransformer(TestParallelExecutorBase):
class
ParallelExecutorTestingDuringTraining
(
unittest
.
TestCase
):
def
check_network_convergence
(
self
,
b
alance_parameter_opt_between_cards
):
def
check_network_convergence
(
self
,
b
uild_strategy
=
None
):
main
=
fluid
.
Program
()
startup
=
fluid
.
Program
()
with
fluid
.
program_guard
(
main
,
startup
):
...
...
@@ -571,15 +575,13 @@ class ParallelExecutorTestingDuringTraining(unittest.TestCase):
use_cuda
=
True
,
loss_name
=
loss
.
name
,
main_program
=
main
,
balance_parameter_opt_between_cards
=
balance_parameter_opt_between_cards
)
build_strategy
=
build_strategy
)
test_exe
=
fluid
.
ParallelExecutor
(
use_cuda
=
True
,
main_program
=
test_program
,
share_vars_from
=
train_exe
,
balance_parameter_opt_between_cards
=
balance_parameter_opt_between_cards
)
build_strategy
=
build_strategy
)
for
i
in
xrange
(
5
):
test_loss
,
=
test_exe
.
run
([
loss
.
name
],
feed
=
feed_dict
)
...
...
@@ -594,10 +596,14 @@ class ParallelExecutorTestingDuringTraining(unittest.TestCase):
str
(
test_loss
))
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
):
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
...
...
@@ -617,7 +623,7 @@ embedding_name = 'emb'
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
predicate_embedding
=
fluid
.
layers
.
embedding
(
input
=
predicate
,
...
...
@@ -686,9 +692,7 @@ def db_lstm(word, predicate, ctx_n2, ctx_n1, ctx_0, ctx_p1, ctx_p2, mark,
class
TestCRFModel
(
unittest
.
TestCase
):
def
check_network_convergence
(
self
,
is_sparse
,
balance_parameter_opt_between_cards
=
False
):
def
check_network_convergence
(
self
,
is_sparse
,
build_strategy
=
None
):
main
=
fluid
.
Program
()
startup
=
fluid
.
Program
()
with
fluid
.
program_guard
(
main
,
startup
):
...
...
@@ -739,8 +743,7 @@ class TestCRFModel(unittest.TestCase):
pe
=
fluid
.
ParallelExecutor
(
use_cuda
=
True
,
loss_name
=
avg_cost
.
name
,
balance_parameter_opt_between_cards
=
balance_parameter_opt_between_cards
)
build_strategy
=
build_strategy
)
feeder
=
fluid
.
DataFeeder
(
feed_list
=
[
...
...
@@ -756,19 +759,29 @@ class TestCRFModel(unittest.TestCase):
pe
.
run
(
feed
=
feeder
.
feed
(
cur_batch
),
fetch_list
=
[
avg_cost
.
name
]))[
0
]
def
test_update_sparse_parameter
(
self
):
self
.
check_network_convergence
(
is_sparse
=
True
)
def
test_update_sparse_parameter_all_reduce
(
self
):
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
):
self
.
check_network_convergence
(
is_sparse
=
False
)
def
test_update_dense_parameter_all_reduce
(
self
):
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
(
is_sparse
=
False
,
b
alance_parameter_opt_between_cards
=
True
)
is_sparse
=
False
,
b
uild_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
(
is_sparse
=
False
,
b
alance_parameter_opt_between_cards
=
True
)
is_sparse
=
False
,
b
uild_strategy
=
build_strategy
)
# test fetch all the variables of global_block
...
...
@@ -836,7 +849,8 @@ class TestFetchOp(unittest.TestCase):
assert
not
math
.
isnan
(
np
.
sum
(
ret
[
i
]))
and
\
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_iter
=
tst_reader
()
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录