-//$$CDS-header$$
+/*
+ This file is a part of libcds - Concurrent Data Structures library
+
+ (C) Copyright Maxim Khizhinsky (libcds.dev@gmail.com) 2006-2016
+
+ Source code repo: http://github.com/khizmax/libcds/
+ Download: http://sourceforge.net/projects/libcds/files/
+
+ Redistribution and use in source and binary forms, with or without
+ modification, are permitted provided that the following conditions are met:
+
+ * Redistributions of source code must retain the above copyright notice, this
+ list of conditions and the following disclaimer.
+
+ * Redistributions in binary form must reproduce the above copyright notice,
+ this list of conditions and the following disclaimer in the documentation
+ and/or other materials provided with the distribution.
+
+ THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+ FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+ SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+ CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+ OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+*/
#ifndef CDSLIB_CONTAINER_VYUKOV_MPMC_CYCLE_QUEUE_H
#define CDSLIB_CONTAINER_VYUKOV_MPMC_CYCLE_QUEUE_H
/// Padding for internal critical atomic data. Default is \p opt::cache_line_padding
enum { padding = opt::cache_line_padding };
+
+ /// Back-off strategy
+ typedef cds::backoff::Default back_off;
+
+ /// Single-consumer version
+ /**
+ For single-consumer version of algorithm some additional functions
+
+ (\p front(), \p pop_front()) is available.
+
+ Default is \p false
+ */
+ static CDS_CONSTEXPR bool const single_consumer = false;
};
/// Metafunction converting option list to \p vyukov_queue::traits
After an item is dequeued, \p value_cleaner cleans the cell that the item has been occupied.
If \p T is a complex type, \p value_cleaner can be an useful feature.
Default value is \ref opt::v::destruct_cleaner
+ - \p opt::back_off - back-off strategy used. If the option is not specified, the \p cds::backoff::Default is used.
- \p opt::item_counter - the type of item counting feature. Default is \p cds::atomicity::empty_item_counter (item counting disabled)
To enable item counting use \p cds::atomicity::item_counter
- \p opt::padding - padding for internal critical atomic data. Default is \p opt::cache_line_padding
No dynamic memory allocation/management during operation. Producers and consumers are separated from each other (as in the two-lock queue),
i.e. do not touch the same data while queue is not empty.
+ There is multiple producer/single consumer version \p cds::container::VyukovMPSCCycleQueue
+ that supports \p front() and \p pop_front() functions.
+
Source:
- http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue
typedef typename traits::item_counter item_counter; ///< Item counter type
typedef typename traits::memory_model memory_model; ///< Memory ordering. See cds::opt::memory_model option
typedef typename traits::value_cleaner value_cleaner; ///< Value cleaner, see \p vyukov_queue::traits::value_cleaner
+ typedef typename traits::back_off back_off; ///< back-off strategy
+
+ /// \p true for single-consumer version, \p false otherwise
+ static CDS_CONSTEXPR bool const c_single_consumer = traits::single_consumer;
/// Rebind template arguments
template <typename T2, typename Traits2>
protected:
//@cond
buffer m_buffer;
- typename opt::details::apply_padding< buffer, traits::padding >::padding_type pad1_;
size_t const m_nBufferMask;
+ typename opt::details::apply_padding< size_t, traits::padding >::padding_type pad1_;
sequence_type m_posEnqueue;
typename opt::details::apply_padding< sequence_type, traits::padding >::padding_type pad2_;
sequence_type m_posDequeue;
bool enqueue_with(Func f)
{
cell_type* cell;
- size_t pos = m_posEnqueue.load(memory_model::memory_order_relaxed);
+ back_off bkoff;
+ size_t pos = m_posEnqueue.load(memory_model::memory_order_relaxed);
for (;;)
{
cell = &m_buffer[pos & m_nBufferMask];
if ( m_posEnqueue.compare_exchange_weak(pos, pos + 1, memory_model::memory_order_relaxed, atomics::memory_order_relaxed ))
break;
}
- else if (dif < 0)
- return false;
+ else if (dif < 0) {
+ // Queue full?
+ if ( pos - m_posDequeue.load( memory_model::memory_order_relaxed ) == capacity() )
+ return false; // queue full
+ bkoff();
+ pos = m_posEnqueue.load( memory_model::memory_order_relaxed );
+ }
else
pos = m_posEnqueue.load(memory_model::memory_order_relaxed);
}
/// Enqueues data of type \ref value_type constructed with <tt>std::forward<Args>(args)...</tt>
template <typename... Args>
bool emplace( Args&&... args )
- {
+ {
#if (CDS_COMPILER == CDS_COMPILER_GCC) && (CDS_COMPILER_VERSION < 40900)
- //work around unsupported feature in g++ 4.8 for forwarding parameter packs to lambda.
- return enqueue_with ( std::bind([]( value_type& dest,Args ... args ){ new ( &dest ) value_type( std::forward<Args>(args)... );}, std::placeholders::_1 ,args...));
-#else
- return enqueue_with( [&args ...]( value_type& dest ){ new ( &dest ) value_type( std::forward<Args>(args)... ); });
-#endif
+ //work around unsupported feature in g++ 4.8 for forwarding parameter packs to lambda.
+ return enqueue_with ( std::bind([]( value_type& dest,Args ... args ){ new ( &dest ) value_type( std::forward<Args>(args)... );}, std::placeholders::_1 ,args...));
+#else
+ return enqueue_with( [&args ...]( value_type& dest ){ new ( &dest ) value_type( std::forward<Args>(args)... ); });
+#endif
}
/// Dequeues a value using a functor
bool dequeue_with( Func f )
{
cell_type * cell;
- size_t pos = m_posDequeue.load(memory_model::memory_order_relaxed);
+ back_off bkoff;
+ size_t pos = m_posDequeue.load( memory_model::memory_order_relaxed );
for (;;)
{
cell = &m_buffer[pos & m_nBufferMask];
if ( m_posDequeue.compare_exchange_weak(pos, pos + 1, memory_model::memory_order_relaxed, atomics::memory_order_relaxed))
break;
}
- else if (dif < 0)
- return false;
+ else if (dif < 0) {
+ // Queue empty?
+ if ( pos - m_posEnqueue.load( memory_model::memory_order_relaxed ) == 0 )
+ return false; // queue empty
+ bkoff();
+ pos = m_posDequeue.load( memory_model::memory_order_relaxed );
+ }
else
pos = m_posDequeue.load(memory_model::memory_order_relaxed);
}
return dequeue_with( f );
}
+ /// Returns a pointer to top element of the queue or \p nullptr if queue is empty (only for single-consumer version)
+ template <bool SC = c_single_consumer >
+ typename std::enable_if<SC, value_type *>::type front()
+ {
+ static_assert( c_single_consumer, "front() is enabled only if traits::single_consumer is true");
+
+ cell_type * cell;
+ back_off bkoff;
+
+ size_t pos = m_posDequeue.load( memory_model::memory_order_relaxed );
+ for ( ;;)
+ {
+ cell = &m_buffer[pos & m_nBufferMask];
+ size_t seq = cell->sequence.load( memory_model::memory_order_acquire );
+ intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos + 1);
+
+ if ( dif == 0 )
+ return &cell->data;
+ else if ( dif < 0 ) {
+ // Queue empty?
+ if ( pos - m_posEnqueue.load( memory_model::memory_order_relaxed ) == 0 )
+ return nullptr; // queue empty
+ bkoff();
+ pos = m_posDequeue.load( memory_model::memory_order_relaxed );
+ }
+ else
+ pos = m_posDequeue.load( memory_model::memory_order_relaxed );
+ }
+ }
+
+ /// Pops top element; returns \p true if queue is not empty, \p false otherwise (only for single-consumer version)
+ template <bool SC = c_single_consumer >
+ typename std::enable_if<SC, bool>::type pop_front()
+ {
+ return dequeue_with( []( value_type& ) {} );
+ }
+
/// Checks if the queue is empty
bool empty() const
{
const cell_type * cell;
- size_t pos = m_posDequeue.load(memory_model::memory_order_relaxed);
+ back_off bkoff;
+ size_t pos = m_posDequeue.load(memory_model::memory_order_relaxed);
for (;;)
{
cell = &m_buffer[pos & m_nBufferMask];
if (dif == 0)
return false;
- else if (dif < 0)
- return true;
- else
- pos = m_posDequeue.load(memory_model::memory_order_relaxed);
+ else if (dif < 0) {
+ if ( pos - m_posEnqueue.load( memory_model::memory_order_relaxed ) == 0 )
+ return true;
+ }
+ bkoff();
+ pos = m_posDequeue.load(memory_model::memory_order_relaxed);
}
}
return m_buffer.capacity();
}
};
+
+ //@cond
+ namespace vyukov_queue {
+
+ template <typename Traits>
+ struct single_consumer_traits : public Traits
+ {
+ static CDS_CONSTEXPR bool const single_consumer = true;
+ };
+ } // namespace vyukov_queue
+ //@endcond
+
+ /// Vyukov's queue multiple producer - single consumer version
+
+ template <typename T, typename Traits = vyukov_queue::traits >
+ using VyukovMPSCCycleQueue = VyukovMPMCCycleQueue< T, vyukov_queue::single_consumer_traits<Traits> >;
+
}} // namespace cds::container
#endif // #ifndef CDSLIB_CONTAINER_VYUKOV_MPMC_CYCLE_QUEUE_H
projects / libcds.git / blobdiff