YSTest: concurrency.h 源文件

YSTest PreAlpha_b800_20170810 The YSLib Test Project

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

- 文件列表
- 文件成员

concurrency.h
浏览该文件的文档.
1 /*
2 © 2014-2016 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 threading::unlock_delete;
32 #include <mutex> // for std::mutex, std::unique_lock;
33 #include <thread> // for std::thread;
34 #include <vector> // for std::vector;
35 #include <queue> // for std::queue;
36 #include "future.hpp" // for std::packaged_task, future_result_t, pack_task;
37 #include <condition_variable> // for std::condition_variable;
38 #include "functional.hpp" // for std::bind, std::function, ystdex::invoke;
39 #include "cassert.h" // for yassume;
40
41 namespace ystdex
42 {
43
49 template<class _tMutex = std::mutex,
50 class _tLock = std::unique_lock<_tMutex>>
51 using unlock_delete = threading::unlock_delete<_tMutex, _tLock>;
52
53
59 template<typename _type, class _tMutex = std::mutex,
60 class _tLock = std::unique_lock<_tMutex>>
61 using locked_ptr = threading::locked_ptr<_type, _tMutex, _tLock>;
62
63
69 YB_API std::string
70 to_string(const std::thread::id&);
71
77 inline std::string
78 get_this_thread_id()
79 {
80 return ystdex::to_string(std::this_thread::get_id());
81 }
82
83
84 # if !__GLIBCXX__ || (defined(_GLIBCXX_HAS_GTHREADS) \
85 && defined(_GLIBCXX_USE_C99_STDINT_TR1) && (ATOMIC_INT_LOCK_FREE > 1))
86
92 class YB_API thread_pool
93 {
94 private:
95 std::vector<std::thread> workers{};
97 std::queue<std::packaged_task<void()>> tasks{};
98 mutable std::mutex queue_mutex{};
99 std::condition_variable condition{};
105 bool stopped = {};
106
107 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 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
229
230 template<typename _func, typename _fCallable, typename... _tParams>
231 future_result_t<_fCallable, _tParams...>
232 poll(_func poller, _fCallable&& f, _tParams&&... args)
233 {
234 return wait_to_enqueue([=](std::unique_lock<std::mutex>& lck){
235 enqueue_condition.wait(lck, [=]{
236 return poller() && can_enqueue_unlocked();
237 });
238 return true;
239 }, yforward(f), yforward(args)...);
240 }
241
242 template<typename _func, typename _tDuration, typename _fCallable,
243 typename... _tParams>
244 future_result_t<_fCallable, _tParams...>
245 poll_for(_func poller, const _tDuration& duration, _fCallable&& f,
246 _tParams&&... args)
247 {
248 return wait_to_enqueue([=](std::unique_lock<std::mutex>& lck){
249 return enqueue_condition.wait_for(lck, duration, [=]{
250 return poller() && can_enqueue_unlocked();
251 });
252 }, yforward(f), yforward(args)...);
253 }
254
255 template<typename _func, typename _tTimePoint, typename _fCallable,
256 typename... _tParams>
257 future_result_t<_fCallable, _tParams...>
258 poll_until(_func poller, const _tTimePoint& abs_time,
259 _fCallable&& f, _tParams&&... args)
260 {
261 return wait_to_enqueue([=](std::unique_lock<std::mutex>& lck){
262 return enqueue_condition.wait_until(lck, abs_time, [=]{
263 return poller() && can_enqueue_unlocked();
264 });
265 }, yforward(f), yforward(args)...);
266 }
268
270
271 void
273 reset()
274 {
275 reset(max_tasks);
276 }
277
282 void
283 reset(size_t);
285
286 using thread_pool::size;
287
289
290 template<typename _fCallable, typename... _tParams>
291 inline future_result_t<_fCallable, _tParams...>
292 wait(_fCallable&& f, _tParams&&... args)
293 {
294 return poll([]{
295 return true;
296 }, yforward(f), yforward(args)...);
297 }
298
299 template<typename _tDuration, typename _fCallable, typename... _tParams>
300 inline future_result_t<_fCallable, _tParams...>
301 wait_for(const _tDuration& duration, _fCallable&& f, _tParams&&... args)
302 {
303 return poll_for([]{
304 return true;
305 }, duration, yforward(f), yforward(args)...);
306 }
307
308 template<typename _fWaiter, typename _fCallable, typename... _tParams>
309 future_result_t<_fCallable, _tParams...>
310 wait_to_enqueue(_fWaiter waiter, _fCallable&& f, _tParams&&... args)
311 {
312 return thread_pool::wait_to_enqueue(
313 [=](std::unique_lock<std::mutex>& lck) -> bool{
314 while(!can_enqueue_unlocked())
315 if(!waiter(lck))
316 return {};
317 return true;
318 }, [=](_tParams&&... f_args){
319 // TODO: Blocked. Use C++14 lambda initializers to implement
320 // passing %f with %ystdex::decay_copy.
321 enqueue_condition.notify_one();
322 // XXX: Blocked. 'yforward' cause G++ 5.2 to fail (perhaps silently)
323 // with exit code 1.
324 return ystdex::invoke(f, std::forward<_tParams&&>(f_args)...);
325 }, yforward(args)...);
326 }
327
328 template<typename _tTimePoint, typename _fCallable, typename... _tParams>
329 inline future_result_t<_fCallable, _tParams...>
330 wait_until(const _tTimePoint& abs_time, _fCallable&& f, _tParams&&... args)
331 {
332 return poll_until([]{
333 return true;
334 }, abs_time, yforward(f), yforward(args)...);
335 }
337 };
338 # endif
339
340 } // namespace ystdex;
341
342 #endif
343
ystdex::reset
bool reset(std::unique_ptr< _type, _tDeleter > &p)
Definition: memory.hpp:708
ystdex::thread_pool::size
size_t size() const
ystdex::to_string
std::string to_string(const std::thread::id &)
转换 thread::id 为字符串表示。
ystdex::task_pool::poll_for
future_result_t< _fCallable, _tParams... > poll_for(_func poller, const _tDuration &duration, _fCallable &&f, _tParams &&...args)
Definition: concurrency.h:245
ystdex::task_pool::reset
void reset()
重置线程池。
Definition: concurrency.h:273
ystdex::task_pool::wait_until
future_result_t< _fCallable, _tParams... > wait_until(const _tTimePoint &abs_time, _fCallable &&f, _tParams &&...args)
Definition: concurrency.h:330
CHRLib::f
f(d, s)
ystdex::pack_task
packed_task_t< _fCallable &&, _tParams &&... > pack_task(_fCallable &&f, _tParams &&...args)
Definition: future.hpp:98
ystdex::task_pool::poll
future_result_t< _fCallable, _tParams... > poll(_func poller, _fCallable &&f, _tParams &&...args)
Definition: concurrency.h:232
future.hpp
标准库 扩展。
CHRLib::args
_tParams && args
Definition: CharacterProcessing.h:538
platform::containers::string
basic_string< char > string
Definition: Container.h:101
ystdex::thread_pool::size_unlocked
size_t size_unlocked() const
Definition: concurrency.h:146
ystdex::task_pool::wait_for
future_result_t< _fCallable, _tParams... > wait_for(const _tDuration &duration, _fCallable &&f, _tParams &&...args)
Definition: concurrency.h:301
ystdex::task_pool::wait
future_result_t< _fCallable, _tParams... > wait(_fCallable &&f, _tParams &&...args)
Definition: concurrency.h:292
ystdex::threading::unlock_delete
解锁删除器。
Definition: pseudo_mutex.h:1045
ystdex::get_this_thread_id
std::string get_this_thread_id()
取当前线程标识的字符串表示。
Definition: concurrency.h:78
ynothrow
#define ynothrow
YSLib 无异常抛出保证：若支持 noexcept 关键字， 指定特定的 noexcept 异常规范。
Definition: ydef.h:766
ystdex::task_pool::task_pool
task_pool(size_t n, std::function< void()> on_enter={}, std::function< void()> on_exit={})
初始化：使用指定的初始化和退出回调指定数量的工作线程和最大任务数。
Definition: concurrency.h:209
ystdex
特性测试宏。
Definition: ydef.h:822
ystdex::task_pool::wait_to_enqueue
future_result_t< _fCallable, _tParams... > wait_to_enqueue(_fWaiter waiter, _fCallable &&f, _tParams &&...args)
Definition: concurrency.h:310
pseudo_mutex.h
伪互斥量。
cassert.h
ISO C 断言/调试跟踪扩展。
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:258
ystdex::future_result_t
std::future< result_of_t< _fCallable &&(_tParams &&...)>> future_result_t
Definition: future.hpp:86
ystdex::task_pool
任务池：带有队列大小限制的线程池。
Definition: concurrency.h:197
ystdex::task_pool::can_enqueue_unlocked
bool can_enqueue_unlocked() const
Definition: concurrency.h:217
ystdex::locked_ptr
threading::locked_ptr< _type, _tMutex, _tLock > locked_ptr
独占所有权的锁定指针：默认使用标准库互斥量和锁。
Definition: concurrency.h:61
yforward
#define yforward(_expr)
根据参数类型使用 std::forward 传递对应参数。
Definition: ydef.h:995
yassume
#define yassume(_expr)
假定：环境语义。
Definition: cassert.h:105
ystdex::invoke
result_of_t< _fCallable &&(_tParams &&...)> invoke(_fCallable &&f, _tParams &&...args)
调用可调用对象。
Definition: functional.hpp:339
ystdex::thread_pool
线程池。
Definition: concurrency.h:92
ystdex::threading::locked_ptr
std::unique_ptr< _type, unlock_delete< _tMutex, _tLock >> locked_ptr
独占所有权的锁定指针。
Definition: pseudo_mutex.h:1083
YB_API
#define YB_API
YBase 应用程序编程接口：用于向库文件约定链接。
Definition: ydef.h:592
ystdex::size
auto size(const _tRange &c) -> decltype(c.size())
类容器访问。
Definition: range.hpp:233
ystdex::task_pool::max_tasks
size_t max_tasks
Definition: concurrency.h:200
ystdex::thread_pool::wait_to_enqueue
future_result_t< _fCallable, _tParams... > wait_to_enqueue(_fWaiter waiter, _fCallable &&f, _tParams &&...args)
等待操作进入队列。
Definition: concurrency.h:164
ystdex::task_pool::get_max_task_num
size_t get_max_task_num() const
Definition: concurrency.h:223