Testing distributed memory pools

#include "stdafx.h" 
#include "perf_counter.h"
#include "processor_pool.h"
#include "thread_pool.h"


#ifdef TEST_CONTENTION_COUNT
volatile long g_contention_count;
#endif

#if defined(TEST_PROCESSOR_POOL) || defined(TEST_SINGLE_POOL)
#pragma message("===== Using processor_pool namespace =====")
using namespace processor_pool;
#else
#pragma message("===== Using thread_pool namespace =====")
using namespace thread_pool;
#endif



HANDLE g_start_event;
volatile long g_stop_test; 

const int writer_test_iterations = 1000000;
const int number_of_writers = 4;
const int reader_test_iterations = 4000000;
const int number_of_readers = 4;

const int number_of_pools = 4;


#pragma warning(disable:4324)
class test_buffer
{
public:
	bool read(void)
	{
		return (m_checksum == checksum(m_buffer, sizeof(m_buffer)));
	}
	bool write(void)
	{
		if (m_checksum != checksum(m_buffer, sizeof(m_buffer)))
		{
			return false;
		}
		else
		{
			set_buffer();
			return true;
		}
	}
	test_buffer()
	{
		set_buffer();
	}
private:
	USHORT checksum( LPVOID data, int len )
	//
	// The function checksum came from http://ava.org.ua/?2&17&get=87&read=87
	//
	{
		ULONG sum = 0;
		USHORT * dp =  (USHORT *) data;
		USHORT sum_s;
		int words = len >> 1;
		while( words -- )  sum += * dp ++;
		if( len & 1 ) sum += *(UCHAR*) dp;
		sum   = (USHORT) sum + (sum >> 16) & 0xffff;
		sum_s = (USHORT) sum + (USHORT)(sum >> 16);
		return sum_s != 0xffff ? ~sum_s : sum_s;
	}
	void set_buffer(void)
	{
		for(int i = 0; i < sizeof(m_buffer)/sizeof(m_buffer[0]); i ++)
		{
			m_buffer[i] = rand();
		}
		m_checksum = checksum(m_buffer, sizeof(m_buffer));
	}
	enum {BUF_SIZE = 64};
	int m_buffer[BUF_SIZE];
	USHORT m_checksum;
};
#pragma warning(default:4324)


#if defined(TEST_SINGLE_POOL)
#pragma message(" Testing a singe pool Pool_t")
Pool_t<test_buffer, number_of_readers + number_of_writers> g_pool; 
#else
#pragma message(" Testing an array of pools Array_of_pools_t")
Array_of_pools_t<test_buffer, number_of_readers + number_of_writers, number_of_pools> g_pool; 
#endif


template <const int test_iteration>
class Test_worker{
public:
	HANDLE start(void)
	{
		::ResumeThread(m_handle);
		return m_handle;
	}
	double test_duration(void)
	{
		return m_perf_counter.get_duration();
	}
	LONGLONG test_iteration_done(void)
	{
		return m_perf_counter.get_counter();
	}
private:
	static DWORD WINAPI worker_proc(LPVOID lpParam)
	{
		Test_worker& worker = *reinterpret_cast<Test_worker*>(lpParam);

#if !defined(TEST_PROCESSOR_POOL) && !defined(TEST_SINGLE_POOL)
		worker.m_pool_index = g_pool.get_pool_index(); 
#endif

		::WaitForSingleObject(g_start_event, INFINITE );

		worker.m_perf_counter.start();
		int i;

		for (i = 0; i < test_iteration; i++)
		{
			if( false == worker.simulate_work() )
			{
				::RaiseException( STATUS_NONCONTINUABLE_EXCEPTION, 0, 0, 0);
 			}
			else
			{
			}

			if (TRUE == g_stop_test)
			{
				break;
			}
			else
			{
			}

		}
		
		worker.m_perf_counter.end();
		::InterlockedExchange(&g_stop_test, TRUE);
		worker.m_perf_counter.set_iteration_done(i);
		return 0;
	}

	virtual bool simulate_work(void)=0;

	HANDLE m_handle;
	Performance_counter_meter m_perf_counter;
	Test_worker & operator=( const Test_worker & ) {}
protected:
	Test_worker()
	{
		m_handle = ::CreateThread(NULL,0, worker_proc, (LPVOID)this, CREATE_SUSPENDED, NULL);
	}

	size_t m_pool_index;
};


class Writer: public Test_worker <writer_test_iterations>
{
#if defined(TEST_PROCESSOR_POOL) || defined(TEST_SINGLE_POOL)
		virtual bool simulate_work(void)
		{
			test_buffer* buffer = g_pool.pop();
			if (buffer && buffer->write())
			{
				g_pool.push(buffer);
				return true;
			}
			else
			{
				return false;
			}
		}
#else
		virtual bool simulate_work(void)
		{
			test_buffer* buffer = g_pool.pop(m_pool_index);
			if (buffer && buffer->write())
			{
				g_pool.push(buffer, m_pool_index);
				return true;
			}
			else
			{
				return false;
			}
		}
#endif
};

class Reader: public Test_worker <reader_test_iterations>
{
#if defined(TEST_PROCESSOR_POOL) || defined(TEST_SINGLE_POOL)
		virtual bool simulate_work(void)
		{
			test_buffer* buffer = g_pool.pop();
			if (buffer && buffer->read())
			{
				g_pool.push(buffer);
				return true;
			}
			else
			{
				return false;
			}
		}
#else
		virtual bool simulate_work(void)
		{
			test_buffer* buffer = g_pool.pop(m_pool_index);
			if (buffer && buffer->read())
			{
				g_pool.push(buffer, m_pool_index);
				return true;
			}
			else
			{
				return false;
			}
		}
#endif
};


int _tmain(int , _TCHAR*)
{
	g_start_event = ::CreateEvent(NULL,TRUE,FALSE,NULL);

	HANDLE threads[number_of_writers + number_of_readers];

	Writer* writer[number_of_writers];
	for (int i = 0; i < number_of_writers; i++)
	{
		writer[i] = new Writer();
		threads[i] = writer[i]->start();
	}

	Reader* reader[number_of_readers];
	for (int i = 0; i < number_of_readers; i++)
	{
		reader[i] = new Reader();
		threads[i + number_of_writers] = reader[i]->start();
	}

	::SetEvent(g_start_event);
	::WaitForMultipleObjects(sizeof(threads)/sizeof(threads[0]), threads, TRUE, INFINITE);

	for (int i = 0; i < number_of_readers; i++)
	{
		printf("reader iterations done | per operation(ms),%lld,%f\n",
			reader[i]->test_iteration_done(),
			1000*reader[i]->test_duration()/reader[i]->test_iteration_done());
	}

	for (int i = 0; i < number_of_writers; i++)
	{
		printf("writer iterations done | per operation(ms),%lld,%f\n",
			writer[i]->test_iteration_done(),
			1000*writer[i]->test_duration()/writer[i]->test_iteration_done());
	}

	return 0;
}