Program Listing for File SPSCQueue.h

↰ Return to documentation for file (ADS/concurrent/SPSCQueue.h)

/*--------------------------------------------------------------------------*/
/*---------------------------- File SPSCQueue.h ----------------------------*/
/*--------------------------------------------------------------------------*/
/*--------------------------------------------------------------------------*/
/*----------------------------- DEFINITIONS --------------------------------*/
/*--------------------------------------------------------------------------*/

#ifndef __ADS_SPSCQueue
 #define __ADS_SPSCQueue
                      /* self-identification: #endif constant */

/*--------------------------------------------------------------------------*/
/*------------------------------ INCLUDES ----------------------------------*/
/*--------------------------------------------------------------------------*/

#include "ADS/Contract.h"
#include "ADS/concurrent/EventCount.h"

#include <algorithm>
#include <atomic>
#include <bit>
#include <cstddef>
#include <memory>
#include <new>
#include <optional>
#include <utility>

/*--------------------------------------------------------------------------*/
/*--------------------------- NAMESPACE ------------------------------------*/
/*--------------------------------------------------------------------------*/

namespace ADS {

/*--------------------------------------------------------------------------*/
/*---------------------------- CLASS SPSCQueue -----------------------------*/
/*--------------------------------------------------------------------------*/

template< class T >
class SPSCQueue {

/*--------------------------------------------------------------------------*/
 public:

 using Item = T;

 explicit SPSCQueue( std::size_t capacity );

 SPSCQueue( const SPSCQueue & ) = delete;
 SPSCQueue & operator=( const SPSCQueue & ) = delete;

 ~SPSCQueue();

 bool push( const T & v ) { return( emplace( v ) ); }
 bool push( T && v ) { return( emplace( std::move( v ) ) ); }

 template< class ... Args >
 bool emplace( Args && ... args );

 std::optional< T > pop( void );

 bool tryPop( T & x );

 void notify( void ) noexcept { f_wait.notify(); }

 void waitPop( T & x ) { f_wait.await( [ & ] { return( tryPop( x ) ); } ); }

 [[nodiscard]] std::size_t size( void ) const;

 [[nodiscard]] std::size_t capacity( void ) const { return( f_mask + 1 ); }

 private:
 static constexpr std::size_t line = 64;
 static constexpr std::size_t pad = ( line + sizeof( T ) - 1 ) / sizeof( T );

 [[nodiscard]] T * at( std::size_t i ) const {
  return( f_A + pad + ( i & f_mask ) );
  }

 void refresh( std::size_t h );

 [[nodiscard]] std::size_t batch( void ) const {
  return( std::min< std::size_t >( 64 , capacity() / 2 ) );
  }

 static void relax( void );

 static void prefetch( const T * p ) {
#if defined( __GNUC__ ) || defined( __clang__ )
  __builtin_prefetch( p );
#else
  (void) p;
#endif
  }

 // the queue begins on a pair of lines, which the processor may fetch
 // together (the adjacent line prefetcher of x86), so that each pair
 // holds the line of the places and the tail, the copy of the head and
 // the head, the copy of the tail and nothing: the copy that a thread
 // rewrites while it spins shares no pair with the index of the other
 alignas( 2 * line ) T * f_A;
 std::size_t f_mask;
 alignas( line ) std::atomic< std::size_t > f_tail{ 0 };
 alignas( line ) std::size_t f_headSeen = 0;
 alignas( line ) std::atomic< std::size_t > f_head{ 0 };
 alignas( line ) std::size_t f_tailSeen = 0;
 bool f_streaming = false;
 alignas( line ) char f_end = 0;
 EventCount f_wait;
 };  // end( class SPSCQueue )

/*--------------------------------------------------------------------------*/

}  // end( namespace ADS )

/*--------------------------------------------------------------------------*/
/*-------------------------- IMPLEMENTATION --------------------------------*/
/*--------------------------------------------------------------------------*/

#include "ADS/concurrent/impl/SPSCQueue.tpp"

/*--------------------------------------------------------------------------*/

#endif  /* SPSCQueue.h included */

/*--------------------------------------------------------------------------*/
/*-------------------------- End File SPSCQueue.h --------------------------*/
/*--------------------------------------------------------------------------*/