MPSC Wait-free Queue for C++, Windows

WaitFreeQueue.cpp

#pragma once

/****************************************************************************
* Wait-Free MPSC Queue
* Minimum requirement: Windows XP or Windows Server 2003
* Author: @sm9kr
* License type: GPL v3
* References
** [1] http://groups.google.ru/group/comp.programming.threads/browse_frm/thread/33f79c75146582f3
** [2] http://www.boost.org/doc/libs/1_35_0/doc/html/intrusive/intrusive_vs_nontrusive.html
****************************************************************************/

// 멀티스레드에서 테스트할 때는 콘솔 출력을 막던가 스레드세이프하게 변경해야 한다
#include <iostream>
#include <Windows.h>

/// 노드 구조
class NodeEntry
{
public:
	NodeEntry() : mNext(nullptr) {}
	NodeEntry* volatile mNext;
};

/// 아래와 같은 형태로 intrusive방식으로 노드 구성해야 함 [2]
class DataNode
{
public:
	DataNode() {}

	__int64		mData;
	NodeEntry	mNodeEntry; ///< 반드시 NodeEntry를 포함해야 함
};


/**
* [1]을 참고하여 C++ Windows 환경에 맞게 변경한 MPSC 큐.
* 여러 쓰레드에서 Push는 되지만 Pop은 하나의 지정된 쓰레드에서만 해야 함
* 사용 예
WaitFreeQueue<DataNode> testQueue ;
DataNode* pushData = new DataNode ;
testQueue.Push(newData) ;
DataNode* popData = testQueue.Pop() ;
delete popData ;
* 물론, DataNode*가 큐 안에 있을때 다른 쓰레드에서 날려버지 않도록 스마트 포인터로 만들어서 쓰는 것이 좋다.
*/
template <class T>
class WaitFreeQueue
{
public:
	WaitFreeQueue() : mHead(&mStub), mTail(&mStub)
	{
		mOffset = reinterpret_cast<__int64>(&((reinterpret_cast<T*>(0))->mNodeEntry));
	}
	~WaitFreeQueue() {}

	void Push(T* newData)
	{
		std::cout << "Push" << std::endl;
		NodeEntry* prevNode = (NodeEntry*)InterlockedExchangePointer((PVOID volatile*)&mHead, (PVOID*)&(newData->mNodeEntry));
		prevNode->mNext = &(newData->mNodeEntry);
	}

	T* Pop()
	{
		NodeEntry* tail = mTail;
		NodeEntry* next = tail->mNext;

		if (tail == &mStub)
		{
			/// 데이터가 없을 때
			if (nullptr == next) {
				return nullptr;
			}

			std::cout << "처음 꺼낼 때 - ";

			mTail = next;
			tail = next;
			next = next->mNext;
		}

		/// 대부분의 경우에 데이터를 빼낼 때
		if (next)
		{
			std::cout << "데이터를 빼낸다 case-1" << std::endl;

			mTail = next;

			return reinterpret_cast<T*>(reinterpret_cast<__int64>(tail) - mOffset);
		}

		NodeEntry* head = mHead;
		if (tail != head) {
			return nullptr;
		}

		std::cout << "마지막 데이터 꺼낼 때 case-2" << std::endl;
		mStub.mNext = nullptr;
		NodeEntry* prev = (NodeEntry*)InterlockedExchangePointer((PVOID volatile*)&mHead, (PVOID*)&mStub);
		prev->mNext = &mStub;

		next = tail->mNext;
		if (next)
		{
			mTail = next;

			return reinterpret_cast<T*>(reinterpret_cast<__int64>(tail) - mOffset);
		}

		return nullptr;
	}


private:

	NodeEntry* volatile	mHead;
	NodeEntry*			mTail;
	NodeEntry			mStub;

	__int64				mOffset;

};



int main()
{
	std::cout << "1개 넣고 뺀다" << std::endl;
	{
		WaitFreeQueue<DataNode> testQueue;
		DataNode* newData1 = new DataNode;
		testQueue.Push(newData1);
		DataNode* popData1 = testQueue.Pop();
		delete popData1;
	}
	std::cout << std::endl;

	std::cout << "3개 넣고 뺀다" << std::endl;
	{
		WaitFreeQueue<DataNode> testQueue;
		DataNode* newData1 = new DataNode;
		testQueue.Push(newData1);
		DataNode* newData2 = new DataNode;
		testQueue.Push(newData2);
		DataNode* newData3 = new DataNode;
		testQueue.Push(newData3);

		DataNode* popData1 = testQueue.Pop();
		delete popData1;
		DataNode* popData2 = testQueue.Pop();
		delete popData2;
		DataNode* popData3 = testQueue.Pop();
		delete popData3;
	}
	std::cout << std::endl;

	return 0;
}