queue.hpp 19 KB

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