Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
溜达小灰鼠
Paddle
提交
43771ad4
P
Paddle
项目概览
溜达小灰鼠
/
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,发现更多精彩内容 >>
提交
43771ad4
编写于
6月 22, 2017
作者:
Z
zlx
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'develop' of
https://github.com/PaddlePaddle/Paddle
into improve_pruning
上级
a266292a
65aa8941
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
66 addition
and
11 deletion
+66
-11
python/paddle/v2/dataset/common.py
python/paddle/v2/dataset/common.py
+11
-7
python/paddle/v2/reader/decorator.py
python/paddle/v2/reader/decorator.py
+33
-4
python/paddle/v2/reader/tests/decorator_test.py
python/paddle/v2/reader/tests/decorator_test.py
+22
-0
未找到文件。
python/paddle/v2/dataset/common.py
浏览文件 @
43771ad4
...
...
@@ -27,13 +27,17 @@ __all__ = ['DATA_HOME', 'download', 'md5file', 'split', 'cluster_files_reader']
DATA_HOME
=
os
.
path
.
expanduser
(
'~/.cache/paddle/dataset'
)
if
not
os
.
path
.
exists
(
DATA_HOME
):
try
:
os
.
makedirs
(
DATA_HOME
)
except
OSError
as
exc
:
if
exc
.
errno
!=
errno
.
EEXIST
:
raise
pass
# When running unit tests, there could be multiple processes that
# trying to create DATA_HOME directory simultaneously, so we cannot
# use a if condition to check for the existence of the directory;
# instead, we use the filesystem as the synchronization mechanism by
# catching returned errors.
try
:
os
.
makedirs
(
DATA_HOME
)
except
OSError
as
exc
:
if
exc
.
errno
!=
errno
.
EEXIST
:
raise
pass
def
md5file
(
fname
):
...
...
python/paddle/v2/reader/decorator.py
浏览文件 @
43771ad4
...
...
@@ -230,7 +230,7 @@ class XmapEndSignal():
pass
def
xmap_readers
(
mapper
,
reader
,
process_num
,
buffer_size
):
def
xmap_readers
(
mapper
,
reader
,
process_num
,
buffer_size
,
order
=
False
):
"""
Use multiprocess to map samples from reader by a mapper defined by user.
And this function contains a buffered decorator.
...
...
@@ -242,12 +242,15 @@ def xmap_readers(mapper, reader, process_num, buffer_size):
:type process_num: int
:param buffer_size: max buffer size
:type buffer_size: int
:param order: keep the order of reader
:type order: bool
:return: the decarated reader
:rtype: callable
"""
end
=
XmapEndSignal
()
in_queue
=
Queue
(
buffer_size
)
out_queue
=
Queue
(
buffer_size
)
out_order
=
[
0
]
# define a worker to read samples from reader to in_queue
def
read_worker
(
reader
,
in_queue
):
...
...
@@ -255,8 +258,17 @@ def xmap_readers(mapper, reader, process_num, buffer_size):
in_queue
.
put
(
i
)
in_queue
.
put
(
end
)
# define a worker to read samples from reader to in_queue with order flag
def
order_read_worker
(
reader
,
in_queue
):
in_order
=
0
for
i
in
reader
():
in_queue
.
put
((
in_order
,
i
))
in_order
+=
1
in_queue
.
put
(
end
)
# start a read worker in a thread
t
=
Thread
(
target
=
read_worker
,
args
=
(
reader
,
in_queue
))
target
=
order_read_worker
if
order
else
read_worker
t
=
Thread
(
target
=
target
,
args
=
(
reader
,
in_queue
))
t
.
daemon
=
True
t
.
start
()
...
...
@@ -271,11 +283,28 @@ def xmap_readers(mapper, reader, process_num, buffer_size):
in_queue
.
put
(
end
)
out_queue
.
put
(
end
)
# define a worker to handle samples from in_queue by mapper
# and put mapped samples into out_queue by order
def
order_handle_worker
(
in_queue
,
out_queue
,
mapper
,
out_order
):
ins
=
in_queue
.
get
()
while
not
isinstance
(
ins
,
XmapEndSignal
):
order
,
sample
=
ins
r
=
mapper
(
sample
)
while
order
!=
out_order
[
0
]:
pass
out_queue
.
put
(
r
)
out_order
[
0
]
+=
1
ins
=
in_queue
.
get
()
in_queue
.
put
(
end
)
out_queue
.
put
(
end
)
# start several handle_workers
target
=
order_handle_worker
if
order
else
handle_worker
args
=
(
in_queue
,
out_queue
,
mapper
,
out_order
)
if
order
else
(
in_queue
,
out_queue
,
mapper
)
workers
=
[]
for
i
in
xrange
(
process_num
):
worker
=
Thread
(
target
=
handle_worker
,
args
=
(
in_queue
,
out_queue
,
mapper
))
worker
=
Thread
(
target
=
target
,
args
=
args
)
worker
.
daemon
=
True
workers
.
append
(
worker
)
for
w
in
workers
:
...
...
python/paddle/v2/reader/tests/decorator_test.py
浏览文件 @
43771ad4
...
...
@@ -121,5 +121,27 @@ class TestShuffle(unittest.TestCase):
self
.
assertEqual
(
total
,
10
)
class
TestXmap
(
unittest
.
TestCase
):
def
test_xmap
(
self
):
def
mapper
(
x
):
return
(
x
+
1
)
orders
=
(
True
,
False
)
thread_nums
=
(
1
,
2
,
4
,
8
,
16
)
buffered_size
=
(
1
,
2
,
4
,
8
,
16
)
for
order
in
orders
:
for
tNum
in
thread_nums
:
for
size
in
buffered_size
:
result
=
[]
for
i
in
paddle
.
v2
.
reader
.
xmap_readers
(
mapper
,
reader_creator_10
(
0
),
tNum
,
size
,
order
)():
result
.
append
(
i
)
if
not
order
:
result
.
sort
()
for
idx
,
e
in
enumerate
(
result
):
self
.
assertEqual
(
e
,
mapper
(
idx
))
if
__name__
==
'__main__'
:
unittest
.
main
()
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录