// UnitTest_Concurrency_Test.cpp : Defines the entry point for the console application.
//

#include "stdafx.h"
#include <boost/coroutine/all.hpp>
#include <boost/lockfree/spsc_queue.hpp>
#include <boost/lockfree/queue.hpp>
#include <boost/circular_buffer.hpp>
#include <boost/bind.hpp>
#include <boost/chrono.hpp>
#include <cstdio>
#include <iostream>
#include <thread>
#include <future>
#include <queue>
#include <atomic>
#include <random>
#include <numeric>      // std::accumulate

typedef boost::coroutines::coroutine< void > coro_t;
static std::atomic_size_t iterCounters[3];

struct JobTask1
{
	JobTask1( const uint32_t kIters, bool foo ) 
		: kIters(kIters)
	{}

	//virtual ~JobTask1(){}
	const uint32_t kIters;
#ifdef _DEBUG
		size_t iterCounter_;
#endif

	void operator() ( coro_t::push_type& yield )
	{
#ifdef _DEBUG
		iterCounter_ = 0;
#endif

		for( uint32_t i = 0; i < kIters; ++i)
		{
			++iterCounters[0]; //< Atomic increment, woudl think it would be okay to not be!
#ifdef _DEBUG
			++iterCounter_;
#endif
		// std::cout << "fn(): local variable i == " << i << " of " << kIters << std::endl;

			// save current coroutine
			// value of local variable is preserved
			// transfer execution control back to main()
			yield();

			// coroutine<>::operator()() was called
			// execution control transferred back from main()
		}
		
#ifdef _DEBUG
		assert( iterCounter_ == kIters );
#endif
	}
};


struct JobTask2
{
	JobTask2( const uint32_t kIters ) 
		: kIters(kIters)
	{}	
	const uint32_t kIters;

	void operator() ( coro_t::push_type& yield )
	{
		for( uint32_t i = 0; i < kIters; ++i)
		{
			++iterCounters[1]; //< Atomic increment, woudl think it would be okay to not be!
			yield();
		}
	}
};

struct JobTask3
{
	JobTask3( const uint32_t kIters ) 
		: kIters(kIters)
	{}	
	const uint32_t kIters;

	void operator() ( coro_t::push_type& yield )
	{
		for( uint32_t i = 0; i < kIters; ++i)
		{
			++iterCounters[2]; //< Atomic increment, woudl think it would be okay to not be!
			yield();
		}
	}
};


class Job 
{
public:
	virtual ~Job(){}

	//Job( Job&& rhs ) : coroutine_( rhs ) {}

	typedef void (*Fn)( Job* inst, coro_t::push_type& yield );

	//template< class Task >
	Job( Fn fn ) 
		: fn_(fn)
	{
		if ( !fn )
			std::terminate();
	}


	Fn fn() const
	{ return fn_; }

private:
	Fn fn_;
};

template< class Task >
class Job_t : public Job, public Task
{
public:	
	

	Job_t( Task&& task ) 
		: Job( &sproc )
		, Task(task) 
	{
		int i = 0;
	}

	static void sproc( Job* inst, coro_t::push_type& yield )
	{
		Task task( *static_cast<Job_t*>(inst) );
		task( yield );
	}

};


class JobRun : public coro_t
{
public:
	JobRun( JobRun&& rhs ) 
		: coroutine_( std::move(rhs.coroutine_) ) 
	{}

	JobRun( Job* job )
		: coroutine_(  boost::bind( job->fn(), job, _1 ) )
	{
	}
	
	void run()
	{ coroutine_(); }

	operator bool() const
	{ return coroutine_; }

	bool isFinished() const
	{ return !coroutine_; }

private:
	pull_type coroutine_;
};


static const size_t kThreadBufferSize = 1024;//*4; //< more than enough jobs queued up, scheduler may have more than this still but these are pending work queues!

#define SPSC_QUEUE 1 //< spsc_queue technically the one to chose as it is wait free (i.e. no chance fo deadlock as it uses no locks!)
#if SPSC_QUEUE
typedef boost::lockfree::spsc_queue<Job*, boost::lockfree::capacity<kThreadBufferSize> > JobQueue; //< Single-producer-single-consumer (94-119ns) mult-thread (30ns)
#else
typedef boost::lockfree::queue<Job*, boost::lockfree::capacity<kThreadBufferSize> > JobQueue; //< many to many queue (94-121ns) mult-thread (32ns)
#endif

#define CIRCULAR_BUFFER 0 //< Circular-buffer has preallocate and may technically be faster as new/malloc can block, however cannot grow but queue is limited by JobQueue being fixed size anyway!
#if CIRCULAR_BUFFER
typedef std::queue<JobRun,  boost::circular_buffer<JobRun> > JobRunQueue;  //,circular_buffer (94-113ns) mult-thread (30ns)
#else
typedef std::queue<JobRun > JobRunQueue; //<deque (117-128ns) mult-thread (34ns)
#endif

