Program Listing for File MPSCQueue.h

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

/*--------------------------------------------------------------------------*/
/*---------------------------- File MPSCQueue.h ----------------------------*/
/*--------------------------------------------------------------------------*/
/*--------------------------------------------------------------------------*/
/*----------------------------- DEFINITIONS --------------------------------*/
/*--------------------------------------------------------------------------*/

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

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

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

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

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

namespace ADS {

/*--------------------------------------------------------------------------*/
/*---------------------------- CLASS MPSCQueue -----------------------------*/
/*--------------------------------------------------------------------------*/

template< class T , std::size_t B = 1024 >
class MPSCQueue {
 static_assert( ( B >= 2 ) && ( ( B & ( B - 1 ) ) == 0 ) ,
                "a block has 2^k places" );

 static constexpr std::size_t line = 64;
 static constexpr std::size_t want = B < 64 ? B : 64;
 static constexpr int pauses = 64;

 struct Block {
  alignas( 2 * line ) std::atomic< std::size_t > filled{ 0 };
  std::atomic< Block * > next{ nullptr };
  Block * spare = nullptr;
  alignas( 2 * line ) unsigned char raw[ B * sizeof( T ) ];
  [[nodiscard]] T * at( std::size_t i ) {
   return( std::launder( reinterpret_cast< T * >( raw ) ) + i );
   }
  [[nodiscard]] void * slot( std::size_t i ) {
   return( raw + i * sizeof( T ) );
   }
  };

 struct Lane {
  alignas( 2 * line ) Block * in;
  std::size_t pos = 0;
  Block * spare = nullptr;
  alignas( 2 * line ) Block * out;
  std::size_t at = 0;
  bool streaming = false;
  alignas( 2 * line ) std::atomic< Block * > given{ nullptr };
  Lane * nextLane = nullptr;
  EventCount * wait;
  explicit Lane( EventCount * w ) : in( new Block ) , out( in ) ,
                                    wait( w ) {}
  };

/*--------------------------------------------------------------------------*/
 public:
 using Item = T;

 class Producer {
  public:
  Producer( void ) = default;
  Producer( Producer && o ) noexcept
   : f_lane( std::exchange( o.f_lane , nullptr ) ) {}
  Producer & operator=( Producer && o ) noexcept {
   f_lane = std::exchange( o.f_lane , nullptr );
   return( *this );
   }

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

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

  void notify( void ) {
   ADS_PRE( f_lane != nullptr );
   f_lane->wait->notify();
   }

  private:
  friend class MPSCQueue;
  explicit Producer( Lane * l ) : f_lane( l ) {}
  Lane * f_lane = nullptr;
  };  // end( class Producer )

 MPSCQueue( void ) = default;
 MPSCQueue( const MPSCQueue & ) = delete;
 MPSCQueue & operator=( const MPSCQueue & ) = delete;

 ~MPSCQueue();

 [[nodiscard]] Producer producer( void );

 bool tryPop( T & x );

 std::size_t popInto( T * out , std::size_t n );

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

 std::size_t waitPopInto( T * out , std::size_t n ) {
  ADS_PRE( n >= 1 );
  std::size_t k = 0;
  f_wait.await( [ & ] { return( ( k = popInto( out , n ) ) > 0 ); } );
  return( k );
  }

 [[nodiscard]] bool empty( void );

/*--------------------------------------------------------------------------*/
 private:
 std::size_t drain( Lane * l , T * out , std::size_t n );

 static void giveBack( Lane * l , Block * b );

 static void relax( void );

 std::atomic< Lane * > f_lanes{ nullptr };
 Lane * f_at = nullptr;
 EventCount f_wait;
 };  // end( class MPSCQueue )

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

}  // end( namespace ADS )

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

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

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

#endif  /* MPSCQueue.h included */

/*--------------------------------------------------------------------------*/
/*-------------------------- End File MPSCQueue.h --------------------------*/
/*--------------------------------------------------------------------------*/