//  Copyright (c) 2008 Oliver Kowalke. Distributed under the Boost
//  Software License, Version 1.0. (See accompanying file
//  LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

#include <cstdlib>
#include <iostream>
#include <map>
#include <stdexcept>
#include <vector>

#include <boost/bind.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/function.hpp>
#include <boost/future/future.hpp>
#include <boost/ref.hpp>
#include <boost/test/unit_test.hpp>
#include <boost/thread.hpp>
#include <boost/thread/barrier.hpp>
#include <boost/utility.hpp>

#include <boost/tp/bounded_channel.hpp>
#include <boost/tp/exceptions.hpp>
#include <boost/tp/fifo.hpp>
#include <boost/tp/pool.hpp>
#include <boost/tp/poolsize.hpp>
#include <boost/tp/task.hpp>
#include <boost/tp/watermark.hpp>

#include "test_functions.hpp"

namespace pt = boost::posix_time;
namespace tp = boost::tp;

class fixed_bounded_channel_fifo
{
public:
	// check size, active, idle
	void test_case_1()
	{
		tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool(
			tp::poolsize( 3),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		BOOST_CHECK_EQUAL( pool.size(), std::size_t( 3) );
		BOOST_CHECK_EQUAL( pool.idle(), std::size_t( 3) );
		BOOST_CHECK_EQUAL( pool.active(), std::size_t( 0) );
	}

	// check submit
	void test_case_2()
	{
		tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool(
			tp::poolsize( 1),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		tp::task< int > t(
			pool.submit(
				boost::bind(
					fibonacci_fn,
					10) ) );
		BOOST_CHECK_EQUAL( t.get(), 55);
	}

	// check chained_submit
	void test_case_3()
	{
		tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool(
			tp::poolsize( 3),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		tp::task< int > t1(
			pool.submit(
				boost::bind(
					add_fn,
					1,
					2) ) );
		tp::task< int > t2(
			pool.chained_submit(
				boost::bind(
					add_fn,
					3,
					t1),
				t1) );
		tp::task< int > t3(
			pool.chained_submit(
				boost::bind(
					add_fn,
					4,
					t2),
				t2) );
		BOOST_CHECK_EQUAL( t3.get(), 10);
	}

	// check lazy_submit
	void test_case_4()
	{
		tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool(
			tp::poolsize( 3),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		std::vector<
			tp::task< int >
		>	f;
		f.reserve(20);
		for ( int i( 0); i < 11; ++i)
			f.push_back(
				pool.lazy_submit(
					boost::bind(
						fibonacci_fn,
						i) ) );
		BOOST_CHECK_EQUAL( f[0].get(), 0);
		BOOST_CHECK_EQUAL( f[2].get(), 1);
		BOOST_CHECK_EQUAL( f[4].get(), 3);
		BOOST_CHECK_EQUAL( f[6].get(), 8);
		BOOST_CHECK_EQUAL( f[8].get(), 21);
		BOOST_CHECK_EQUAL( f[10].get(), 55);
		BOOST_CHECK( ! f[1].ready() );
		BOOST_CHECK( ! f[3].ready() );
		BOOST_CHECK( ! f[5].ready() );
		BOOST_CHECK( ! f[7].ready() );
		BOOST_CHECK( ! f[9].ready() );
	}

	// check timed_submit
	void test_case_5()
	{
		tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool(
			tp::poolsize( 1),
			tp::high_watermark( 1),
			tp::low_watermark( 1) );
		boost::barrier b( 3);
		boost::function< int() > fn(
			boost::bind(
				fibonacci_fn,
				10) );
		pool.submit(
			boost::bind(
				( int ( *)( boost::function< int() > const&, boost::barrier &) ) barrier_fn,
				fn,
				boost::ref( b) ) );
		pool.submit(
			boost::bind(
				( int ( *)( boost::function< int() > const&, boost::barrier &) ) barrier_fn,
				fn,
				boost::ref( b) ) );
		bool thrown( false);
		try
		{
			pool.timed_submit(
				boost::bind(
					fibonacci_fn,
					10),
				pt::millisec( 1) );
		}
		catch ( tp::task_rejected const&)
		{ thrown = true; }
		BOOST_CHECK( thrown);
	}

	// check shutdown
	void test_case_6()
	{
		tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool(
			tp::poolsize( 1),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		tp::task< int > t(
			pool.submit(
				boost::bind(
					fibonacci_fn,
					10) ) );
		pool.shutdown();
		BOOST_CHECK( pool.terminated() );
		BOOST_CHECK_EQUAL( t.get(), 55);
	}

	// check runtime_error throw inside task
	void test_case_7()
	{
		tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool(
			tp::poolsize( 1),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		tp::task< void > t(
			pool.submit(
				boost::bind(
					throwing_fn) ) );
		pool.shutdown();
		bool thrown( false);
		try
		{ t.get(); }
		catch ( std::runtime_error const&)
		{ thrown = true; }
		BOOST_CHECK( thrown);
	}

	// check shutdown with task_rejected exception
	void test_case_8()
	{
		tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool(
			tp::poolsize( 1),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		pool.shutdown();
		BOOST_CHECK( pool.terminated() );
		bool thrown( false);
		try
		{
			pool.submit(
				boost::bind(
					fibonacci_fn,
					10) );
		}
		catch ( tp::task_rejected const&)
		{ thrown = true; }
		BOOST_CHECK( thrown);
	}

	// check shutdown_now with thread_interrupted exception
	void test_case_9()
	{
		tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool(
			tp::poolsize( 1),
			tp::high_watermark( 1),
			tp::low_watermark( 1) );
		boost::function< int() > fn(
			boost::bind(
				fibonacci_fn,
				10) );
		tp::task< int > t(
			pool.submit(
				boost::bind(
					( int ( *)( boost::function< int() > const&, pt::time_duration const&) ) delay_fn,
					fn,
					pt::millisec( 500) ) ) );
		boost::this_thread::sleep( pt::millisec( 250) );
		BOOST_CHECK_EQUAL( pool.size(), std::size_t( 1) );
		pool.shutdown_now();
		BOOST_CHECK( pool.terminated() );
		BOOST_CHECK_EQUAL( pool.size(), std::size_t( 1) );
		BOOST_CHECK_EQUAL( pool.idle(), std::size_t( 1) );
		BOOST_CHECK_EQUAL( pool.active(), std::size_t( 0) );
		bool thrown( false);
		try
		{ t.get(); }
		catch ( boost::thread_interrupted const&)
		{ thrown = true; }
		BOOST_CHECK( thrown);
	}

	// check pending
	void test_case_10()
	{
		typedef tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool_type;
		pool_type pool(
			tp::poolsize( 1),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		boost::barrier b( 2);
		boost::function< int() > fn(
			boost::bind(
				fibonacci_fn,
				10) );
		tp::task< int > t1(
			pool.submit(
				boost::bind(
					( int ( *)( boost::function< int() > const&, boost::barrier &) ) barrier_fn,
					fn,
					boost::ref( b) ) ) );
		boost::this_thread::sleep( pt::millisec( 250) );
		BOOST_CHECK_EQUAL( pool.pending(), std::size_t( 0) );
		tp::task< int > t2( pool.submit( fn) );
		boost::this_thread::sleep( pt::millisec(250) );
		BOOST_CHECK_EQUAL( pool.pending(), std::size_t( 1) );
		tp::task< int > t3( pool.submit( fn) );
		boost::this_thread::sleep( pt::millisec(250) );
		BOOST_CHECK_EQUAL( pool.pending(), std::size_t( 2) );
		b.wait();
		BOOST_CHECK_EQUAL( t1.get(), 55);
		BOOST_CHECK_EQUAL( t2.get(), 55);
		BOOST_CHECK_EQUAL( t3.get(), 55);
		BOOST_CHECK_EQUAL( pool.pending(), std::size_t( 0) );
	}

	// check fifo scheduling
	void test_case_11()
	{
		typedef tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool_type;
		pool_type pool(
			tp::poolsize( 1),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		boost::barrier b( 2);
		boost::function< int() > fn(
			boost::bind(
				fibonacci_fn,
				10) );
		pool.submit(
			boost::bind(
				( int ( *)( boost::function< int() > const&, boost::barrier &) ) barrier_fn,
				fn,
				boost::ref( b) ) );
		std::vector< int > buffer;
		pool.submit(
			boost::bind(
				buffer_fibonacci_fn,
				boost::ref( buffer),
				10) );
		pool.submit(
			boost::bind(
				buffer_fibonacci_fn,
				boost::ref( buffer),
				0) );
		b.wait();
		pool.shutdown();
		BOOST_CHECK_EQUAL( buffer[0], 55);
		BOOST_CHECK_EQUAL( buffer[1], 0);
		BOOST_CHECK_EQUAL( buffer.size(), std::size_t( 2) );
	}

	// check cancelation
	void test_case_12()
	{
		typedef tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool_type;
		pool_type pool(
			tp::poolsize( 1),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		boost::barrier b( 2);
		boost::function< int() > fn(
			boost::bind(
				fibonacci_fn,
				10) );
		pool.submit(
			boost::bind(
				( int ( *)( boost::function< int() > const&, boost::barrier &) ) barrier_fn,
				fn,
				boost::ref( b) ) );
		std::vector< int > buffer;
		tp::task< void > t(
			pool.submit(
				boost::bind(
					buffer_fibonacci_fn,
					boost::ref( buffer),
					10) ) );
		pool.submit(
			boost::bind(
				buffer_fibonacci_fn,
				boost::ref( buffer),
				0) );
		t.interrupt();
		b.wait();
		pool.shutdown();
		BOOST_CHECK_EQUAL( buffer[0], 0);
		BOOST_CHECK_EQUAL( buffer.size(), std::size_t( 1) );
		bool thrown( false);
		try
		{ t.get(); }
		catch ( boost::thread_interrupted const&)
		{ thrown = true; }
		BOOST_CHECK( thrown);
	}

	// check cancelation
	void test_case_13()
	{
		typedef tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool_type;
		pool_type pool(
			tp::poolsize( 1),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		boost::barrier b( 2);
		boost::function< int() > fn(
			boost::bind(
				fibonacci_fn,
				10) );
		pool.submit(
			boost::bind(
				( int ( *)( boost::function< int() > const&, boost::barrier &) ) barrier_fn,
				fn,
				boost::ref( b) ) );
		tp::task< int > t1(
			pool.submit(
				boost::bind(
					add_fn,
					1,
					2) ) );
		tp::task< int > t2(
			pool.chained_submit(
				boost::bind(
					add_fn,
					3,
					t1.get() ),
				t1) );
		tp::task< int > t3(
			pool.chained_submit(
				boost::bind(
					add_fn,
					4,
					t2.get() ),
				t2) );
		t2.interrupt();
		b.wait();
		bool thrown( false);
		try
		{ t3.get(); }
		catch ( boost::thread_interrupted const&)
		{ thrown = true; }
		BOOST_CHECK( thrown);
	}

	// check cancelation
	void test_case_14()
	{
		typedef tp::pool<
			tp::bounded_channel< tp::fifo >
		> pool_type;
		pool_type pool(
			tp::poolsize( 1),
			tp::high_watermark( 10),
			tp::low_watermark( 10) );
		boost::barrier b( 2);
		boost::function< int() > fn(
			boost::bind(
				fibonacci_fn,
				10) );
		pool.submit(
			boost::bind(
				( int ( *)( boost::function< int() > const&, boost::barrier &) ) barrier_fn,
				fn,
				boost::ref( b) ) );
		std::vector< tp::task< int > > f;
		for ( int i( 0); i <= 5; ++i)
			f.push_back(
				pool.lazy_submit(
					boost::bind(
						fibonacci_fn,
						i) ) );
		f[2].interrupt();
		f[3].interrupt();
		b.wait();
		BOOST_CHECK_EQUAL( f[0].get(), 0);
		BOOST_CHECK_EQUAL( f[4].get(), 3);
		BOOST_CHECK( ! f[1].ready() );
		BOOST_CHECK( ! f[5].ready() );
		bool thrown( false);
		try
		{ f[2].get(); }
		catch ( boost::thread_interrupted const&)
		{ thrown = true; }
		BOOST_CHECK( thrown);
	}
};

boost::unit_test::test_suite * init_unit_test_suite( int, char* [])
{
	boost::unit_test::test_suite * test( BOOST_TEST_SUITE("Boost.ThreadPool: fixed bounded_channel< fifo > pool test suite") );

	boost::shared_ptr< fixed_bounded_channel_fifo > instance( new fixed_bounded_channel_fifo() );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_1, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_2, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_3, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_4, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_5, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_6, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_7, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_8, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_9, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_10, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_11, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_12, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_13, instance) );
	test->add( BOOST_CLASS_TEST_CASE( & fixed_bounded_channel_fifo::test_case_14, instance) );

	return test;
}