medfall

nonblocking_fixed_mpsc_queue.h (1529B)

---

 #pragma once
 
 #include "intrinsics.h"
 
 #include "platform_atomic.h"
 #include "platform_mutex.h"
 #include "platform_thread.h"
 
 template< typename T, size_t N >
 class NonblockingFixedMPSCQueue {
 public:
 	NonblockingFixedMPSCQueue() {
 		store_relaxed( &num_elems, 0 );
 		head = 0;
 		reader_acquired = false;
 		mutex_init( &mutex );
 	}
 
 	~NonblockingFixedMPSCQueue() {
 		mutex_destroy( &mutex );
 	}
 
 	bool enqueue( const T & x ) {
 		if( load_acquire( &num_elems ) >= N ) return false;
 		SCOPED_MUTEX_LOCK( &mutex );
 		u32 n = load_relaxed( &num_elems );
 		if( n >= N ) return false;
 
 		elems[ ( head + n ) % N ] = x;
 		store_release( &num_elems, n + 1 );
 
 		return true;
 	}
 
 	T * dequeue_acquire() {
 		ASSERT( !reader_acquired );
 
 		if( load_acquire( &num_elems ) == 0 ) return NULL;
 
 		reader_acquired = true;
 		return elems + head;
 	}
 
 	void dequeue_release() {
 		ASSERT( reader_acquired );
 
 		SCOPED_MUTEX_LOCK( &mutex );
 
 		head = ( head + 1 ) % N;
 		fetch_sub_release( &num_elems, 1 );
 		reader_acquired = false;
 	}
 
 	void enqueue_spin( const T & x ) {
 		int attempt = 0;
 		while( !enqueue( x ) ) {
 			const int ATTEMPTS_BEFORE_YIELD = 128;
 			if( attempt < ATTEMPTS_BEFORE_YIELD ) {
 				attempt++;
 			}
 			else {
 				thread_yield();
 			}
 		}
 	}
 
 	bool dequeue( T * x ) {
 		T * t = dequeue_acquire();
 		if( t == NULL ) return false;
 		*x = *t;
 		dequeue_release();
 		return true;
 	}
 
 private:
 	T elems[ N ];
 
 	CACHE_LINE_PADDING;
 
 	u32 head;
 	bool reader_acquired;
 
 	CACHE_LINE_PADDING;
 
 	atomic_u32 num_elems;
 	Mutex mutex;
 };