first commit

include/platform/container/mpmc_bounded_queue.h

@@ -0,0 +1,136 @@
+/*
+	Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
+
+	Redistribution and use in source and binary forms, with or without
+	modification, are permitted provided that the following conditions are met:
+
+	   1. Redistributions of source code must retain the above copyright notice,
+		  this list of conditions and the following disclaimer.
+
+	   2. Redistributions in binary form must reproduce the above copyright
+		  notice, this list of conditions and the following disclaimer in the
+		  documentation and/or other materials provided with the distribution.
+
+	THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR
+	IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+	MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
+	EVENT SHALL DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+	INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+	(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+	 LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+	 ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 
+	(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
+	 THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+	The views and conclusions contained in the software and documentation are
+	those of the authors and should not be interpreted as representing official
+	policies, either expressed or implied, of Dmitry Vyukov.
+*/
+
+
+#ifndef	__MPMC_BOUNDED_QUEUE__
+#define	__MPMC_BOUNDED_QUEUE__
+
+
+		#include "thread/atomic_value.h"
+
+
+// Adapted from https://sites.google.com/site/1024cores/home/lock-free-algorithms/queues/bounded-mpmc-queue
+template<typename T> class mpmc_bounded_queue
+{
+	struct cell_t
+	{
+		GS::Threading::Atomic32 sequence_;
+		T			data_;
+	};
+
+		static size_t const	cacheline_size = 64;
+		typedef char cacheline_pad_t[cacheline_size];
+
+		cacheline_pad_t	pad0_;
+		cell_t * const	buffer_;
+		size_t const	buffer_mask_;
+
+		cacheline_pad_t	pad1_;
+		GS::Threading::Atomic32 enqueue_pos_;
+		cacheline_pad_t	pad2_;
+		GS::Threading::Atomic32 dequeue_pos_;
+		cacheline_pad_t	pad3_;
+
+		void			operator = (mpmc_bounded_queue const&);
+
+		mpmc_bounded_queue(mpmc_bounded_queue const&);
+
+public:
+
+	bool	enqueue(T const &data)
+	{
+		cell_t *cell;
+		int pos = enqueue_pos_.Get();
+
+		for (;;)
+		{
+			cell = &buffer_[pos & buffer_mask_];
+			size_t seq = cell->sequence_.Get();
+
+			intptr_t dif = (intptr_t)seq - (intptr_t)pos;
+
+			if (dif == 0)
+			{
+				if (enqueue_pos_.Cas(pos, pos + 1) == pos)
+					break;
+			}
+			else if (dif < 0)
+				return false;
+			else
+				pos = enqueue_pos_.Get();
+		}
+
+		cell->data_ = data;
+		cell->sequence_.Set(pos + 1);
+		return true;
+	}
+	bool	dequeue(T &data)
+	{
+		cell_t *cell;
+		int pos = dequeue_pos_.Get();
+
+		for (;;)
+		{
+			cell = &buffer_[pos & buffer_mask_];
+			size_t seq = cell->sequence_.Get();
+
+			intptr_t dif = (intptr_t)seq - (intptr_t)(pos + 1);
+
+			if	(dif == 0)
+			{
+				if	(dequeue_pos_.Cas(pos, pos + 1) == pos)
+					break;
+			}
+			else if	(dif < 0)
+				return false;
+			else
+				pos = dequeue_pos_.Get();
+		}
+		data = cell->data_;
+		cell->sequence_.Set(pos + buffer_mask_ + 1);
+
+		return true;
+	}
+
+	mpmc_bounded_queue(size_t buffer_size) : buffer_(new cell_t [buffer_size]), buffer_mask_(buffer_size - 1)
+	{
+		for (size_t i = 0; i != buffer_size; i += 1)
+			buffer_[i].sequence_.Set(i);
+
+		enqueue_pos_.Set(0);
+		dequeue_pos_.Set(0);
+	}
+	~mpmc_bounded_queue()
+	{
+		delete [] buffer_;
+	}
+}; 
+
+
+#endif	// __MPMC_BOUNDED_QUEUE__