#include "XPlatform.h"
#include "BaseMultiStreamIO.h"
#include "Socket.h"

//////////////
// IOClient //
//////////////
const int IOClient::BYTES_PER_SECTOR = 1<<9;					// 512B Sector sizes
//const int IOClient::IO_BLOCK_SIZE = BYTES_PER_SECTOR<<13;		// 4MB IO Chunks
const int IOClient::IO_BLOCK_SIZE = BYTES_PER_SECTOR<<12;		// 2MB IO Chunks
long long IOClient::ReadBytes  = 0;
long long IOClient::WriteBytes = 0;

IOClient::IOClient( void )
{
	InitializeCriticalSection( &lock );
	server = NULL;
}
IOClient::~IOClient( void )
{
	DeleteCriticalSection( &lock );
}
void IOClient::SetServer( class MultiStreamIOServer* server )
{
	this->server = server;
	if( server ) server->AddClient( this );
}
/////////////////////////
// MultiStreamIOServer //
/////////////////////////
MultiStreamIOServer::MultiStreamIOServer(void)
{
	InitializeCriticalSection( &pendingLock );
	InitializeCriticalSection( &clientLock );

	pendingClient = NULL;
	ioThread = RunThread( IOThread , this );
	if( !TestThreadHandle( ioThread ) ) fprintf( stderr , "Failed to create I/O thread\n" ) , exit( 0 );
}
MultiStreamIOServer::~MultiStreamIOServer(void)
{
	WaitOnIO();
	DeleteCriticalSection( &pendingLock );
	DeleteCriticalSection( &clientLock );
}
bool MultiStreamIOServer::SetPending( IOClient* client )
{
	EnterCriticalSection( &pendingLock );
	if( !pendingClient && client )
	{
		pendingClient = client;
		LeaveCriticalSection( &pendingLock );
		return true;
	}
	else
	{
		LeaveCriticalSection( &pendingLock );
		return false;
	}
}
void MultiStreamIOServer::AddClient( IOClient* client )
{
	EnterCriticalSection( &clientLock );
	clients.push_back( client );
	LeaveCriticalSection( &clientLock );
}
void MultiStreamIOServer::WaitOnIO( void )
{
	if( IsActiveThread( ioThread ) )
	{
		while( 1 )
		{
			// As long as the server has work to do, it cannot be terminated.
			EnterCriticalSection( &clientLock );
			if( !clients.size() )
			{
				if( !InterruptThread( ioThread ) ) fprintf( stderr , "Failed to interrupt MultiStreamIOServer thread\n" );
				LeaveCriticalSection( &clientLock );
				break;
			}
			LeaveCriticalSection( &clientLock );
		}
	}
}
int MultiStreamIOServer::clientNum( void )
{
	EnterCriticalSection( &clientLock );
	int sz = int( clients.size() );
	LeaveCriticalSection( &clientLock );
	return sz;
}
THREAD_FUNCTION_OUT MultiStreamIOServer::IOThread( THREAD_FUNCTION_IN vparams )
{
	MultiStreamIOServer* server = (MultiStreamIOServer*)vparams;
	std::vector< IOClient* >& clients = server->clients;
	int idx = 0;
	{
		while( 1 )
		{
#ifdef USE_BOOST_THREADS
			if( boost::this_thread::interruption_requested() ) break;
#endif // USE_BOOST_THREADS
			bool hasPending = false;
			{
				EnterCriticalSection( &server->pendingLock );
				if( server->pendingClient )
				{
					if( server->pendingClient->Service() == IOClient::NONE ) SleepThisThread( 0 );
					else server->pendingClient = NULL;
					hasPending = true;
				}
				LeaveCriticalSection( &server->pendingLock );
			}
			if( !hasPending )
			{
				EnterCriticalSection( &server->clientLock );
				bool ioDone = false;
				for( int i=0 ; i<clients.size() && !ioDone ; i++ )
				{
					idx = (idx+1)%clients.size();
					switch( clients[idx]->Service() )
					{
					case IOClient::COMPLETE:
						clients[idx]->SetServer( NULL );
						clients[idx] = clients[clients.size()-1];
						clients.pop_back();
					case IOClient::SUCCESS:
						ioDone = true;
						break;
					}
				}
				LeaveCriticalSection( &server->clientLock );
				if( !ioDone ) SleepThisThread( 1 );
				else SleepThisThread( 0 );
			}
		}
	}
	return 0;
}
