boost/lockfree/ringbuffer.hpp

   1 //  lock-free single-producer/single-consumer ringbuffer
   2 //  this algorithm is implemented in various projects (linux kernel)
   3 //
   4 //  implementation for c++
   5 //
   6 //  Copyright (C) 2009 Tim Blechmann
   7 //
   8 //  Distributed under the Boost Software License, Version 1.0. (See
   9 //  accompanying file LICENSE_1_0.txt or copy at
  10 //  http://www.boost.org/LICENSE_1_0.txt)
  11
  12 //  Disclaimer: Not a Boost library.
  13
  14 #ifndef BOOST_LOCKFREE_RINGBUFFER_HPP_INCLUDED
  15 #define BOOST_LOCKFREE_RINGBUFFER_HPP_INCLUDED
  16 #include <boost/lockfree/detail/atomic.hpp>
  17 #include <boost/array.hpp>
  18 #include <boost/noncopyable.hpp>
  19 #include <boost/smart_ptr/scoped_array.hpp>
  20 #include <boost/range.hpp>
  21
  22 #include "detail/branch_hints.hpp"
  23 #include "detail/prefix.hpp"
  24
  25 #include <algorithm>
  26
  27 namespace boost
  28 {
  29 namespace lockfree
  30 {
  31
  32 namespace detail
  33 {
  34
  35 template <typename T>
  36 class ringbuffer_base:
  37     boost::noncopyable
  38 {
  39 #ifndef BOOST_DOXYGEN_INVOKED
  40     typedef std::size_t size_t;
  41     static const int padding_size = BOOST_LOCKFREE_CACHELINE_BYTES - sizeof(size_t);
  42     atomic<size_t> write_index_;
  43     char padding1[padding_size]; /* force read_index and write_index to different cache lines */
  44     atomic<size_t> read_index_;
  45
  46 protected:
  47     ringbuffer_base(void):
  48         write_index_(0), read_index_(0)
  49     {}
  50
  51     static size_t next_index(size_t arg, size_t max_size)
  52     {
  53         size_t ret = arg + 1;
  54         while (unlikely(ret >= max_size))
  55             ret -= max_size;
  56         return ret;
  57     }
  58
  59     static size_t read_available(size_t write_index, size_t read_index, size_t max_size)
  60     {
  61         if (write_index >= read_index)
  62             return write_index - read_index;
  63
  64         size_t ret = write_index + max_size - read_index;
  65         return ret;
  66     }
  67
  68     static size_t write_available(size_t write_index, size_t read_index, size_t max_size)
  69     {
  70         size_t ret = read_index - write_index - 1;
  71         if (write_index >= read_index)
  72             ret += max_size;
  73         return ret;
  74     }
  75
  76     bool enqueue(T const & t, T * buffer, size_t max_size)
  77     {
  78         size_t write_index = write_index_.load(memory_order_relaxed);  // only written from enqueue thread
  79         size_t next = next_index(write_index, max_size);
  80
  81         if (next == read_index_.load(memory_order_acquire))
  82             return false; /* ringbuffer is full */
  83
  84         buffer[write_index] = t;
  85
  86         write_index_.store(next, memory_order_release);
  87
  88         return true;
  89     }
  90
  91     size_t enqueue(const T * input_buffer, size_t input_count, T * internal_buffer, size_t max_size)
  92     {
  93         size_t write_index = write_index_.load(memory_order_relaxed);  // only written from enqueue thread
  94         const size_t read_index  = read_index_.load(memory_order_acquire);
  95         const size_t avail = write_available(write_index, read_index, max_size);
  96
  97         if (avail == 0)
  98             return 0;
  99
 100         input_count = std::min(input_count, avail);
 101
 102         size_t new_write_index = write_index + input_count;
 103
 104         if (write_index + input_count > max_size) {
 105             /* copy data in two sections */
 106             size_t count0 = max_size - write_index;
 107
 108             std::copy(input_buffer, input_buffer + count0, internal_buffer + write_index);
 109             std::copy(input_buffer + count0, input_buffer + input_count, internal_buffer);
 110             new_write_index -= max_size;
 111         } else {
 112             std::copy(input_buffer, input_buffer + input_count, internal_buffer + write_index);
 113
 114             if (new_write_index == max_size)
 115                 new_write_index = 0;
 116         }
 117
 118         write_index_.store(new_write_index, memory_order_release);
 119         return input_count;
 120     }
 121
 122     template <typename ConstIterator>
 123     ConstIterator enqueue(ConstIterator begin, ConstIterator end, T * internal_buffer, size_t max_size)
 124     {
 125         // FIXME: avoid std::distance and std::advance
 126
 127         size_t write_index = write_index_.load(memory_order_relaxed);  // only written from enqueue thread
 128         const size_t read_index  = read_index_.load(memory_order_acquire);
 129         const size_t avail = write_available(write_index, read_index, max_size);
 130
 131         if (avail == 0)
 132             return begin;
 133
 134         size_t input_count = std::distance(begin, end);
 135         input_count = std::min(input_count, avail);
 136
 137         size_t new_write_index = write_index + input_count;
 138
 139         ConstIterator last = begin;
 140         std::advance(last, input_count);
 141
 142         if (write_index + input_count > max_size) {
 143             /* copy data in two sections */
 144             size_t count0 = max_size - write_index;
 145             ConstIterator midpoint = begin;
 146             std::advance(midpoint, count0);
 147
 148             std::copy(begin, midpoint, internal_buffer + write_index);
 149             std::copy(midpoint, last, internal_buffer);
 150             new_write_index -= max_size;
 151         } else {
 152             std::copy(begin, last, internal_buffer + write_index);
 153
 154             if (new_write_index == max_size)
 155                 new_write_index = 0;
 156         }
 157
 158         write_index_.store(new_write_index, memory_order_release);
 159         return last;
 160     }
 161
 162     bool dequeue (T & ret, T * buffer, size_t max_size)
 163     {
 164         size_t write_index = write_index_.load(memory_order_acquire);
 165         size_t read_index  = read_index_.load(memory_order_relaxed); // only written from dequeue thread
 166         if (empty(write_index, read_index))
 167             return false;
 168
 169         ret = buffer[read_index];
 170         size_t next = next_index(read_index, max_size);
 171         read_index_.store(next, memory_order_release);
 172         return true;
 173     }
 174
 175     size_t dequeue (T * output_buffer, size_t output_count, const T * internal_buffer, size_t max_size)
 176     {
 177         const size_t write_index = write_index_.load(memory_order_acquire);
 178         size_t read_index = read_index_.load(memory_order_relaxed); // only written from dequeue thread
 179
 180         const size_t avail = read_available(write_index, read_index, max_size);
 181
 182         if (avail == 0)
 183             return 0;
 184
 185         output_count = std::min(output_count, avail);
 186
 187         size_t new_read_index = read_index + output_count;
 188
 189         if (read_index + output_count > max_size) {
 190             /* copy data in two sections */
 191             size_t count0 = max_size - read_index;
 192             size_t count1 = output_count - count0;
 193
 194             std::copy(internal_buffer + read_index, internal_buffer + max_size, output_buffer);
 195             std::copy(internal_buffer, internal_buffer + count1, output_buffer + count0);
 196
 197             new_read_index -= max_size;
 198         } else {
 199             std::copy(internal_buffer + read_index, internal_buffer + read_index + output_count, output_buffer);
 200             if (new_read_index == max_size)
 201                 new_read_index = 0;
 202         }
 203
 204         read_index_.store(new_read_index, memory_order_release);
 205         return output_count;
 206     }
 207
 208     template <typename OutputIterator>
 209     size_t dequeue (OutputIterator it, const T * internal_buffer, size_t max_size)
 210     {
 211         const size_t write_index = write_index_.load(memory_order_acquire);
 212         size_t read_index = read_index_.load(memory_order_relaxed); // only written from dequeue thread
 213
 214         const size_t avail = read_available(write_index, read_index, max_size);
 215         if (avail == 0)
 216             return 0;
 217
 218         size_t new_read_index = read_index + avail;
 219
 220         if (read_index + avail > max_size) {
 221             /* copy data in two sections */
 222             size_t count0 = max_size - read_index;
 223             size_t count1 = avail - count0;
 224
 225             std::copy(internal_buffer + read_index, internal_buffer + max_size, it);
 226             std::copy(internal_buffer, internal_buffer + count1, it);
 227
 228             new_read_index -= max_size;
 229         } else {
 230             std::copy(internal_buffer + read_index, internal_buffer + read_index + avail, it);
 231             if (new_read_index == max_size)
 232                 new_read_index = 0;
 233         }
 234
 235         read_index_.store(new_read_index, memory_order_release);
 236         return avail;
 237     }
 238 #endif
 239
 240
 241 public:
 242     /** reset the ringbuffer
 243      *
 244      * \warning Not thread-safe, use for debugging purposes only
 245      * */
 246     void reset(void)
 247     {
 248         write_index_.store(0, memory_order_relaxed);
 249         read_index_.store(0, memory_order_release);
 250     }
 251
 252     /** Check if the ringbuffer is empty
 253      *
 254      * \warning Not thread-safe, use for debugging purposes only
 255      * */
 256     bool empty(void)
 257     {
 258         return empty(write_index_.load(memory_order_relaxed), read_index_.load(memory_order_relaxed));
 259     }
 260
 261     //! \copydoc boost::lockfree::stack::is_lock_free
 262     bool is_lock_free(void) const
 263     {
 264         return write_index_.is_lock_free() && read_index_.is_lock_free();
 265     }
 266
 267 private:
 268     bool empty(size_t write_index, size_t read_index)
 269     {
 270         return write_index == read_index;
 271     }
 272 };
 273
 274 } /* namespace detail */
 275
 276 template <typename T, size_t max_size>
 277 class ringbuffer:
 278     public detail::ringbuffer_base<T>
 279 {
 280     typedef std::size_t size_t;
 281     boost::array<T, max_size> array_;
 282
 283 public:
 284     /** Enqueues object t to the ringbuffer. Enqueueing may fail, if the ringbuffer is full.
 285      *
 286      * \return true, if the enqueue operation is successful.
 287      *
 288      * \note Thread-safe and non-blocking
 289      * */
 290     bool enqueue(T const & t)
 291     {
 292         return detail::ringbuffer_base<T>::enqueue(t, array_.c_array(), max_size);
 293     }
 294
 295     /** Dequeue object from ringbuffer.
 296      *
 297      * If dequeue operation is successful, object is written to memory location denoted by ret.
 298      *
 299      * \return true, if the dequeue operation is successful, false if ringbuffer was empty.
 300      *
 301      * \note Thread-safe and non-blocking
 302      */
 303     bool dequeue(T & ret)
 304     {
 305         return detail::ringbuffer_base<T>::dequeue(ret, array_.c_array(), max_size);
 306     }
 307
 308     /** Enqueues size objects from the array t to the ringbuffer.
 309      *
 310      *  Will enqueue as many objects as there is space available
 311      *
 312      * \Returns number of enqueued items
 313      *
 314      * \note Thread-safe and non-blocking
 315      */
 316     size_t enqueue(T const * t, size_t size)
 317     {
 318         return detail::ringbuffer_base<T>::enqueue(t, size, array_.c_array(), max_size);
 319     }
 320
 321     /** Enqueues all objects from the array t to the ringbuffer.
 322      *
 323      *  Will enqueue as many objects as there is space available
 324      *
 325      * \Returns number of enqueued items
 326      *
 327      * \note Thread-safe and non-blocking
 328      */
 329     template <size_t size>
 330     size_t enqueue(T const (&t)[size])
 331     {
 332         return enqueue(t, size);
 333     }
 334
 335     /** Enqueues size objects from the iterator range [begin, end[ to the ringbuffer.
 336      *
 337      *  Enqueueing may fail, if the ringbuffer is full.
 338      *
 339      * \return iterator to the first element, which has not been enqueued
 340      *
 341      * \note Thread-safe and non-blocking
 342      */
 343     template <typename ConstIterator>
 344     ConstIterator enqueue(ConstIterator begin, ConstIterator end)
 345     {
 346         return detail::ringbuffer_base<T>::enqueue(begin, end, array_.c_array(), max_size);
 347     }
 348
 349     /** Dequeue a maximum of size objects from ringbuffer.
 350      *
 351      * If dequeue operation is successful, object is written to memory location denoted by ret.
 352      *
 353      * \return number of dequeued items
 354      *
 355      * \note Thread-safe and non-blocking
 356      * */
 357     /* @{ */
 358     size_t dequeue(T * ret, size_t size)
 359     {
 360         return detail::ringbuffer_base<T>::dequeue(ret, size, array_.c_array(), max_size);
 361     }
 362
 363     /** Enqueues all objects from the array t to the ringbuffer.
 364      *
 365      *  Will enqueue as many objects as there is space available
 366      *
 367      * \Returns number of enqueued items
 368      *
 369      * \note Thread-safe and non-blocking
 370      */
 371     template <size_t size>
 372     size_t dequeue(T (&t)[size])
 373     {
 374         return dequeue(t, size);
 375     }
 376
 377     /** Dequeue objects to the output iterator it
 378      *
 379      * \return number of dequeued items
 380      *
 381      * \note Thread-safe and non-blocking
 382      * */
 383     template <typename OutputIterator>
 384     size_t dequeue(OutputIterator it)
 385     {
 386         return detail::ringbuffer_base<T>::dequeue(it, array_.c_array(), max_size);
 387     }
 388 };
 389
 390 template <typename T>
 391 class ringbuffer<T, 0>:
 392     public detail::ringbuffer_base<T>
 393 {
 394     typedef std::size_t size_t;
 395     size_t max_size_;
 396     scoped_array<T> array_;
 397
 398 public:
 399     //! Constructs a ringbuffer for max_size elements
 400     explicit ringbuffer(size_t max_size):
 401         max_size_(max_size), array_(new T[max_size])
 402     {}
 403
 404     /** Enqueues object t to the ringbuffer. Enqueueing may fail, if the ringbuffer is full.
 405      *
 406      * \return true, if the enqueue operation is successful.
 407      *
 408      * \note Thread-safe and non-blocking
 409      * */
 410     bool enqueue(T const & t)
 411     {
 412         return detail::ringbuffer_base<T>::enqueue(t, array_.get(), max_size_);
 413     }
 414
 415     /** Dequeue object from ringbuffer.
 416      *
 417      * If dequeue operation is successful, object is written to memory location denoted by ret.
 418      *
 419      * \return true, if the dequeue operation is successful, false if ringbuffer was empty.
 420      *
 421      * \note Thread-safe and non-blocking
 422      */
 423     bool dequeue(T & ret)
 424     {
 425         return detail::ringbuffer_base<T>::dequeue(ret, array_.get(), max_size_);
 426     }
 427
 428     /** Enqueues size objects from the array t to the ringbuffer.
 429      *
 430      *  Will enqueue as many objects as there is space available
 431      *
 432      * \Returns number of enqueued items
 433      *
 434      * \note Thread-safe and non-blocking
 435      */
 436     size_t enqueue(T const * t, size_t size)
 437     {
 438         return detail::ringbuffer_base<T>::enqueue(t, size, array_.get(), max_size_);
 439     }
 440
 441     /** Enqueues all objects from the array t to the ringbuffer.
 442      *
 443      *  Will enqueue as many objects as there is space available
 444      *
 445      * \Returns number of enqueued items
 446      *
 447      * \note Thread-safe and non-blocking
 448      */
 449     template <size_t size>
 450     size_t enqueue(T const (&t)[size])
 451     {
 452         return enqueue(t, size);
 453     }
 454
 455     /** Enqueues size objects from the iterator range [begin, end[ to the ringbuffer.
 456      *
 457      *  Enqueueing may fail, if the ringbuffer is full.
 458      *
 459      * \return iterator to the first element, which has not been enqueued
 460      *
 461      * \note Thread-safe and non-blocking
 462      */
 463     template <typename ConstIterator>
 464     ConstIterator enqueue(ConstIterator begin, ConstIterator end)
 465     {
 466         return detail::ringbuffer_base<T>::enqueue(begin, end, array_.get(), max_size_);
 467     }
 468
 469     /** Dequeue a maximum of size objects from ringbuffer.
 470      *
 471      * If dequeue operation is successful, object is written to memory location denoted by ret.
 472      *
 473      * \return number of dequeued items
 474      *
 475      * \note Thread-safe and non-blocking
 476      * */
 477     size_t dequeue(T * ret, size_t size)
 478     {
 479         return detail::ringbuffer_base<T>::dequeue(ret, size, array_.get(), max_size_);
 480     }
 481
 482     /** Dequeue objects from ringbuffer.
 483      *
 484      * If dequeue operation is successful, object is written to memory location denoted by ret.
 485      *
 486      * \return number of dequeued items
 487      *
 488      * \note Thread-safe and non-blocking
 489      * */
 490     template <size_t size>
 491     size_t dequeue(T (&t)[size])
 492     {
 493         return dequeue(t, size);
 494     }
 495
 496     /** Dequeue objects to the output iterator it
 497      *
 498      * \return number of dequeued items
 499      *
 500      * \note Thread-safe and non-blocking
 501      * */
 502     template <typename OutputIterator>
 503     size_t dequeue(OutputIterator it)
 504     {
 505         return detail::ringbuffer_base<T>::dequeue(it, array_.get(), max_size_);
 506     }
 507 };
 508
 509
 510 } /* namespace lockfree */
 511 } /* namespace boost */
 512
 513
 514 #endif /* BOOST_LOCKFREE_RINGBUFFER_HPP_INCLUDED */