Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
Crayon鑫
Paddle
提交
869ef01d
P
Paddle
项目概览
Crayon鑫
/
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看板
提交
869ef01d
编写于
3月 29, 2018
作者:
F
fengjiayi
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'develop' of
https://github.com/PaddlePaddle/Paddle
into kernel_of_increment_op
上级
52574733
241f3c98
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
142 addition
and
4 deletion
+142
-4
doc/fluid/design/concurrent/channel.md
doc/fluid/design/concurrent/channel.md
+139
-0
doc/fluid/design/concurrent/images/channel_recv.png
doc/fluid/design/concurrent/images/channel_recv.png
+0
-0
doc/fluid/design/concurrent/images/channel_send.png
doc/fluid/design/concurrent/images/channel_send.png
+0
-0
paddle/fluid/framework/CMakeLists.txt
paddle/fluid/framework/CMakeLists.txt
+1
-1
python/paddle/fluid/layer_helper.py
python/paddle/fluid/layer_helper.py
+2
-3
未找到文件。
doc/fluid/design/concurrent/channel.md
0 → 100644
浏览文件 @
869ef01d
# Channel Design
## Introduction
A Channel is a data structure that allows for synchronous interprocess
communication via message passing. It is a fundemental component of CSP
(communicating sequential processes), and allows for users to pass data
between threads without having to worry about synchronization.
## How to use it
Paddle offers python APIs to open and close channels, along with sending
and receiving data to/from a channel.
### Create a channel
Creates a new channel that takes in variables of a specific dtype.
-
**fluid.make_channel(dtype, capacity=0)**
-
**dtype**
: The data type of variables being sent/received through channel
-
**capacity**
: The capacity of the channel. A capacity of 0 represents
an unbuffered channel. Capacity > 0 represents a buffered channel
```
ch = fluid.make_channel(dtype=core.VarDesc.VarType.LOD_TENSOR, 10)
```
### Close a channel
Closes a channel. Any pending senders and receivers will be awoken during
this time. Receivers can still receive from a closed channel, but senders
are not allowed to send any additional data to the channel (Paddle will
raise an exception if users try to send to a closed channel.)
-
**fluid.channel_close(channel)**
```
fluid.channel_close(ch)
```
### Send data to a channel
Sends a variable to a channel. Currently, variables of dtype
`LoDTensor`
,
`LoDRankTable`
,
`LoDTensorArray`
,
`SelectedRows`
,
`ReaderHolder`
, and
`ChannelHolder`
are supported.
By default, the data of the Variable is moved from the sender to the receiver,
however the user can optionally copy the data before performing the send.
-
**channel_send(channel, variable, is_copy=False)**
-
**channel**
: The channel to send the variable to
-
**variable**
: The variable to send to the channel
-
**is_copy**
: If set to True, channel_send will perform a variable assign
to copy the source variable to a new variable to be sent.
```
ch = fluid.make_channel(dtype=core.VarDesc.VarType.LOD_TENSOR)
var = fill_constant(shape=[1],dtype=core.VarDesc.VarType.INT32, value=100)
fluid.channel_send(ch, var, True)
```
### Receive data from a channel
Receives a variable from a channel. The data of the variable is moved to the
receiving variable.
-
**channel_recv(channel, return_variable)**
-
**channel**
: The channel to receive the variable from
-
**return_variable**
: The destination variable used to store the data of the
variable received from the channel
```
ch = fluid.make_channel(dtype=core.VarDesc.VarType.LOD_TENSOR)
var = fill_constant(shape=[1],dtype=core.VarDesc.VarType.INT32, value=-1)
fluid.channel_recv(ch, var)
```
## How it Works
Channels provides a simple interface for different threads to share data.
To support the synchronization requirements, channels utilizes a series of
internal queues, locks, and conditional variables.
### QueueMessage
QueueMessage encapsulates the state of the channel send/receive operation to be
put in the
**sendq/recvq**
. It contains a condition variable used to lock the
thread (when there are no available sends/receives). In addition, it contains
a callback function to notify a thread when the QueueMessage is being
processed by the channel.
### Queues
-
**buff_**
: This queue holds the data buffer in a buffered channel. The
capacity is set to the capacity of the channel. This data buffer is not
used in an unbuffered channel.
-
**sendq**
: This queue holds the QueueMessage of any pending senders of a
channel. When a thread performs a channel_send operation on the channel, the
channel_send operation will put a new QueueMessage on the sendq and block the
current thread under two conditions:
1.
The channel is buffered and is full
2.
The channel is unbuffered and does not have a receiver
-
**recvq**
: This queue holds the QueueMessage of any pending receivers of a
channel. When a thread performs a channel_recv operation on the channel, the
channel_recv operation will put a new QueueMessage on the recvq and block the
current thread under two conditions:
1.
The channel is buffered and there is no data on the buff_
2.
The channel is unbuffered and does not have a sender
### State diagram
#### Channel Send
<p
align=
"center"
>
<img
src=
"./images/channel_send.png"
/><br/>
</p>
#### Channel Receive
<p
align=
"center"
>
<img
src=
"./images/channel_recv.png"
/><br/>
</p>
## Limitations and Considerations
### Variable Copy
In golang, variables in channels are copied from the sender to the receiver.
In Paddle, the data from our variables are
**moved**
from sender to receiver.
As a result, these variables should not be used after they are sent. We
provide a flag in channel_send method to allow users to copy the variable to
be sent before it is sent.
Please note that this is acheived by adding an
**assign**
operator and creating
a temporary variable that is sent in place of the original variable. Please
note that
**assign**
operator has limited support for only certain variables
datatypes.
doc/fluid/design/concurrent/images/channel_recv.png
0 → 100644
浏览文件 @
869ef01d
133.4 KB
doc/fluid/design/concurrent/images/channel_send.png
0 → 100644
浏览文件 @
869ef01d
83.6 KB
paddle/fluid/framework/CMakeLists.txt
浏览文件 @
869ef01d
...
@@ -100,7 +100,7 @@ cc_test(init_test SRCS init_test.cc DEPS init)
...
@@ -100,7 +100,7 @@ cc_test(init_test SRCS init_test.cc DEPS init)
cc_test
(
op_kernel_type_test SRCS op_kernel_type_test.cc DEPS place device_context framework_proto
)
cc_test
(
op_kernel_type_test SRCS op_kernel_type_test.cc DEPS place device_context framework_proto
)
cc_test
(
cow_ptr_tests SRCS details/cow_ptr_test.cc
)
cc_test
(
cow_ptr_tests SRCS details/cow_ptr_test.cc
)
cc_test
(
channel_test SRCS channel_test.cc
)
#
cc_test(channel_test SRCS channel_test.cc)
cc_test
(
tuple_test SRCS tuple_test.cc
)
cc_test
(
tuple_test SRCS tuple_test.cc
)
cc_test
(
concurrency_test SRCS concurrency_test.cc DEPS go_op channel_close_op channel_create_op
cc_test
(
concurrency_test SRCS concurrency_test.cc DEPS go_op channel_close_op channel_create_op
channel_send_op channel_recv_op sum_op select_op elementwise_add_op compare_op
channel_send_op channel_recv_op sum_op select_op elementwise_add_op compare_op
...
...
python/paddle/fluid/layer_helper.py
浏览文件 @
869ef01d
...
@@ -398,7 +398,6 @@ class LayerHelper(object):
...
@@ -398,7 +398,6 @@ class LayerHelper(object):
return
input_var
return
input_var
if
isinstance
(
act
,
basestring
):
if
isinstance
(
act
,
basestring
):
act
=
{
'type'
:
act
}
act
=
{
'type'
:
act
}
tmp
=
self
.
create_tmp_variable
(
dtype
=
input_var
.
dtype
)
if
'use_mkldnn'
in
self
.
kwargs
:
if
'use_mkldnn'
in
self
.
kwargs
:
act
[
'use_mkldnn'
]
=
self
.
kwargs
.
get
(
'use_mkldnn'
)
act
[
'use_mkldnn'
]
=
self
.
kwargs
.
get
(
'use_mkldnn'
)
...
@@ -408,9 +407,9 @@ class LayerHelper(object):
...
@@ -408,9 +407,9 @@ class LayerHelper(object):
self
.
append_op
(
self
.
append_op
(
type
=
act_type
,
type
=
act_type
,
inputs
=
{
"X"
:
[
input_var
]},
inputs
=
{
"X"
:
[
input_var
]},
outputs
=
{
"Out"
:
[
tmp
]},
outputs
=
{
"Out"
:
[
input_var
]},
attrs
=
act
)
attrs
=
act
)
return
tmp
return
input_var
def
_get_default_initializer
(
self
,
dtype
):
def
_get_default_initializer
(
self
,
dtype
):
if
dtype
is
None
or
dtype_is_floating
(
dtype
)
is
True
:
if
dtype
is
None
or
dtype_is_floating
(
dtype
)
is
True
:
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录