queue.hpp 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588
  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/core/allocator_access.hpp>
  15. #include <boost/type_traits/has_trivial_assign.hpp>
  16. #include <boost/type_traits/has_trivial_destructor.hpp>
  17. #include <boost/config.hpp> // for BOOST_LIKELY & BOOST_ALIGNMENT
  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. 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 a fixed-sized queue
  150. *
  151. * \pre Must specify a capacity<> argument
  152. * */
  153. queue(void):
  154. head_(tagged_node_handle(0, 0)),
  155. tail_(tagged_node_handle(0, 0)),
  156. pool(node_allocator(), capacity)
  157. {
  158. // Don't use BOOST_STATIC_ASSERT() here since it will be evaluated when compiling
  159. // this function and this function may be compiled even when it isn't being used.
  160. BOOST_ASSERT(has_capacity);
  161. initialize();
  162. }
  163. /** Construct a fixed-sized queue with a custom allocator
  164. *
  165. * \pre Must specify a capacity<> argument
  166. * */
  167. template <typename U>
  168. explicit queue(typename boost::allocator_rebind<node_allocator, U>::type const & alloc):
  169. head_(tagged_node_handle(0, 0)),
  170. tail_(tagged_node_handle(0, 0)),
  171. pool(alloc, capacity)
  172. {
  173. BOOST_STATIC_ASSERT(has_capacity);
  174. initialize();
  175. }
  176. /** Construct a fixed-sized queue with a custom allocator
  177. *
  178. * \pre Must specify a capacity<> argument
  179. * */
  180. explicit queue(allocator const & alloc):
  181. head_(tagged_node_handle(0, 0)),
  182. tail_(tagged_node_handle(0, 0)),
  183. pool(alloc, capacity)
  184. {
  185. // Don't use BOOST_STATIC_ASSERT() here since it will be evaluated when compiling
  186. // this function and this function may be compiled even when it isn't being used.
  187. BOOST_ASSERT(has_capacity);
  188. initialize();
  189. }
  190. /** Construct a variable-sized queue
  191. *
  192. * Allocate n nodes initially for the freelist
  193. *
  194. * \pre Must \b not specify a capacity<> argument
  195. * */
  196. explicit queue(size_type n):
  197. head_(tagged_node_handle(0, 0)),
  198. tail_(tagged_node_handle(0, 0)),
  199. pool(node_allocator(), n + 1)
  200. {
  201. // Don't use BOOST_STATIC_ASSERT() here since it will be evaluated when compiling
  202. // this function and this function may be compiled even when it isn't being used.
  203. BOOST_ASSERT(!has_capacity);
  204. initialize();
  205. }
  206. /** Construct a variable-sized queue with a custom allocator
  207. *
  208. * Allocate n nodes initially for the freelist
  209. *
  210. * \pre Must \b not specify a capacity<> argument
  211. * */
  212. template <typename U>
  213. queue(size_type n, typename boost::allocator_rebind<node_allocator, U>::type const & alloc):
  214. head_(tagged_node_handle(0, 0)),
  215. tail_(tagged_node_handle(0, 0)),
  216. pool(alloc, n + 1)
  217. {
  218. BOOST_STATIC_ASSERT(!has_capacity);
  219. initialize();
  220. }
  221. /** \copydoc boost::lockfree::stack::reserve
  222. * */
  223. void reserve(size_type n)
  224. {
  225. pool.template reserve<true>(n);
  226. }
  227. /** \copydoc boost::lockfree::stack::reserve_unsafe
  228. * */
  229. void reserve_unsafe(size_type n)
  230. {
  231. pool.template reserve<false>(n);
  232. }
  233. /** Destroys queue, free all nodes from freelist.
  234. * */
  235. ~queue(void)
  236. {
  237. T dummy;
  238. while(unsynchronized_pop(dummy))
  239. {}
  240. pool.template destruct<false>(head_.load(memory_order_relaxed));
  241. }
  242. /** Check if the queue is empty
  243. *
  244. * \return true, if the queue is empty, false otherwise
  245. * \note The result is only accurate, if no other thread modifies the queue. Therefore it is rarely practical to use this
  246. * value in program logic.
  247. * */
  248. bool empty(void) const
  249. {
  250. return pool.get_handle(head_.load()) == pool.get_handle(tail_.load());
  251. }
  252. /** Pushes object t to the queue.
  253. *
  254. * \post object will be pushed to the queue, if internal node can be allocated
  255. * \returns true, if the push operation is successful.
  256. *
  257. * \note Thread-safe. If internal memory pool is exhausted and the memory pool is not fixed-sized, a new node will be allocated
  258. * from the OS. This may not be lock-free.
  259. * */
  260. bool push(T const & t)
  261. {
  262. return do_push<false>(t);
  263. }
  264. /** Pushes object t to the queue.
  265. *
  266. * \post object will be pushed to the queue, if internal node can be allocated
  267. * \returns true, if the push operation is successful.
  268. *
  269. * \note Thread-safe and non-blocking. If internal memory pool is exhausted, operation will fail
  270. * \throws if memory allocator throws
  271. * */
  272. bool bounded_push(T const & t)
  273. {
  274. return do_push<true>(t);
  275. }
  276. private:
  277. #ifndef BOOST_DOXYGEN_INVOKED
  278. template <bool Bounded>
  279. bool do_push(T const & t)
  280. {
  281. node * n = pool.template construct<true, Bounded>(t, pool.null_handle());
  282. handle_type node_handle = pool.get_handle(n);
  283. if (n == NULL)
  284. return false;
  285. for (;;) {
  286. tagged_node_handle tail = tail_.load(memory_order_acquire);
  287. node * tail_node = pool.get_pointer(tail);
  288. tagged_node_handle next = tail_node->next.load(memory_order_acquire);
  289. node * next_ptr = pool.get_pointer(next);
  290. tagged_node_handle tail2 = tail_.load(memory_order_acquire);
  291. if (BOOST_LIKELY(tail == tail2)) {
  292. if (next_ptr == 0) {
  293. tagged_node_handle new_tail_next(node_handle, next.get_next_tag());
  294. if ( tail_node->next.compare_exchange_weak(next, new_tail_next) ) {
  295. tagged_node_handle new_tail(node_handle, tail.get_next_tag());
  296. tail_.compare_exchange_strong(tail, new_tail);
  297. return true;
  298. }
  299. }
  300. else {
  301. tagged_node_handle new_tail(pool.get_handle(next_ptr), tail.get_next_tag());
  302. tail_.compare_exchange_strong(tail, new_tail);
  303. }
  304. }
  305. }
  306. }
  307. #endif
  308. public:
  309. /** Pushes object t to the queue.
  310. *
  311. * \post object will be pushed to the queue, if internal node can be allocated
  312. * \returns true, if the push operation is successful.
  313. *
  314. * \note Not Thread-safe. If internal memory pool is exhausted and the memory pool is not fixed-sized, a new node will be allocated
  315. * from the OS. This may not be lock-free.
  316. * \throws if memory allocator throws
  317. * */
  318. bool unsynchronized_push(T const & t)
  319. {
  320. node * n = pool.template construct<false, false>(t, pool.null_handle());
  321. if (n == NULL)
  322. return false;
  323. for (;;) {
  324. tagged_node_handle tail = tail_.load(memory_order_relaxed);
  325. tagged_node_handle next = tail->next.load(memory_order_relaxed);
  326. node * next_ptr = next.get_ptr();
  327. if (next_ptr == 0) {
  328. tail->next.store(tagged_node_handle(n, next.get_next_tag()), memory_order_relaxed);
  329. tail_.store(tagged_node_handle(n, tail.get_next_tag()), memory_order_relaxed);
  330. return true;
  331. }
  332. else
  333. tail_.store(tagged_node_handle(next_ptr, tail.get_next_tag()), memory_order_relaxed);
  334. }
  335. }
  336. /** Pops object from queue.
  337. *
  338. * \post if pop operation is successful, object will be copied to ret.
  339. * \returns true, if the pop operation is successful, false if queue was empty.
  340. *
  341. * \note Thread-safe and non-blocking
  342. * */
  343. bool pop (T & ret)
  344. {
  345. return pop<T>(ret);
  346. }
  347. /** Pops object from queue.
  348. *
  349. * \pre type U must be constructible by T and copyable, or T must be convertible to U
  350. * \post if pop operation is successful, object will be copied to ret.
  351. * \returns true, if the pop operation is successful, false if queue was empty.
  352. *
  353. * \note Thread-safe and non-blocking
  354. * */
  355. template <typename U>
  356. bool pop (U & ret)
  357. {
  358. for (;;) {
  359. tagged_node_handle head = head_.load(memory_order_acquire);
  360. node * head_ptr = pool.get_pointer(head);
  361. tagged_node_handle tail = tail_.load(memory_order_acquire);
  362. tagged_node_handle next = head_ptr->next.load(memory_order_acquire);
  363. node * next_ptr = pool.get_pointer(next);
  364. tagged_node_handle head2 = head_.load(memory_order_acquire);
  365. if (BOOST_LIKELY(head == head2)) {
  366. if (pool.get_handle(head) == pool.get_handle(tail)) {
  367. if (next_ptr == 0)
  368. return false;
  369. tagged_node_handle new_tail(pool.get_handle(next), tail.get_next_tag());
  370. tail_.compare_exchange_strong(tail, new_tail);
  371. } else {
  372. if (next_ptr == 0)
  373. /* this check is not part of the original algorithm as published by michael and scott
  374. *
  375. * however we reuse the tagged_ptr part for the freelist and clear the next part during node
  376. * allocation. we can observe a null-pointer here.
  377. * */
  378. continue;
  379. detail::copy_payload(next_ptr->data, ret);
  380. tagged_node_handle new_head(pool.get_handle(next), head.get_next_tag());
  381. if (head_.compare_exchange_weak(head, new_head)) {
  382. pool.template destruct<true>(head);
  383. return true;
  384. }
  385. }
  386. }
  387. }
  388. }
  389. /** Pops object from queue.
  390. *
  391. * \post if pop operation is successful, object will be copied to ret.
  392. * \returns true, if the pop operation is successful, false if queue was empty.
  393. *
  394. * \note Not thread-safe, but non-blocking
  395. *
  396. * */
  397. bool unsynchronized_pop (T & ret)
  398. {
  399. return unsynchronized_pop<T>(ret);
  400. }
  401. /** Pops object from queue.
  402. *
  403. * \pre type U must be constructible by T and copyable, or T must be convertible to U
  404. * \post if pop operation is successful, object will be copied to ret.
  405. * \returns true, if the pop operation is successful, false if queue was empty.
  406. *
  407. * \note Not thread-safe, but non-blocking
  408. *
  409. * */
  410. template <typename U>
  411. bool unsynchronized_pop (U & ret)
  412. {
  413. for (;;) {
  414. tagged_node_handle head = head_.load(memory_order_relaxed);
  415. node * head_ptr = pool.get_pointer(head);
  416. tagged_node_handle tail = tail_.load(memory_order_relaxed);
  417. tagged_node_handle next = head_ptr->next.load(memory_order_relaxed);
  418. node * next_ptr = pool.get_pointer(next);
  419. if (pool.get_handle(head) == pool.get_handle(tail)) {
  420. if (next_ptr == 0)
  421. return false;
  422. tagged_node_handle new_tail(pool.get_handle(next), tail.get_next_tag());
  423. tail_.store(new_tail);
  424. } else {
  425. if (next_ptr == 0)
  426. /* this check is not part of the original algorithm as published by michael and scott
  427. *
  428. * however we reuse the tagged_ptr part for the freelist and clear the next part during node
  429. * allocation. we can observe a null-pointer here.
  430. * */
  431. continue;
  432. detail::copy_payload(next_ptr->data, ret);
  433. tagged_node_handle new_head(pool.get_handle(next), head.get_next_tag());
  434. head_.store(new_head);
  435. pool.template destruct<false>(head);
  436. return true;
  437. }
  438. }
  439. }
  440. /** consumes one element via a functor
  441. *
  442. * pops one element from the queue and applies the functor on this object
  443. *
  444. * \returns true, if one element was consumed
  445. *
  446. * \note Thread-safe and non-blocking, if functor is thread-safe and non-blocking
  447. * */
  448. template <typename Functor>
  449. bool consume_one(Functor & f)
  450. {
  451. T element;
  452. bool success = pop(element);
  453. if (success)
  454. f(element);
  455. return success;
  456. }
  457. /// \copydoc boost::lockfree::queue::consume_one(Functor & rhs)
  458. template <typename Functor>
  459. bool consume_one(Functor const & f)
  460. {
  461. T element;
  462. bool success = pop(element);
  463. if (success)
  464. f(element);
  465. return success;
  466. }
  467. /** consumes all elements via a functor
  468. *
  469. * sequentially pops all elements from the queue and applies the functor on each object
  470. *
  471. * \returns number of elements that are consumed
  472. *
  473. * \note Thread-safe and non-blocking, if functor is thread-safe and non-blocking
  474. * */
  475. template <typename Functor>
  476. size_t consume_all(Functor & f)
  477. {
  478. size_t element_count = 0;
  479. while (consume_one(f))
  480. element_count += 1;
  481. return element_count;
  482. }
  483. /// \copydoc boost::lockfree::queue::consume_all(Functor & rhs)
  484. template <typename Functor>
  485. size_t consume_all(Functor const & f)
  486. {
  487. size_t element_count = 0;
  488. while (consume_one(f))
  489. element_count += 1;
  490. return element_count;
  491. }
  492. private:
  493. #ifndef BOOST_DOXYGEN_INVOKED
  494. atomic<tagged_node_handle> head_;
  495. static const int padding_size = BOOST_LOCKFREE_CACHELINE_BYTES - sizeof(tagged_node_handle);
  496. char padding1[padding_size];
  497. atomic<tagged_node_handle> tail_;
  498. char padding2[padding_size];
  499. pool_t pool;
  500. #endif
  501. };
  502. } /* namespace lockfree */
  503. } /* namespace boost */
  504. #if defined(BOOST_INTEL) && (BOOST_INTEL_CXX_VERSION > 1000)
  505. #pragma warning(pop)
  506. #endif
  507. #if defined(_MSC_VER)
  508. #pragma warning(pop)
  509. #endif
  510. #endif /* BOOST_LOCKFREE_FIFO_HPP_INCLUDED */