class Worker
{
public:
	Worker() 
		: done(false)
#if CIRCULAR_BUFFER
		, running( boost::circular_buffer<JobRun>(kThreadBufferSize) )
#endif
	{}

	volatile bool done;

	void addJob( Job* job )
	{		
		//TODO: deal with full queue somehow?
		pending.push( job ); //pending is a lock-free queue so is thread safe
	}

	void process()
	{
		//Move all the jobs into our local running-queue
		// - This reduces the round-cost for coroutine executions as the local-queue does not need to be thread safe and therefore more lightweight
		pending.consume_all( [&]( Job* job )
		{ 
			//assert( running.size() < kThreadBufferSize ); //< TODO: we could avoid moving the job to the runnign queue if the queu is full!
			running.push( JobRun(job) );
		} );

		while ( !running.empty() )
		{
			JobRun run( std::move(running.front()) );
			running.pop();
			run.run();

			if ( !run.isFinished() )
			{
				assert( running.size() < kThreadBufferSize );
				running.push( std::move(run) );
			}
		}
	}

	void operator()()
	{		
		while ( !done )
			process();	

		process();

		assert( pending.empty() );
		assert( running.empty() );
	}

protected:
	JobQueue pending;
	JobRunQueue running; //, jobs that have further work to do
};


void test()
{
	for ( auto& iterCount: iterCounters )
		iterCount = 0;
	
	uint32_t kJobTotal = 1024;
	
	//std::random_device rd;
	const uint32_t kSeed = 3;
	std::default_random_engine randomEngine( kSeed/*rd()*/ );
	std::uniform_int_distribution<uint32_t> jobWorkDistribution(100, 200 ); //, number of iterations per job


	size_t kItersTotal = 0;
	std::vector<Job*> jobs;
	uint32_t iJob = 0;
	std::generate_n(std::back_inserter(jobs), kJobTotal, [&]() 
	{ 
		uint32_t iters = jobWorkDistribution(randomEngine);
		kItersTotal += iters;
		
		Job* newJob = 0;
		switch ( (iJob++) % 3 )
		{
		case 0: newJob = new Job_t<JobTask1>( JobTask1(iters, false) ); break;
		case 1: newJob = new Job_t<JobTask2>( JobTask2(iters) ); break;
		case 2: newJob = new Job_t<JobTask3>( JobTask3(iters) ); break;
		default: assert(false); break;
		}
		assert(newJob);
		return newJob;
	});

	assert( jobs.size() <= kThreadBufferSize );

	//std::cout << "main() starts coroutine c" << std::endl;

	//Target: coroutine time = 40ns
	//Local-running queue: 49-51ns

	typedef boost::chrono::high_resolution_clock Clock;
	//typedef high_resolution_clock Clock;

#define THREAD 1
#define ASYNC 0

	const uint32_t kOverOccupancy = 2;
	const uint32_t kWorkerCount = std::thread::hardware_concurrency() * kOverOccupancy;

	std::vector< std::unique_ptr<Worker> > workers; //<TODO: Worker not copyable so must use pointers!? TODO: solve this albeit jsut a niggle!
	std::generate_n(std::back_inserter(workers), kWorkerCount, [&]() 
	{ 
		return std::unique_ptr<Worker>(new Worker()); 
	});

#if THREAD
	std::vector<std::thread> threads;
	uint32_t iThread = 0;
	std::generate_n(std::back_inserter(threads), kWorkerCount, [&]()
	{ 
		return std::thread( std::ref(*workers[iThread++]) ); 
	});
#endif

	auto t1 = Clock::now();
	
	for ( uint32_t iJob = 0; iJob != jobs.size(); ++iJob )	
		workers[iJob%kWorkerCount]->addJob( jobs[iJob] );

	for ( auto& worker: workers )
		worker->done = true; //< Signal we want the worker to complete all its work so the thread may end

#if THREAD	
	for ( auto& thread: threads )
		thread.join();
	threads.clear();

#elif ASYNC
	 std::async(std::launch::async, std::ref(worker) ).wait();
#else
	for ( auto& worker: workers )
		(*worker)();
#endif

	auto t2 = Clock::now();

	size_t iterCounter = std::accumulate( std::begin(iterCounters), std::end(iterCounters), 0);
	std::cout << "Iters =  " << iterCounter << " of " << kItersTotal << " [" << (iterCounter==kItersTotal ? "TRUE" : "FALSE") << "]\n";

	assert( iterCounter == kItersTotal );

	auto nanos = (t2-t1).count();
	std::cout << "Time: " <<  nanos / 1000000.f << "ms (iteration=" << nanos/kItersTotal << "ns)\n";

	std::cout << "Done" << std::endl;
}

int main( int argc, char * argv[])
{
#if 1 //< Issue seems to only occur on first run of tests, related to startup of application?
	int i = 3;
	while ( i-- )
#endif
		test();
	
	return EXIT_SUCCESS;
}