queue.hpp 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564
  1. // lock-free queue from
  2. // Michael, M. M. and Scott, M. L.,
  3. // "simple, fast and practical non-blocking and blocking concurrent queue algorithms"
  4. //
  5. // Copyright (C) 2008-2013 Tim Blechmann
  6. //
  7. // Distributed under the Boost Software License, Version 1.0. (See
  8. // accompanying file LICENSE_1_0.txt or copy at
  9. // http://www.boost.org/LICENSE_1_0.txt)
  10. #ifndef BOOST_LOCKFREE_FIFO_HPP_INCLUDED
  11. #define BOOST_LOCKFREE_FIFO_HPP_INCLUDED
  12. #include <boost/assert.hpp>
  13. #include <boost/static_assert.hpp>
  14. #include <boost/type_traits/has_trivial_assign.hpp>
  15. #include <boost/type_traits/has_trivial_destructor.hpp>
  16. #include <boost/config.hpp> // for BOOST_LIKELY & BOOST_ALIGNMENT
  17. #include <boost/lockfree/detail/allocator_rebind_helper.hpp>
  18. #include <boost/lockfree/detail/atomic.hpp>
  19. #include <boost/lockfree/detail/copy_payload.hpp>
  20. #include <boost/lockfree/detail/freelist.hpp>
  21. #include <boost/lockfree/detail/parameter.hpp>
  22. #include <boost/lockfree/detail/tagged_ptr.hpp>
  23. #include <boost/lockfree/lockfree_forward.hpp>
  24. #ifdef BOOST_HAS_PRAGMA_ONCE
  25. #pragma once
  26. #endif
  27. #if defined(_MSC_VER)
  28. #pragma warning(push)
  29. #pragma warning(disable: 4324) // structure was padded due to __declspec(align())
  30. #endif
  31. #if defined(BOOST_INTEL) && (BOOST_INTEL_CXX_VERSION > 1000)
  32. #pragma warning(push)
  33. #pragma warning(disable:488) // template parameter unused in declaring parameter types,
  34. // gets erronously triggered the queue constructor which
  35. // takes an allocator of another type and rebinds it
  36. #endif
  37. namespace boost {
  38. namespace lockfree {
  39. namespace detail {
  40. typedef parameter::parameters<boost::parameter::optional<tag::allocator>,
  41. boost::parameter::optional<tag::capacity>
  42. > queue_signature;
  43. } /* namespace detail */
  44. /** The queue class provides a multi-writer/multi-reader queue, pushing and popping is lock-free,
  45. * construction/destruction has to be synchronized. It uses a freelist for memory management,
  46. * freed nodes are pushed to the freelist and not returned to the OS before the queue is destroyed.
  47. *
  48. * \b Policies:
  49. * - \ref boost::lockfree::fixed_sized, defaults to \c boost::lockfree::fixed_sized<false> \n
  50. * Can be used to completely disable dynamic memory allocations during push in order to ensure lockfree behavior. \n
  51. * If the data structure is configured as fixed-sized, the internal nodes are stored inside an array and they are addressed
  52. * by array indexing. This limits the possible size of the queue to the number of elements that can be addressed by the index
  53. * type (usually 2**16-2), but on platforms that lack double-width compare-and-exchange instructions, this is the best way
  54. * to achieve lock-freedom.
  55. *
  56. * - \ref boost::lockfree::capacity, optional \n
  57. * If this template argument is passed to the options, the size of the queue is set at compile-time.\n
  58. * This option implies \c fixed_sized<true>
  59. *
  60. * - \ref boost::lockfree::allocator, defaults to \c boost::lockfree::allocator<std::allocator<void>> \n
  61. * Specifies the allocator that is used for the internal freelist
  62. *
  63. * \b Requirements:
  64. * - T must have a copy constructor
  65. * - T must have a trivial assignment operator
  66. * - T must have a trivial destructor
  67. *
  68. * */
  69. #ifdef BOOST_NO_CXX11_VARIADIC_TEMPLATES
  70. template <typename T, class A0, class A1, class A2>
  71. #else
  72. template <typename T, typename ...Options>
  73. #endif
  74. class queue
  75. {
  76. private:
  77. #ifndef BOOST_DOXYGEN_INVOKED
  78. #ifdef BOOST_HAS_TRIVIAL_DESTRUCTOR
  79. BOOST_STATIC_ASSERT((boost::has_trivial_destructor<T>::value));
  80. #endif
  81. #ifdef BOOST_HAS_TRIVIAL_ASSIGN
  82. BOOST_STATIC_ASSERT((boost::has_trivial_assign<T>::value));
  83. #endif
  84. #ifdef BOOST_NO_CXX11_VARIADIC_TEMPLATES
  85. typedef typename detail::queue_signature::bind<A0, A1, A2>::type bound_args;
  86. #else
  87. typedef typename detail::queue_signature::bind<Options...>::type bound_args;
  88. #endif
  89. static const bool has_capacity = detail::extract_capacity<bound_args>::has_capacity;
  90. static const size_t capacity = detail::extract_capacity<bound_args>::capacity + 1; // the queue uses one dummy node
  91. static const bool fixed_sized = detail::extract_fixed_sized<bound_args>::value;
  92. static const bool node_based = !(has_capacity || fixed_sized);
  93. static const bool compile_time_sized = has_capacity;
  94. struct BOOST_ALIGNMENT(BOOST_LOCKFREE_CACHELINE_BYTES) node
  95. {
  96. typedef typename detail::select_tagged_handle<node, node_based>::tagged_handle_type tagged_node_handle;
  97. typedef typename detail::select_tagged_handle<node, node_based>::handle_type handle_type;
  98. node(T const & v, handle_type null_handle):
  99. next(tagged_node_handle(null_handle, 0)), data(v)
  100. {
  101. /* increment tag to avoid ABA problem */
  102. tagged_node_handle old_next = next.load(memory_order_relaxed);
  103. tagged_node_handle new_next (null_handle, old_next.get_next_tag());
  104. next.store(new_next, memory_order_release);
  105. }
  106. node (handle_type null_handle):
  107. next(tagged_node_handle(null_handle, 0))
  108. {}
  109. node(void)
  110. {}
  111. atomic<tagged_node_handle> next;
  112. T data;
  113. };
  114. typedef typename detail::extract_allocator<bound_args, node>::type node_allocator;
  115. typedef typename detail::select_freelist<node, node_allocator, compile_time_sized, fixed_sized, capacity>::type pool_t;
  116. typedef typename pool_t::tagged_node_handle tagged_node_handle;
  117. typedef typename detail::select_tagged_handle<node, node_based>::handle_type handle_type;
  118. void initialize(void)
  119. {
  120. node * n = pool.template construct<true, false>(pool.null_handle());
  121. tagged_node_handle dummy_node(pool.get_handle(n), 0);
  122. head_.store(dummy_node, memory_order_relaxed);
  123. tail_.store(dummy_node, memory_order_release);
  124. }
  125. struct implementation_defined
  126. {
  127. typedef node_allocator allocator;
  128. typedef std::size_t size_type;
  129. };
  130. #endif
  131. BOOST_DELETED_FUNCTION(queue(queue const&))
  132. BOOST_DELETED_FUNCTION(queue& operator= (queue const&))
  133. public:
  134. typedef T value_type;
  135. typedef typename implementation_defined::allocator allocator;
  136. typedef typename implementation_defined::size_type size_type;
  137. /**
  138. * \return true, if implementation is lock-free.
  139. *
  140. * \warning It only checks, if the queue head and tail nodes and the freelist can be modified in a lock-free manner.
  141. * On most platforms, the whole implementation is lock-free, if this is true. Using c++0x-style atomics, there is
  142. * no possibility to provide a completely accurate implementation, because one would need to test every internal
  143. * node, which is impossible if further nodes will be allocated from the operating system.
  144. * */
  145. bool is_lock_free (void) const
  146. {
  147. return head_.is_lock_free() && tail_.is_lock_free() && pool.is_lock_free();
  148. }
  149. //! Construct queue
  150. // @{
  151. queue(void):
  152. head_(tagged_node_handle(0, 0)),
  153. tail_(tagged_node_handle(0, 0)),
  154. pool(node_allocator(), capacity)
  155. {
  156. BOOST_ASSERT(has_capacity);
  157. initialize();
  158. }
  159. template <typename U>
  160. explicit queue(typename detail::allocator_rebind_helper<node_allocator, U>::type const & alloc):
  161. head_(tagged_node_handle(0, 0)),
  162. tail_(tagged_node_handle(0, 0)),
  163. pool(alloc, capacity)
  164. {
  165. BOOST_STATIC_ASSERT(has_capacity);
  166. initialize();
  167. }
  168. explicit queue(allocator const & alloc):
  169. head_(tagged_node_handle(0, 0)),
  170. tail_(tagged_node_handle(0, 0)),
  171. pool(alloc, capacity)
  172. {
  173. BOOST_ASSERT(has_capacity);
  174. initialize();
  175. }
  176. // @}
  177. //! Construct queue, allocate n nodes for the freelist.
  178. // @{
  179. explicit queue(size_type n):
  180. head_(tagged_node_handle(0, 0)),
  181. tail_(tagged_node_handle(0, 0)),
  182. pool(node_allocator(), n + 1)
  183. {
  184. BOOST_ASSERT(!has_capacity);
  185. initialize();
  186. }
  187. template <typename U>
  188. queue(size_type n, typename detail::allocator_rebind_helper<node_allocator, U>::type const & alloc):
  189. head_(tagged_node_handle(0, 0)),
  190. tail_(tagged_node_handle(0, 0)),
  191. pool(alloc, n + 1)
  192. {
  193. BOOST_STATIC_ASSERT(!has_capacity);
  194. initialize();
  195. }
  196. // @}
  197. /** \copydoc boost::lockfree::stack::reserve
  198. * */
  199. void reserve(size_type n)
  200. {
  201. pool.template reserve<true>(n);
  202. }
  203. /** \copydoc boost::lockfree::stack::reserve_unsafe
  204. * */
  205. void reserve_unsafe(size_type n)
  206. {
  207. pool.template reserve<false>(n);
  208. }
  209. /** Destroys queue, free all nodes from freelist.
  210. * */
  211. ~queue(void)
  212. {
  213. T dummy;
  214. while(unsynchronized_pop(dummy))
  215. {}
  216. pool.template destruct<false>(head_.load(memory_order_relaxed));
  217. }
  218. /** Check if the queue is empty
  219. *
  220. * \return true, if the queue is empty, false otherwise
  221. * \note The result is only accurate, if no other thread modifies the queue. Therefore it is rarely practical to use this
  222. * value in program logic.
  223. * */
  224. bool empty(void) const
  225. {
  226. return pool.get_handle(head_.load()) == pool.get_handle(tail_.load());
  227. }
  228. /** Pushes object t to the queue.
  229. *
  230. * \post object will be pushed to the queue, if internal node can be allocated
  231. * \returns true, if the push operation is successful.
  232. *
  233. * \note Thread-safe. If internal memory pool is exhausted and the memory pool is not fixed-sized, a new node will be allocated
  234. * from the OS. This may not be lock-free.
  235. * */
  236. bool push(T const & t)
  237. {
  238. return do_push<false>(t);
  239. }
  240. /** Pushes object t to the queue.
  241. *
  242. * \post object will be pushed to the queue, if internal node can be allocated
  243. * \returns true, if the push operation is successful.
  244. *
  245. * \note Thread-safe and non-blocking. If internal memory pool is exhausted, operation will fail
  246. * \throws if memory allocator throws
  247. * */
  248. bool bounded_push(T const & t)
  249. {
  250. return do_push<true>(t);
  251. }
  252. private:
  253. #ifndef BOOST_DOXYGEN_INVOKED
  254. template <bool Bounded>
  255. bool do_push(T const & t)
  256. {
  257. node * n = pool.template construct<true, Bounded>(t, pool.null_handle());
  258. handle_type node_handle = pool.get_handle(n);
  259. if (n == NULL)
  260. return false;
  261. for (;;) {
  262. tagged_node_handle tail = tail_.load(memory_order_acquire);
  263. node * tail_node = pool.get_pointer(tail);
  264. tagged_node_handle next = tail_node->next.load(memory_order_acquire);
  265. node * next_ptr = pool.get_pointer(next);
  266. tagged_node_handle tail2 = tail_.load(memory_order_acquire);
  267. if (BOOST_LIKELY(tail == tail2)) {
  268. if (next_ptr == 0) {
  269. tagged_node_handle new_tail_next(node_handle, next.get_next_tag());
  270. if ( tail_node->next.compare_exchange_weak(next, new_tail_next) ) {
  271. tagged_node_handle new_tail(node_handle, tail.get_next_tag());
  272. tail_.compare_exchange_strong(tail, new_tail);
  273. return true;
  274. }
  275. }
  276. else {
  277. tagged_node_handle new_tail(pool.get_handle(next_ptr), tail.get_next_tag());
  278. tail_.compare_exchange_strong(tail, new_tail);
  279. }
  280. }
  281. }
  282. }
  283. #endif
  284. public:
  285. /** Pushes object t to the queue.
  286. *
  287. * \post object will be pushed to the queue, if internal node can be allocated
  288. * \returns true, if the push operation is successful.
  289. *
  290. * \note Not Thread-safe. If internal memory pool is exhausted and the memory pool is not fixed-sized, a new node will be allocated
  291. * from the OS. This may not be lock-free.
  292. * \throws if memory allocator throws
  293. * */
  294. bool unsynchronized_push(T const & t)
  295. {
  296. node * n = pool.template construct<false, false>(t, pool.null_handle());
  297. if (n == NULL)
  298. return false;
  299. for (;;) {
  300. tagged_node_handle tail = tail_.load(memory_order_relaxed);
  301. tagged_node_handle next = tail->next.load(memory_order_relaxed);
  302. node * next_ptr = next.get_ptr();
  303. if (next_ptr == 0) {
  304. tail->next.store(tagged_node_handle(n, next.get_next_tag()), memory_order_relaxed);
  305. tail_.store(tagged_node_handle(n, tail.get_next_tag()), memory_order_relaxed);
  306. return true;
  307. }
  308. else
  309. tail_.store(tagged_node_handle(next_ptr, tail.get_next_tag()), memory_order_relaxed);
  310. }
  311. }
  312. /** Pops object from queue.
  313. *
  314. * \post if pop operation is successful, object will be copied to ret.
  315. * \returns true, if the pop operation is successful, false if queue was empty.
  316. *
  317. * \note Thread-safe and non-blocking
  318. * */
  319. bool pop (T & ret)
  320. {
  321. return pop<T>(ret);
  322. }
  323. /** Pops object from queue.
  324. *
  325. * \pre type U must be constructible by T and copyable, or T must be convertible to U
  326. * \post if pop operation is successful, object will be copied to ret.
  327. * \returns true, if the pop operation is successful, false if queue was empty.
  328. *
  329. * \note Thread-safe and non-blocking
  330. * */
  331. template <typename U>
  332. bool pop (U & ret)
  333. {
  334. for (;;) {
  335. tagged_node_handle head = head_.load(memory_order_acquire);
  336. node * head_ptr = pool.get_pointer(head);
  337. tagged_node_handle tail = tail_.load(memory_order_acquire);
  338. tagged_node_handle next = head_ptr->next.load(memory_order_acquire);
  339. node * next_ptr = pool.get_pointer(next);
  340. tagged_node_handle head2 = head_.load(memory_order_acquire);
  341. if (BOOST_LIKELY(head == head2)) {
  342. if (pool.get_handle(head) == pool.get_handle(tail)) {
  343. if (next_ptr == 0)
  344. return false;
  345. tagged_node_handle new_tail(pool.get_handle(next), tail.get_next_tag());
  346. tail_.compare_exchange_strong(tail, new_tail);
  347. } else {
  348. if (next_ptr == 0)
  349. /* this check is not part of the original algorithm as published by michael and scott
  350. *
  351. * however we reuse the tagged_ptr part for the freelist and clear the next part during node
  352. * allocation. we can observe a null-pointer here.
  353. * */
  354. continue;
  355. detail::copy_payload(next_ptr->data, ret);
  356. tagged_node_handle new_head(pool.get_handle(next), head.get_next_tag());
  357. if (head_.compare_exchange_weak(head, new_head)) {
  358. pool.template destruct<true>(head);
  359. return true;
  360. }
  361. }
  362. }
  363. }
  364. }
  365. /** Pops object from queue.
  366. *
  367. * \post if pop operation is successful, object will be copied to ret.
  368. * \returns true, if the pop operation is successful, false if queue was empty.
  369. *
  370. * \note Not thread-safe, but non-blocking
  371. *
  372. * */
  373. bool unsynchronized_pop (T & ret)
  374. {
  375. return unsynchronized_pop<T>(ret);
  376. }
  377. /** Pops object from queue.
  378. *
  379. * \pre type U must be constructible by T and copyable, or T must be convertible to U
  380. * \post if pop operation is successful, object will be copied to ret.
  381. * \returns true, if the pop operation is successful, false if queue was empty.
  382. *
  383. * \note Not thread-safe, but non-blocking
  384. *
  385. * */
  386. template <typename U>
  387. bool unsynchronized_pop (U & ret)
  388. {
  389. for (;;) {
  390. tagged_node_handle head = head_.load(memory_order_relaxed);
  391. node * head_ptr = pool.get_pointer(head);
  392. tagged_node_handle tail = tail_.load(memory_order_relaxed);
  393. tagged_node_handle next = head_ptr->next.load(memory_order_relaxed);
  394. node * next_ptr = pool.get_pointer(next);
  395. if (pool.get_handle(head) == pool.get_handle(tail)) {
  396. if (next_ptr == 0)
  397. return false;
  398. tagged_node_handle new_tail(pool.get_handle(next), tail.get_next_tag());
  399. tail_.store(new_tail);
  400. } else {
  401. if (next_ptr == 0)
  402. /* this check is not part of the original algorithm as published by michael and scott
  403. *
  404. * however we reuse the tagged_ptr part for the freelist and clear the next part during node
  405. * allocation. we can observe a null-pointer here.
  406. * */
  407. continue;
  408. detail::copy_payload(next_ptr->data, ret);
  409. tagged_node_handle new_head(pool.get_handle(next), head.get_next_tag());
  410. head_.store(new_head);
  411. pool.template destruct<false>(head);
  412. return true;
  413. }
  414. }
  415. }
  416. /** consumes one element via a functor
  417. *
  418. * pops one element from the queue and applies the functor on this object
  419. *
  420. * \returns true, if one element was consumed
  421. *
  422. * \note Thread-safe and non-blocking, if functor is thread-safe and non-blocking
  423. * */
  424. template <typename Functor>
  425. bool consume_one(Functor & f)
  426. {
  427. T element;
  428. bool success = pop(element);
  429. if (success)
  430. f(element);
  431. return success;
  432. }
  433. /// \copydoc boost::lockfree::queue::consume_one(Functor & rhs)
  434. template <typename Functor>
  435. bool consume_one(Functor const & f)
  436. {
  437. T element;
  438. bool success = pop(element);
  439. if (success)
  440. f(element);
  441. return success;
  442. }
  443. /** consumes all elements via a functor
  444. *
  445. * sequentially pops all elements from the queue and applies the functor on each object
  446. *
  447. * \returns number of elements that are consumed
  448. *
  449. * \note Thread-safe and non-blocking, if functor is thread-safe and non-blocking
  450. * */
  451. template <typename Functor>
  452. size_t consume_all(Functor & f)
  453. {
  454. size_t element_count = 0;
  455. while (consume_one(f))
  456. element_count += 1;
  457. return element_count;
  458. }
  459. /// \copydoc boost::lockfree::queue::consume_all(Functor & rhs)
  460. template <typename Functor>
  461. size_t consume_all(Functor const & f)
  462. {
  463. size_t element_count = 0;
  464. while (consume_one(f))
  465. element_count += 1;
  466. return element_count;
  467. }
  468. private:
  469. #ifndef BOOST_DOXYGEN_INVOKED
  470. atomic<tagged_node_handle> head_;
  471. static const int padding_size = BOOST_LOCKFREE_CACHELINE_BYTES - sizeof(tagged_node_handle);
  472. char padding1[padding_size];
  473. atomic<tagged_node_handle> tail_;
  474. char padding2[padding_size];
  475. pool_t pool;
  476. #endif
  477. };
  478. } /* namespace lockfree */
  479. } /* namespace boost */
  480. #if defined(BOOST_INTEL) && (BOOST_INTEL_CXX_VERSION > 1000)
  481. #pragma warning(pop)
  482. #endif
  483. #if defined(_MSC_VER)
  484. #pragma warning(pop)
  485. #endif
  486. #endif /* BOOST_LOCKFREE_FIFO_HPP_INCLUDED */