Here's an implementation of a lock-free multi-reader/multi-writer array-based queue

    template <typename T>    class channel    {      typedef T value_t;      typedef boost::optional<value_t> optional_t;      typedef std::queue<value_t> queue_t;      typedef boost::mutex mutex_t;      typedef mutex_t::scoped_lock lock_t;      typedef boost::condition condition_t;      // Input-output elements ================================================      // Any operation over the data passes through the      // data mutex. This allows sync of both input and      // output threads.      queue_t data;       volatile bool is_over;      mutex_t data_mutex;       condition_t written_to;      // Reference counting elements ==========================================      // Any operation over the handles passes through       // the handle mutex.            mutex_t handle_mutex;      volatile unsigned output_handles;      volatile unsigned input_handles;      volatile bool is_silent;    public:      // Initialize a clean channel      channel() :         is_over(false),         output_handles(0),         input_handles(0),         is_silent(false)      {      }      void acquire(bool input)       {        lock_t lock(handle_mutex);        if (input)           ++input_handles;        else          ++output_handles;      }      // Release a reference to the flow.      // - If all threads cease to reference the flow, then      //   the memory is deallocated with operator delete      // - If all input threads cease to reference the flow,      //   sets the is-over flag and wakes up an output       //   thread.      // - If all output threads cease to reference the      //   flow, sets the is-silent flag.      void release(bool input)      {        bool should_delete = false;        {          lock_t lock(handle_mutex);          if (input)             {               if (--input_handles == 0)                 {                  lock_t lock(data_mutex);                  is_over = true;                  written_to.notify_all();                }            }          else            {              if (--output_handles == 0)                {                  is_silent = true;                }            }          should_delete = (input_handles + output_handles == 0);        }        if (should_delete)           {            delete this;          }      }      // Is the flow silent?      bool silent()      {        lock_t lock(handle_mutex);        return is_silent;      }      // Add an element to the flow.      // - If any output threads were waiting for input,       //   one of them is reactivated and reads the input.      // - Otherwise, stores the element for later collection.      void add(const value_t &t)       {        lock_t lock(data_mutex);        data.push(t);        written_to.notify_one();      }      // Attempt to read an element from the flow.      // - If the flow is over, leaves the argument untouched.      // - If the flow is currently empty, but there are still      //   input threads, blocks until input (or an is-over      //   signal) arrives.      void get(optional_t &t)      {        lock_t lock(data_mutex);        while (!is_over && data.empty())          {            written_to.wait(lock);          }        if (is_over && data.empty())           {            return;          }        t = data.front();        data.pop();      }    };

STATIC_ASSERT(sizeof(T) == 4);STATIC_ASSERT(sizeof(unsigned long) == 4);STATIC_ASSERT(sizeof(unsigned long long) == 8);

template<typename T, std::size_t Size>struct element_traits{    typedef T *element_type;    STATIC_ASSERT(sizeof(element_type) == 4);    static void create_at_location(const T &x, void *location) { *static_cast<T **>(location) = new T(x); }    static void destroy_at_location(void *location) { delete *static_cast<T **>(location); }    static T &get(void *location) { return **static_cast<T **>(location); }};template<typename T>struct element_traits<T, 4>{    typedef T element_type;    static void create_at_location(const T &x, void *location) { new(location) T(x); } // placement new with copy-constructor    static void destroy_at_location(void *location) { static_cast<T *>(location)->~T(); } // manual destructor call    static T &get(void *location) { return *static_cast<T *>(location); }};

#if !defined PROCESSOR_H#define PROCESSOR_H// Note: MSVC may require /Oi compiler option// CPU load/store reordering barriers and fast yield// TODO: For other than _MSC_VER and __ICL#if defined _M_IA64 // TODO: Probably no IA64 support neededextern "C" __mf#pragma intrinsic(__mf)#define MemoryBarrier __mfextern "C" void __yield(void);#pragma intrinsic(__yield)#define YieldProcessor __yield#else#if !defined __ICL || __ICC || __ECCextern "C" void _mm_mfence(void);#pragma intrinsic(_mm_mfence)extern "C" void _mm_lfence(void);#pragma intrinsic(_mm_lfence)extern "C" void _mm_sfence(void);#pragma intrinsic(_mm_sfence)extern "C" void _mm_pause(void);#pragma intrinsic(_mm_pause)#endif#define YieldProcessor _mm_pause#if defined _M_AMD64 // TODO: Check if this is OK if load barrier neededextern "C" void __faststorefence(void);#pragma intrinsic(__faststorefence)#define MemoryBarrier __faststorefence#elif defined _M_IX86/*__forceinline void MemoryBarrier(void){	long Barrier;	__asm xchg Barrier, eax}*/#define MemoryBarrier _mm_mfence#else#error "Unsupported environment"#endif#endif// Compiler load/store reordering barriers// TODO: Check if these are implied by the CPU barriers; __ECC should use explicit ...with acquire/release instructions instead#if defined __ICL || __ICC || __ECC#define _ReadWriteBarrier __memory_barrier#define _ReadBarrier __memory_barrier#define _WriteBarrier __memory_barrier#elif defined _MSC_VERextern "C" void _ReadWriteBarrier(void);#pragma intrinsic(_ReadWriteBarrier)extern "C" void _ReadBarrier(void);#pragma intrinsic(_ReadBarrier)extern "C" void _WriteBarrier(void);#pragma intrinsic(_WriteBarrier)#elif defined __GNUC__ // TODO: Other options for read- or write-only barriers?#define _ReadWriteBarrier() __asm__ __volatile__("" ::: "memory")#define _ReadBarrier() __asm__ __volatile__("" ::: "memory")#define _WriteBarrier() __asm__ __volatile__("" ::: "memory")#else#error "Unsupported environment"#endiftemplate<typename T>__forceinline T Acquire(T volatile const &p){	T t = p;	_ReadWriteBarrier();	return t;}template<typename S, typename T>__forceinline void Release(volatile S &p, T const t){	_ReadWriteBarrier();	p = t;}// Interlocked intrinsics// TODO: For other than _MSC_VER and __ICL; note that __GNUC__ __sync_bool_compare_and_swap does the comparison but _Interlocked... doesn't#if !defined __ICL || __ICC || __ECCextern "C" long _InterlockedCompareExchange(volatile unsigned long *, unsigned long, unsigned long);#pragma intrinsic(_InterlockedCompareExchange)extern "C" long long _InterlockedCompareExchange64(volatile unsigned long long *, unsigned long long, unsigned long long);#pragma intrinsic(_InterlockedCompareExchange64)#endif// Aligned memory allocator and deallocator#if !defined __ICL || __ICC || __ECC#if defined _MSC_VER#include <malloc.h>#elif defined __GNUC__#include <mm_malloc.h>#else#error "Unsupported environment"#endif#endiftemplate<bool> struct CompileError;template<> struct CompileError<true>{};#define STATIC_ASSERT(X) (CompileError<(X)>())template<typename S, typename T>union Pack{	inline Pack(void);	inline Pack(Pack const &);	inline Pack(unsigned long long const &);	inline Pack(S const &, T const &);	unsigned long long pack;	struct // TODO: Check if the below order is only correct for little-endian architectures	{		S lower;		T upper;	};};template<typename S, typename T>inline Pack<S, T>::Pack(void){	STATIC_ASSERT(sizeof(S) == 4 && sizeof(T) == 4);}template<typename S, typename T>inline Pack<S, T>::Pack(Pack const &other){	STATIC_ASSERT(sizeof(S) == 4 && sizeof(T) == 4);	pack = other.pack;}template<typename S, typename T>inline Pack<S, T>::Pack(unsigned long long const &op){	STATIC_ASSERT(sizeof(S) == 4 && sizeof(T) == 4);	pack = op;}template<typename S, typename T>inline Pack<S, T>::Pack(S const &nl, T const ν){	STATIC_ASSERT(sizeof(S) == 4 && sizeof(T) == 4);	lower = nl;	upper = nu;}#endif

// Copyright (C) 2008 Borislav Trifonov// Based on algorithm from "A Practical Nonblocking Queue Algorithm Using Compare-And-Swap"// TODO: For 64-bit system, if cannot adapt to _InterlockedCompare64Exchange128() or _InterlockedCompareExchange128(), use instead algorithm from Fig.5 in "Non-Blocking Concurrent FIFO Queues with Single Word Synchronization Primitives"#if !defined ATOMIC_QUEUE#define ATOMIC_QUEUE#include <exception>#include "processor.h"template<typename T>class AtomicQueue{public:	class Exc : public std::exception	{	public:		inline Exc(const char []);	};	inline AtomicQueue(unsigned long);	inline ~AtomicQueue(void);	inline bool Push(T const); // Returns false if full	inline bool Pop(T &); // Returns false if emtpy	inline T operator[](unsigned long); // No error checkingprivate:	inline AtomicQueue(void);	AtomicQueue(AtomicQueue const &);	AtomicQueue &operator=(AtomicQueue const &);	__declspec(align(4)) unsigned long head, tail;	unsigned long size;	typedef Pack<T, unsigned long> AQPack;	AQPack *buff;};template<typename T>inline AtomicQueue<T>::AtomicQueue(unsigned long sz) : head(0), tail(0), size(sz), buff(0){	STATIC_ASSERT(sizeof(T) == 4);	STATIC_ASSERT(sizeof(unsigned long) == 4);	STATIC_ASSERT(sizeof(unsigned long long) == 8);	buff = static_cast<AQPack *>(_mm_malloc(size * sizeof(AQPack), 8));	if (!buff) throw Exc("Not enough memory");	for (unsigned long i = 0; i < size; ++i) buff.pack = 0ull;}#include <iostream>template<typename T>inline AtomicQueue<T>::~AtomicQueue(void){	_mm_free(buff);}template<typename T>inline bool AtomicQueue<T>::Push(T const item){	while (true)	{		__declspec(align(4)) unsigned long const t(Acquire(tail)), h(Acquire(head));		__declspec(align(8)) AQPack a(Acquire(buff[t % size].pack));		if (t != Acquire(tail)) continue;		if (t == Acquire(head) + size)		{			if (Acquire(buff[h % size].lower) && h == Acquire(head)) return false;			_InterlockedCompareExchange(&head, h + 1, h);			continue;		}		if (a.lower)		{			if (Acquire(buff[t % size].lower)) _InterlockedCompareExchange(&tail, t + 1, t);		}		else		{			__declspec(align(8)) AQPack b(item, a.upper + 1);			if (_InterlockedCompareExchange64(&(buff[t % size].pack), b.pack, a.pack) == static_cast<long long>(a.pack))			{				_InterlockedCompareExchange(&tail, t + 1, t);				return true;			}		}	}}template<typename T>inline bool AtomicQueue<T>::Pop(T &item){	while (true)	{		__declspec(align(4)) unsigned long const h(Acquire(head)), t(Acquire(tail));		__declspec(align(8)) AQPack a(Acquire(buff[h % size].pack));		if (h != Acquire(head)) continue;		if (h == Acquire(tail))		{			if (!Acquire(buff[t % size].lower) && t == Acquire(tail)) return false;			_InterlockedCompareExchange(&tail, t + 1, t);			continue;		}		if (!a.lower)		{			if (!Acquire(buff[h % size].lower)) _InterlockedCompareExchange(&head, h + 1, h);		}		else		{			__declspec(align(8)) AQPack b(0, a.upper + 1);			if (_InterlockedCompareExchange64(&(buff[h % size].pack), b.pack, a.pack) == static_cast<long long>(a.pack))			{				_InterlockedCompareExchange(&head, h + 1, h);				item = a.lower;				return true;			}		}	}}template<typename T>inline T AtomicQueue<T>::operator[](unsigned long i){	return buff[(head + i) % size].lower;}template<typename T>inline AtomicQueue<T>::Exc::Exc(const char msg[]) : std::exception(msg){}#endif

template<typename T>inline bool AtomicQueue<T>::Push(T const item){	while (true)	{		__declspec(align(4)) unsigned long const t(Acquire(tail)), h(Acquire(head));		if (t == h + size)		{			if (Acquire(buff[h % size].lower) && h == Acquire(head)) return false;			_InterlockedCompareExchange(&head, h + 1, h);			continue;		}		__declspec(align(8)) AQPack a(Acquire(buff[t % size].pack));		if (t != Acquire(tail)) continue;		if (a.lower) _InterlockedCompareExchange(&tail, t + 1, t);		else		{			__declspec(align(8)) AQPack b(item, a.upper + 1);			if (_InterlockedCompareExchange64(&(buff[t % size].pack), b.pack, a.pack) == static_cast<long long>(a.pack))			{				_InterlockedCompareExchange(&tail, t + 1, t);				return true;			}		}	}}template<typename T>inline bool AtomicQueue<T>::Pop(T &item){	while (true)	{		__declspec(align(4)) unsigned long const h(Acquire(head)), t(Acquire(tail));		if (h == t)		{			if (!Acquire(buff[t % size].lower) && t == Acquire(tail)) return false;			_InterlockedCompareExchange(&tail, t + 1, t);			continue;		}		__declspec(align(8)) AQPack a(Acquire(buff[h % size].pack));		if (h != Acquire(head)) continue;		if (!a.lower) _InterlockedCompareExchange(&head, h + 1, h);		else		{			__declspec(align(8)) AQPack b(0, a.upper + 1);			if (_InterlockedCompareExchange64(&(buff[h % size].pack), b.pack, a.pack) == static_cast<long long>(a.pack))			{				_InterlockedCompareExchange(&head, h + 1, h);				item = a.lower;				return true;			}		}	}}

//copyright 2008 Hodgman - Free for personal or educational use only ;)	enum IncrementType	{		Pre,		Post	};	/** @brief Integer type that maintains a counter of modifications made.	 * The Counted Index is designed to avoid the "ABA problem" in lock-free code.	 * It can also act as a fast user-space mutex.	 * @todo finish documenting this class	 */	class CCountedIndex	{	public:		CCountedIndex(uint32 v=0) : i(v) {}		CCountedIndex(const CCountedIndex& o) : i(o.i) {}		/** Tries to lock the mutex bit		 * @return true if the mutex was locked, else false */		bool Lock()		{			uint v = i;			if( v & LockMask )				return false;			return i.SetIfEqual( v | LockMask, v ); //set lock		}		/** Unlocks the mutex bit. The mutex is assumed to be locked prior to calling. */		void Unlock()		{			uint v = i;			ASSERT( v & LockMask );			i = v & ~LockMask;                      //remove lock		}		/** Assigns a new value to the index bits. No thread-safety checks! */		void Set( uint32 index )		{			uint v = i;			ASSERT( v & LockMask );			uint n = ((v+(1<<IndexBits))&CountMask) | //increment counter			         ((index)&IndexMask)            | //set index			         LockMask;                        //keep lock			i = n;		}		/** Increment the index bits. May fail due to concurrent modifications.		 * @param limit One past the maximum index value. Index will wrap to 0 if limit is reached.		 * @return true if the index was incremented, else false */		bool Increment( uint32 limit )		{			uint v = i;			if( v & LockMask )				return false;			uint n = ((v+(1<<IndexBits))&CountMask) | //increment counter			         Inc(v&IndexMask,limit);          //increment index			return i.SetIfEqual( n, v );		}		/** . */		template<IncrementType T, bool L>		uint32 Increment( uint32 limit, uint32 full, bool& fail )		{			uint32 v = i;			ASSERT( !L || (v & LockMask) )			if( !L && (v & LockMask) )			{				return std::numeric_limits<uint32>::max();			}			uint oldIndex = v&IndexMask			uint newIndex = Inc(oldIndex, limit);          //increment index			fail = (newIndex == full);			if( fail )				return std::numeric_limits<uint32>::max();			int n = newIndex | ((v+(1<<IndexBits))&CountMask) |//increment counter				 (L?LockMask:0);			return i.SetIfEqual( n, v )				? (T==Pre?oldIndex:newIndex)				: std::numeric_limits<uint32>::max();		}		/** . */		template<IncrementType T, bool L>		uint32 Increment( uint32 limit, const CCountedIndex& original )		{			uint32 v = i;			ASSERT( !L || (v & LockMask) )			if( v != original.i || (!L&&(v & LockMask)) )				return std::numeric_limits<uint32>::max();			uint oldIndex = v&IndexMask			uint newIndex = Inc(oldIndex, limit);   //increment index			int n = newIndex |			        ((v+(1<<IndexBits))&CountMask)| //increment counter			        (L?LockMask:0);			return i.SetIfEqual( n, v )				? (T==Pre?oldIndex:newIndex)				: std::numeric_limits<uint32>::max();		}		/** . */		void IncrementAndUnlock( uint32 limit )		{			uint v = i;			ASSERT( v & LockMask );			uint n = ((v+(1<<IndexBits))&CountMask) | //increment counter			         Inc(v&IndexMask,limit);          //increment index			i = n;		}		/** . */		bool IncrementAndLock( uint32 limit )		{			uint v = i;			if( v & LockMask )				return false;			int n = ((v+(1<<IndexBits))&CountMask) | //increment counter			        Inc(v&IndexMask,limit)         | //increment index			        LockMask;                        //set lock			return i.SetIfEqual( n, v );		}		/** . */		operator uint32() { return i&IndexMask }		/** . */		bool operator==( uint32 j ) { return (uint32(i)&IndexMask) == j; }	private:		CCountedIndex& operator=(const CCountedIndex& o);		inline int Inc(uint32 i, uint32 size){ ASSERT(i<IndexMask);return (i+1==size)?0:i+1; }		TAtomic<uint32> i;		const static uint IndexMask = 0x0000FFFF;		const static uint CountMask = 0x7FFF0000;		const static uint LockMask  = 0x80000000;		const static uint IndexBits = 16;		const static uint CountBits = 15;		const static uint LockBits  = 1;	};

struct Node{	T data;	CAtomic valid;	Node():valid(0){}	Node(const Node&o):data(o.data),valid(o.valid){}};TAtomic<Node*> array;CCountedIndex  head, tail;

uint32 limit = 7;//size of arrayuint32 full = head;bool isFull;uint32 write = tail.Increment<Pre,false>( limit, full, isFull );if( write != std::numeric_limits<uint32>::max() )//increment succeeded{	array[write].data = v;//do work	array[write].valid = 1;//then publish	break;//this snippet is inside a while(true) loop}else if( isFull )//increment failed because array is full{	//try to lock head+tail and resize array}