Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
wzpei9
Paddle
提交
a7866113
P
Paddle
项目概览
wzpei9
/
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,发现更多精彩内容 >>
提交
a7866113
编写于
4月 25, 2018
作者:
F
fengjiayi
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Replace Channel in MultiFileReader with BlockingQueue
上级
1a25f3cd
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
12 addition
and
14 deletion
+12
-14
paddle/fluid/operators/reader/open_files_op.cc
paddle/fluid/operators/reader/open_files_op.cc
+12
-14
未找到文件。
paddle/fluid/operators/reader/open_files_op.cc
浏览文件 @
a7866113
...
...
@@ -14,7 +14,7 @@
#include <thread> // NOLINT
#include "paddle/fluid/
framework/channel
.h"
#include "paddle/fluid/
operators/reader/blocking_queue
.h"
#include "paddle/fluid/operators/reader/reader_op_registry.h"
namespace
paddle
{
...
...
@@ -48,9 +48,9 @@ class MultiFileReader : public framework::ReaderBase {
std
::
thread
scheduler_
;
std
::
vector
<
std
::
thread
>
prefetchers_
;
size_t
buffer_size_
;
framework
::
Channel
<
size_t
>*
waiting_file_idx_
;
framework
::
Channel
<
size_t
>*
available_thread_idx_
;
framework
::
Channel
<
std
::
vector
<
framework
::
LoDTensor
>>*
buffer_
;
reader
::
BlockingQueue
<
size_t
>*
waiting_file_idx_
;
reader
::
BlockingQueue
<
size_t
>*
available_thread_idx_
;
reader
::
BlockingQueue
<
std
::
vector
<
framework
::
LoDTensor
>>*
buffer_
;
};
void
MultiFileReader
::
ReadNext
(
std
::
vector
<
framework
::
LoDTensor
>*
out
)
{
...
...
@@ -73,17 +73,17 @@ bool MultiFileReader::HasNext() {
void
MultiFileReader
::
StartNewScheduler
()
{
size_t
thread_num
=
prefetchers_
.
size
();
waiting_file_idx_
=
framework
::
MakeChannel
<
size_t
>
(
file_names_
.
size
());
available_thread_idx_
=
framework
::
MakeChannel
<
size_t
>
(
thread_num
);
buffer_
=
framework
::
MakeChannel
<
std
::
vector
<
framework
::
LoDTensor
>>
(
buffer_size_
);
waiting_file_idx_
=
new
reader
::
BlockingQueue
<
size_t
>
(
file_names_
.
size
());
available_thread_idx_
=
new
reader
::
BlockingQueue
<
size_t
>
(
thread_num
);
buffer_
=
new
reader
::
BlockingQueue
<
std
::
vector
<
framework
::
LoDTensor
>>
(
buffer_size_
);
for
(
size_t
i
=
0
;
i
<
file_names_
.
size
();
++
i
)
{
waiting_file_idx_
->
Send
(
&
i
);
waiting_file_idx_
->
Send
(
i
);
}
waiting_file_idx_
->
Close
();
for
(
size_t
i
=
0
;
i
<
thread_num
;
++
i
)
{
available_thread_idx_
->
Send
(
&
i
);
available_thread_idx_
->
Send
(
i
);
}
scheduler_
=
std
::
thread
([
this
]
{
ScheduleThreadFunc
();
});
...
...
@@ -149,7 +149,7 @@ void MultiFileReader::PrefetchThreadFunc(std::string file_name,
break
;
}
try
{
buffer_
->
Send
(
&
ins
);
buffer_
->
Send
(
std
::
move
(
ins
)
);
}
catch
(
paddle
::
platform
::
EnforceNotMet
e
)
{
VLOG
(
5
)
<<
"WARNING: The buffer channel has been closed. The prefetch "
"thread of file '"
...
...
@@ -158,9 +158,7 @@ void MultiFileReader::PrefetchThreadFunc(std::string file_name,
}
}
try
{
available_thread_idx_
->
Send
(
&
thread_idx
);
}
catch
(
paddle
::
platform
::
EnforceNotMet
e
)
{
if
(
!
available_thread_idx_
->
Send
(
thread_idx
))
{
VLOG
(
5
)
<<
"WARNING: The available_thread_idx_ channel has been closed. "
"Fail to send thread_idx."
;
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录