AltDevBlog - Multi-Threaded Queue: Tests

MultiThreadedQueueTests.cpp

/***************************************************************************
Copyright (c) 2014 Joseph Simons
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
******************************************************************************/

#include <tchar.h>
#include <iostream>
#include <thread>
#include <array>

#include "LockingQueue.h"
#include "LockFreeQueue.h"

static const int DATA_SIZE = 100000;
std::array<bool, DATA_SIZE> processed;
std::array<std::thread, DATA_SIZE> threads;

static const int MAX_WRITE_THREADS = 4;
static const int MAX_READ_THREADS = 4;
std::array<std::thread, MAX_WRITE_THREADS> writeThreads;
std::array<std::thread, MAX_READ_THREADS> readThreads;
static_assert(DATA_SIZE % MAX_WRITE_THREADS == 0, "DATA_SIZE needs to be a multiple of MAX_WRITE_THREADS");
static_assert(DATA_SIZE % MAX_READ_THREADS == 0, "DATA_SIZE needs to be a multiple of MAX_READ_THREADS");

std::array<std::array<int, DATA_SIZE / MAX_WRITE_THREADS>, MAX_WRITE_THREADS> threadData;

bool TestLockFreeQueue()
{
	UnboundedLockFreeQueue<int> multiQueue;

	for (auto& processValue : processed)
	{
		processValue = false;
	}

	for (int dataIndex = 0; dataIndex < DATA_SIZE; dataIndex++)
	{
		multiQueue.Enqueue(std::make_unique<int>(dataIndex));
	}

	for (int dataIndex = 0; dataIndex < DATA_SIZE; dataIndex++)
	{
		auto data = multiQueue.Dequeue();
		processed[*data] = true;
	}

	for (auto processValue : processed)
	{
		if (processValue == false)
		{
			std::wcout << L"LOCKFREE: ERROR PROCESSING SINGLETHREADED ENQUEUE/DEQUEUE!" << std::endl;
			return false;
		}
	}

	// let's test a bunch of threads each with a single number, separately hitting enqueue and dequeue
	for (auto& processValue : processed)
	{
		processValue = false;
	}

	for (int dataIndex = 0; dataIndex < DATA_SIZE; dataIndex++)
	{
		threads[dataIndex] = std::thread([=, &multiQueue](){
			multiQueue.Enqueue(std::make_unique<int>(dataIndex));
		});
	}

	for (auto& thread : threads)
	{
		thread.join();
		thread = std::thread([&](){
			auto data = multiQueue.Dequeue();
			processed[*data] = true;
		});
	}

	for (auto& thread : threads)
	{
		thread.join();
	}

	for (auto processValue : processed)
	{
		if (processValue == false)
		{
			std::wcout << L"LOCKFREE: ERROR PROCESSING SEPERATE MULTITHREADED ENQUEUE/DEQUEUE!" << std::endl;
			return false;
		}
	}

	// need to test just a handful of threads running with more numbers at the same time
	for (auto& processValue : processed)
	{
		processValue = false;
	}

	for (int dataIndex = 0; dataIndex < DATA_SIZE; dataIndex++)
	{
		threadData[dataIndex % MAX_WRITE_THREADS][dataIndex / MAX_WRITE_THREADS] = dataIndex;
	}

	for (int threadIndex = 0; threadIndex < MAX_READ_THREADS; threadIndex++)
	{
		readThreads[threadIndex] = std::thread([&](){
			int numRead = 0;
			while (numRead < DATA_SIZE / MAX_READ_THREADS)
			{
				auto data = multiQueue.Dequeue();
				if (data != nullptr)
				{
					processed[*data] = true;
					numRead++;
				}
			}
		});
	}

	for (int threadIndex = 0; threadIndex < MAX_WRITE_THREADS; threadIndex++)
	{
		writeThreads[threadIndex] = std::thread([&, threadIndex](){
			for (int data : threadData[threadIndex])
			{
				multiQueue.Enqueue(std::make_unique<int>(data));
			}
		});
	}

	for (auto& thread : writeThreads)
	{
		thread.join();
	}
	for (auto& thread : readThreads)
	{
		thread.join();
	}

	for (auto processValue : processed)
	{
		if (processValue == false)
		{
			std::wcout << L"LOCKFREE: ERROR PROCESSING FULL MULTITHREADED!" << std::endl;
			return false;
		}
	}

	return true;
}

int _tmain(int argc, _TCHAR* argv[])
{
	if (TestLockFreeQueue() == false)
	{
		return -1;
	}

	std::wcout << L"SUCCESSFUL PROCESSING!" << std::endl;

	return 0;
}