- \p opt::memory_model - C++ memory ordering model. Can be \p opt::v::relaxed_ordering (relaxed memory model, the default)
or \p opt::v::sequential_consistent (sequentially consisnent memory model).
- Example: declare \p %WeakRingBuffer with static iternal buffer of size 1024:
+ Example: declare \p %WeakRingBuffer with static iternal buffer for 1024 objects:
\code
typedef cds::container::WeakRingBuffer< Foo,
typename cds::container::weak_ringbuffer::make_traits<
There are a specialization \ref cds_nonintrusive_WeakRingBuffer_void "WeakRingBuffer<void, Traits>"
that is not a queue but a "memory pool" between producer and consumer threads.
- \p WeakRingBuffer<void> supports data of different size.
+ \p WeakRingBuffer<void> supports variable-sized data.
+
+ @warning: \p %WeakRingBuffer is developed for 64-bit architecture.
+ On 32-bit platform an integer overflow of internal counters is possible.
*/
template <typename T, typename Traits = weak_ringbuffer::traits>
class WeakRingBuffer: public cds::bounded_container
/// Single-producer single-consumer ring buffer for untyped variable-sized data
/** @ingroup cds_nonintrusive_queue
@anchor cds_nonintrusive_WeakRingBuffer_void
+
+ This SPSC ring-buffer is intended for data of variable size. The producer
+ allocates a buffer from ring, you fill it with data and pushes them back to ring.
+ The consumer thread reads data from front-end and then pops them:
+ \code
+ // allocates 1M ring buffer
+ WeakRingBuffer<void> theRing( 1024 * 1024 );
+
+ void producer_thread()
+ {
+ // Get data of size N bytes
+ size_t size;
+ void* data;
+
+ while ( true ) {
+ // Get external data
+ std::tie( data, size ) = get_data();
+
+ if ( data == nullptr )
+ break;
+
+ // Allocates a buffer from the ring
+ void* buf = theRing.back( size );
+ if ( !buf ) {
+ std::cout << "The ring is full" << std::endl;
+ break;
+ }
+
+ memcpy( buf, data, size );
+
+ // Push data into the ring
+ theRing.push_back();
+ }
+ }
+
+ void consumer_thread()
+ {
+ while ( true ) {
+ auto buf = theRing.front();
+
+ if ( buf.first == nullptr ) {
+ std::cout << "The ring is empty" << std::endl;
+ break;
+ }
+
+ // Process data
+ process_data( buf.first, buf.second );
+
+ // Free buffer
+ theRing.pop_front();
+ }
+ }
+ \endcode
+
+ @warning: \p %WeakRingBuffer is developed for 64-bit architecture.
+ On 32-bit platform an integer overflow of internal counters is possible.
*/
#ifdef CDS_DOXYGEN_INVOKED
template <typename Traits = weak_ringbuffer::traits>
}
/// [producer] Reserve \p size bytes
+ /**
+ The function returns a pointer to reserved buffer of \p size bytes.
+ If no enough space in the ring buffer the function returns \p nullptr.
+
+ After successful \p %back() you should fill the buffer provided and call \p push_back():
+ \code
+ // allocates 1M ring buffer
+ WeakRingBuffer<void> theRing( 1024 * 1024 );
+
+ void producer_thread()
+ {
+ // Get data of size N bytes
+ size_t size;1
+ void* data;
+
+ while ( true ) {
+ // Get external data
+ std::tie( data, size ) = get_data();
+
+ if ( data == nullptr )
+ break;
+
+ // Allocates a buffer from the ring
+ void* buf = theRing.back( size );
+ if ( !buf ) {
+ std::cout << "The ring is full" << std::endl;
+ break;
+ }
+
+ memcpy( buf, data, size );
+
+ // Push data into the ring
+ theRing.push_back();
+ }
+ }
+ \endcode
+ */
void* back( size_t size )
{
+ assert( size > 0 );
+
// Any data is rounded to 8-byte boundary
size_t real_size = calc_real_size( size );
}
}
+ back_.store( back, memory_model::memory_order_release );
reserved = buffer_.buffer();
}
// reserve and store size
- uint8_t* reserved = buffer_.buffer() + buffer_.mod( back );
*reinterpret_cast<size_t*>( reserved ) = size;
return reinterpret_cast<void*>( reserved + sizeof( size_t ) );
}
/// [producer] Push reserved bytes into ring
+ /**
+ The function pushes reserved buffer into the ring. Afte this call,
+ the buffer becomes visible by a consumer:
+ \code
+ // allocates 1M ring buffer
+ WeakRingBuffer<void> theRing( 1024 * 1024 );
+
+ void producer_thread()
+ {
+ // Get data of size N bytes
+ size_t size;1
+ void* data;
+
+ while ( true ) {
+ // Get external data
+ std::tie( data, size ) = get_data();
+
+ if ( data == nullptr )
+ break;
+
+ // Allocates a buffer from the ring
+ void* buf = theRing.back( size );
+ if ( !buf ) {
+ std::cout << "The ring is full" << std::endl;
+ break;
+ }
+
+ memcpy( buf, data, size );
+
+ // Push data into the ring
+ theRing.push_back();
+ }
+ }
+ \endcode
+ */
void push_back()
{
size_t back = back_.load( memory_model::memory_order_relaxed );
}
/// [producer] Push \p data of \p size bytes into ring
+ /**
+ This function invokes \p back( size ), \p memcpy( buf, data, size )
+ and \p push_back() in one call.
+ */
bool push_back( void const* data, size_t size )
{
void* buf = back( size );
}
/// [consumer] Get top data from the ring
+ /**
+ If the ring is empty, the function returns \p nullptr in \p std:pair::first.
+ */
std::pair<void*, size_t> front()
{
size_t front = front_.load( memory_model::memory_order_relaxed );
size = *reinterpret_cast<size_t*>( buf );
assert( !is_tail( size ) );
+ assert( buf == buffer_.buffer() );
}
#ifdef _DEBUG
}
#endif
- return std::make_pair( reinterpret_cast<void*>( buf + sizeof( size_t ) ), size );
+ return std::make_pair( reinterpret_cast<void*>( buf + sizeof( size_t )), size );
}
/// [consumer] Pops top data
+ /**
+ Typical consumer workloop:
+ \code
+ // allocates 1M ring buffer
+ WeakRingBuffer<void> theRing( 1024 * 1024 );
+
+ void consumer_thread()
+ {
+ while ( true ) {
+ auto buf = theRing.front();
+
+ if ( buf.first == nullptr ) {
+ std::cout << "The ring is empty" << std::endl;
+ break;
+ }
+
+ // Process data
+ process_data( buf.first, buf.second );
+
+ // Free buffer
+ theRing.pop_front();
+ }
+ }
+ \endcode
+ */
bool pop_front()
{
size_t front = front_.load( memory_model::memory_order_relaxed );
assert( ( reinterpret_cast<uintptr_t>( buf ) & ( sizeof( uintptr_t ) - 1 ) ) == 0 );
size_t size = *reinterpret_cast<size_t*>( buf );
- assert( !is_tail( size ) );
-
- size_t real_size = calc_real_size( size );
+ size_t real_size = calc_real_size( untail( size ));
#ifdef _DEBUG
if ( cback_ - front < real_size ) {
}
private:
+ //@cond
static size_t calc_real_size( size_t size )
{
size_t real_size = (( size + sizeof( uintptr_t ) - 1 ) & ~( sizeof( uintptr_t ) - 1 )) + sizeof( size_t );
return size | ( size_t( 1 ) << ( sizeof( size_t ) * 8 - 1 ));
}
+ static size_t untail( size_t size )
+ {
+ return size & (( size_t( 1 ) << ( sizeof( size_t ) * 8 - 1 ) ) - 1);
+ }
+ //@endcond
+
private:
//@cond
atomics::atomic<size_t> front_;
projects / libcds.git / blobdiff