YSTest: concurrency.h 源文件

YSTest PreAlpha_b700_20160611 The YSLib Test Project

- 首页
- 相关页面
- 模块
- 命名空间
- 类
- 文件

- 文件列表
- 文件成员

concurrency.h
浏览该文件的文档.
1 /*
2 © 2014-2015 FrankHB.
3
4 This file is part of the YSLib project, and may only be used,
5 modified, and distributed under the terms of the YSLib project
6 license, LICENSE.TXT. By continuing to use, modify, or distribute
7 this file you indicate that you have read the license and
8 understand and accept it fully.
9 */
10
28 #ifndef YB_INC_ystdex_concurrency_h_
29 #define YB_INC_ystdex_concurrency_h_ 1
30
31 #include "pseudo_mutex.h" // for result_of_t, std::declval, std::make_shared,
32 // threading::unlock_delete;
33 #include <mutex> // for std::mutex, std::unique_lock;
34 #include <thread> // for std::thread;
35 #include <vector> // for std::vector;
36 #include <queue> // for std::queue;
37 #include "future.hpp" // for std::packaged_task, future_result_t, pack_task;
38 #include <condition_variable> // for std::condition_variable;
39 #include "functional.hpp" // for std::bind, std::function, ystdex::invoke;
40 #include "cassert.h" // for yassume;
41
42 namespace ystdex
43 {
44
50 template<class _tMutex = std::mutex,
51 class _tLock = std::unique_lock<_tMutex>>
52 using unlock_delete = threading::unlock_delete<_tMutex, _tLock>;
53
54
60 template<typename _type, class _tMutex = std::mutex,
61 class _tLock = std::unique_lock<_tMutex>>
62 using locked_ptr = threading::locked_ptr<_type, _tMutex, _tLock>;
63
64
70 YB_API std::string
71 to_string(const std::thread::id&);
72
78 inline std::string
79 get_this_thread_id()
80 {
81 return ystdex::to_string(std::this_thread::get_id());
82 }
83
84
85 # if !__GLIBCXX__ || (defined(_GLIBCXX_HAS_GTHREADS) \
86 && defined(_GLIBCXX_USE_C99_STDINT_TR1) && (ATOMIC_INT_LOCK_FREE > 1))
87
93 class YB_API thread_pool
94 {
95 private:
96 std::vector<std::thread> workers;
98 std::queue<std::packaged_task<void()>> tasks{};
99 mutable std::mutex queue_mutex{};
100 std::condition_variable condition{};
106 bool stopped = {};
107
108 public:
115 thread_pool(size_t, std::function<void()> = {}, std::function<void()> = {});
123 ~thread_pool() ynothrow;
124
129 template<typename _fCallable, typename... _tParams>
130 future_result_t<_fCallable, _tParams...>
131 enqueue(_fCallable&& f, _tParams&&... args)
132 {
133 return wait_to_enqueue([](std::unique_lock<std::mutex>&) ynothrow{
134 return true;
135 }, yforward(f), yforward(args)...);
136 }
137
138 size_t
139 size() const;
140
145 size_t
146 size_unlocked() const ynothrow
147 {
148 return tasks.size();
149 }
150
162 template<typename _fWaiter, typename _fCallable, typename... _tParams>
163 future_result_t<_fCallable, _tParams...>
164 wait_to_enqueue(_fWaiter waiter, _fCallable&& f, _tParams&&... args)
165 {
166 auto bound(ystdex::pack_task(yforward(f), yforward(args)...));
167 auto res(bound.get_future());
168
169 {
170 std::unique_lock<std::mutex> lck(queue_mutex);
171
172 if(waiter(lck))
173 {
174 yassume(lck.owns_lock());
175 // TODO: Blocked. Use ISO C++14 lambda initializers to reduce
176 // initialization cost by directly moving from %bound.
177 tasks.push(std::packaged_task<void()>(
178 std::bind([](decltype(bound)& tsk){
179 tsk();
180 }, std::move(bound))));
181 }
182 else
183 return {};
184 }
185 condition.notify_one();
186 return res;
187 }
188 };
189
190
197 class YB_API task_pool : private thread_pool
198 {
199 private:
200 size_t max_tasks;
201 std::condition_variable enqueue_condition{};
202
203 public:
209 task_pool(size_t n, std::function<void()> on_enter = {},
210 std::function<void()> on_exit = {})
211 : thread_pool(std::max<size_t>(n, 1), on_enter, on_exit),
212 max_tasks(std::max<size_t>(n, 1))
213 {}
214
216 bool
217 can_enqueue_unlocked() const ynothrow
218 {
219 return size_unlocked() < max_tasks;
220 }
221
222 size_t
223 get_max_task_num() const ynothrow
224 {
225 return max_tasks;
226 }
227
232 void
233 reset();
234
235 using thread_pool::size;
236
238
239 template<typename _func, typename _fCallable, typename... _tParams>
240 future_result_t<_fCallable, _tParams...>
241 poll(_func poller, _fCallable&& f, _tParams&&... args)
242 {
243 return wait_to_enqueue([=](std::unique_lock<std::mutex>& lck){
244 enqueue_condition.wait(lck, [=]{
245 return poller() && can_enqueue_unlocked();
246 });
247 return true;
248 }, yforward(f), yforward(args)...);
249 }
250
251 template<typename _func, typename _tDuration, typename _fCallable,
252 typename... _tParams>
253 future_result_t<_fCallable, _tParams...>
254 poll_for(_func poller, const _tDuration& duration, _fCallable&& f,
255 _tParams&&... args)
256 {
257 return wait_to_enqueue([=](std::unique_lock<std::mutex>& lck){
258 return enqueue_condition.wait_for(lck, duration, [=]{
259 return poller() && can_enqueue_unlocked();
260 });
261 }, yforward(f), yforward(args)...);
262 }
263
264 template<typename _func, typename _tTimePoint, typename _fCallable,
265 typename... _tParams>
266 future_result_t<_fCallable, _tParams...>
267 poll_until(_func poller, const _tTimePoint& abs_time,
268 _fCallable&& f, _tParams&&... args)
269 {
270 return wait_to_enqueue([=](std::unique_lock<std::mutex>& lck){
271 return enqueue_condition.wait_until(lck, abs_time, [=]{
272 return poller() && can_enqueue_unlocked();
273 });
274 }, yforward(f), yforward(args)...);
275 }
276
277 template<typename _fCallable, typename... _tParams>
278 inline future_result_t<_fCallable, _tParams...>
279 wait(_fCallable&& f, _tParams&&... args)
280 {
281 return poll([]{
282 return true;
283 }, yforward(f), yforward(args)...);
284 }
285
286 template<typename _tDuration, typename _fCallable, typename... _tParams>
287 inline future_result_t<_fCallable, _tParams...>
288 wait_for(const _tDuration& duration, _fCallable&& f, _tParams&&... args)
289 {
290 return poll_for([]{
291 return true;
292 }, duration, yforward(f), yforward(args)...);
293 }
294
295 template<typename _fWaiter, typename _fCallable, typename... _tParams>
296 future_result_t<_fCallable, _tParams...>
297 wait_to_enqueue(_fWaiter waiter, _fCallable&& f, _tParams&&... args)
298 {
299 return thread_pool::wait_to_enqueue(
300 [=](std::unique_lock<std::mutex>& lck) -> bool{
301 while(!can_enqueue_unlocked())
302 if(!waiter(lck))
303 return {};
304 return true;
305 }, [=](_tParams&&... f_args){
306 // TODO: Blocked. Use ISO C++14 lambda initializers to implement
307 // passing %f with %ystdex::decay_copy.
308 enqueue_condition.notify_one();
309 // XXX: Blocked. 'yforward' cause G++ 5.2 failed (perhaps silently)
310 // with exit code 1.
311 return ystdex::invoke(f, std::forward<_tParams&&>(f_args)...);
312 }, yforward(args)...);
313 }
314
315 template<typename _tTimePoint, typename _fCallable, typename... _tParams>
316 inline future_result_t<_fCallable, _tParams...>
317 wait_until(const _tTimePoint& abs_time, _fCallable&& f, _tParams&&... args)
318 {
319 return poll_until([]{
320 return true;
321 }, abs_time, yforward(f), yforward(args)...);
322 }
324 };
325 # endif
326
327 } // namespace ystdex;
328
329 #endif
330
yassume
#define yassume(_expr)
假定：环境语义。
Definition: cassert.h:99
ystdex::task_pool::max_tasks
size_t max_tasks
Definition: concurrency.h:200
ystdex::to_string
std::string to_string(const std::thread::id &)
转换 thread::id 为字符串表示。
ynothrow
#define ynothrow
YSLib 无异常抛出保证：若支持 noexcept 关键字， 指定特定的 noexcept 异常规范。
Definition: ydef.h:740
ystdex::thread_pool::size_unlocked
size_t size_unlocked() const
Definition: concurrency.h:146
platform::containers::string
basic_string< char > string
Definition: Container.h:101
cassert.h
ISO C 断言/调试跟踪扩展。
future.hpp
标准库 扩展。
ystdex::thread_pool::workers
std::vector< std::thread > workers
Definition: concurrency.h:96
ystdex::thread_pool
线程池。
Definition: concurrency.h:93
pseudo_mutex.h
伪互斥量。
ystdex::reset
bool reset(std::unique_ptr< _type > &p)
Definition: memory.hpp:584
ystdex::invoke
result_of_t< _fCallable &&(_tParams &&...)> invoke(_fCallable &&f, _tParams &&...args)
调用可调用对象。
Definition: functional.hpp:337
ystdex::task_pool::task_pool
task_pool(size_t n, std::function< void()> on_enter={}, std::function< void()> on_exit={})
初始化：使用指定的初始化和退出回调指定数量的工作线程和最大任务数。
Definition: concurrency.h:209
YB_API
#define YB_API
YBase 应用程序编程接口：用于向库文件约定链接。
Definition: ydef.h:566
ystdex::task_pool::wait_until
future_result_t< _fCallable, _tParams... > wait_until(const _tTimePoint &abs_time, _fCallable &&f, _tParams &&...args)
Definition: concurrency.h:317
ystdex::task_pool::wait_to_enqueue
future_result_t< _fCallable, _tParams... > wait_to_enqueue(_fWaiter waiter, _fCallable &&f, _tParams &&...args)
Definition: concurrency.h:297
ystdex::thread_pool::size
size_t size() const
functional.hpp
函数和可调用对象。
ystdex::task_pool::poll_until
future_result_t< _fCallable, _tParams... > poll_until(_func poller, const _tTimePoint &abs_time, _fCallable &&f, _tParams &&...args)
Definition: concurrency.h:267
ystdex::size
auto size(const _tRange &c) -> decltype(c.size())
类容器访问。
Definition: range.hpp:233
ystdex
特性测试宏。
Definition: ydef.h:796
ystdex::task_pool::wait
future_result_t< _fCallable, _tParams... > wait(_fCallable &&f, _tParams &&...args)
Definition: concurrency.h:279
ystdex::task_pool::poll_for
future_result_t< _fCallable, _tParams... > poll_for(_func poller, const _tDuration &duration, _fCallable &&f, _tParams &&...args)
Definition: concurrency.h:254
ystdex::task_pool
任务池：带有队列大小限制的线程池。
Definition: concurrency.h:197
ystdex::threading::locked_ptr
std::unique_ptr< _type, unlock_delete< _tMutex, _tLock >> locked_ptr
独占所有权的锁定指针。
Definition: pseudo_mutex.h:625
ystdex::pack_task
packed_task_t< _fCallable &&, _tParams &&... > pack_task(_fCallable &&f, _tParams &&...args)
Definition: future.hpp:98
ystdex::task_pool::get_max_task_num
size_t get_max_task_num() const
Definition: concurrency.h:223
ystdex::get_this_thread_id
std::string get_this_thread_id()
取当前线程标识的字符串表示。
Definition: concurrency.h:79
ystdex::thread_pool::wait_to_enqueue
future_result_t< _fCallable, _tParams... > wait_to_enqueue(_fWaiter waiter, _fCallable &&f, _tParams &&...args)
等待操作进入队列。
Definition: concurrency.h:164
yforward
#define yforward(_expr)
根据参数类型使用 std::forward 传递对应参数。
Definition: ydef.h:969
ystdex::task_pool::poll
future_result_t< _fCallable, _tParams... > poll(_func poller, _fCallable &&f, _tParams &&...args)
Definition: concurrency.h:241
ystdex::locked_ptr
threading::locked_ptr< _type, _tMutex, _tLock > locked_ptr
独占所有权的锁定指针：默认使用标准库互斥量和锁。
Definition: concurrency.h:62
ystdex::task_pool::wait_for
future_result_t< _fCallable, _tParams... > wait_for(const _tDuration &duration, _fCallable &&f, _tParams &&...args)
Definition: concurrency.h:288
YSLib::IO::f
const string &path const String &path _func f
Definition: YSLib/Service/FileSystem.h:430
ystdex::task_pool::can_enqueue_unlocked
bool can_enqueue_unlocked() const
Definition: concurrency.h:217
CHRLib::args
_tParams && args
Definition: CharacterProcessing.h:538
ystdex::future_result_t
std::future< result_of_t< _fCallable &&(_tParams &&...)>> future_result_t
Definition: future.hpp:86
platform::n
size_t n
Definition: YCLib/FileSystem.cpp:321
ystdex::threading::unlock_delete
解锁删除器。
Definition: pseudo_mutex.h:587