/*******************************************************************************
 * Copyright (c) 2005 Intel Corporation.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * Contributors:
 *    Vishnu K Naikawadi, Intel - Initial implementation
 *    Andy Kaylor, Intel - Modified for sample client to make it a stress test
 *
 * $Id: MultiThread.cpp,v 1.5 2005/11/07 20:51:59 akaylor Exp $
 *
 *******************************************************************************/ 


#include <iostream>
#include <stdlib.h>
#include <stdio.h>
using namespace std;

#include "tptp/client/INode.h"
#include "tptp/client/NodeFactory.h"
#include "tptp/client/Agent.h"
#include "tptp/client/Collector.h"
#include "tptp/client/Constants.h"
#include "tptp/client/IProcess.h"
#include "tptp/TPTPSupportTypes.h"
#include "TestDataProcessor.h"
#include "tptp/TPTPOSCalls.h"
#include "tptp/TPTPUtils.h"

#include <time.h>

using namespace TPTP::Client;

/* Test setting defaults */
int  numIterations = 30;

bool bTestSendData    = true;
bool bTestReceiveData = true;

/* Number of threads of each type */
int  numGetAndRelease  = 5;
int  numFullCycle      = 0;
int  numSendAndReceive = 0;

/* Forward declarations */
THREAD_USER_FUNC_RET_TYPE getAndReleaseThread(LPVOID args);
THREAD_USER_FUNC_RET_TYPE sendAndReceiveThread(LPVOID args);
THREAD_USER_FUNC_RET_TYPE fullCycleThread(LPVOID args);
int processCommandLineOptions( int argc, char* argv[] );

/* termination status variables */
Semaphore_t  startTestSem;
Semaphore_t  threadStartedSem;
Semaphore_t  threadFinishedSem;
Lock_t       activeThreadsLock;
int          numActiveThreads;

/**
 *********************************************************
 *                                                       
 * @brief                                                
 *    Test Client
 *
 *********************************************************/
int main(int argc, char* argv[])
{
	char*            hostName = "localhost";
	int              portNum = 10006;
	int              rc = 0 ;
	INode*           TestNode = 0;
	int              i;
	AgentController* ACProxy = 0;
	BOOL             done = FALSE;

	rc = processCommandLineOptions( argc, argv );
	if ( rc < 0 )
	{
		getchar();
		return rc;
	}

	numActiveThreads = 0;
	tptp_initializeLock( &activeThreadsLock );

	//Create a Node that represents the Target Machine
	TestNode = NodeFactory::createNode(hostName);

	//Connect to the Agent Controller on that machine.
	printf( "Connecting to the Agent Controller on %s...\n\n", hostName );
	ACProxy = TestNode->connect(portNum);

	if (ACProxy == 0)
	{
		printf( "Unable to connect\n" );
		printf( "Press enter to exit...\n" );
		getchar();
		return -1;
	}

	/* Get ready to synchronize */
	tptp_initializeSemaphore( &startTestSem );
	tptp_initializeSemaphore( &threadStartedSem );
	tptp_initializeSemaphore( &threadFinishedSem );

	for ( i = 0; i < numGetAndRelease; i++ )
	{
		TID    tid;
		HANDLE thandle;

		startNewThread(getAndReleaseThread, ACProxy, &tid, &thandle );

		tptp_getWriteLock( &activeThreadsLock );
		numActiveThreads++;
		tptp_releaseWriteLock( &activeThreadsLock );

		tptp_waitSemaphore( &threadStartedSem );
	}

	for ( i = 0; i < numSendAndReceive; i++ )
	{
		TID    tid;
		HANDLE thandle;

		startNewThread(sendAndReceiveThread, ACProxy, &tid, &thandle );

		tptp_getWriteLock( &activeThreadsLock );
		numActiveThreads++;
		tptp_releaseWriteLock( &activeThreadsLock );

		tptp_waitSemaphore( &threadStartedSem );
	}

	for ( i = 0; i < numFullCycle; i++ )
	{
		TID    tid;
		HANDLE thandle;

		startNewThread(fullCycleThread, ACProxy, &tid, &thandle );

		tptp_getWriteLock( &activeThreadsLock );
		numActiveThreads++;
		tptp_releaseWriteLock( &activeThreadsLock );

		tptp_waitSemaphore( &threadStartedSem );
	}

	tptp_postSemaphore( &startTestSem );

	/* Wait for all threads to complete */
	while ( !done )
	{			 
		/* Wait for a thread to finish */
		tptp_waitSemaphore( &threadFinishedSem );

		/* See if it's the last one */
		tptp_getReadLock( &activeThreadsLock );
		if ( numActiveThreads == 0 )
		{
			done = TRUE;
		}
		tptp_releaseReadLock( &activeThreadsLock );
	}
	tptp_deleteLock( &activeThreadsLock );

	/* Clean up */
	tptp_deleteSemaphore( &startTestSem );
	tptp_deleteSemaphore( &threadStartedSem );
	tptp_deleteSemaphore( &threadFinishedSem );

	ACProxy->destroy();

	NodeFactory::deleteNode(TestNode);
		
	printf( "All finished\n" );

	printf( "Press enter to exit...\n" );
	getchar();

	return 0;
}


THREAD_USER_FUNC_RET_TYPE getAndReleaseThread(LPVOID args)
{
	int i;
	AgentController* ACProxy = (AgentController*)args;

	if ( ACProxy == NULL )
	{
		printf( "Internal error in test code.  ACProxy is NULL.\n" );
		return 0;
	}

	/* Tell the main thread we're ready */
	tptp_postSemaphore( &threadStartedSem );
	/* Wait for all the other threads to be ready too */
	tptp_waitSemaphore( &startTestSem );
	/* Trigger any other threads to wake up also */
	tptp_postSemaphore( &startTestSem );

	/* This takes less time so we do it more often */
	for ( i = 0; i < numIterations*10; i++ )
	{
		//Get an Agent - StressAgent in this case
		Collector* stressAgent = new Collector("org.eclipse.tptp.StressAgent");

		if ( stressAgent == NULL )
		{
			printf( "Unable to allocate Agent object org.eclipse.tptp.StressAgent\n" );
			return (THREAD_USER_FUNC_RET_TYPE)-1;
		}
		
		printf( "\nGetting the Agent...\n" );

		int agentref = ACProxy->getAgent(stressAgent, TPTP_CONTROLLER_ACCESS | TPTP_CREATE_INSTANCE);
		if(agentref == -1)
		{
			printf( "Unable to get Agent org.eclipse.tptp.StressAgent\n" );
			return (THREAD_USER_FUNC_RET_TYPE)-1;
		}
		printf( "The Stress Agent ID: %d\n", stressAgent->getAgentID() );
		
		// Release the time collector 
		printf( "\nReleasing the Agent...\n" );
		stressAgent->releaseAgent();
		
		delete stressAgent;
	}

	tptp_getWriteLock( &activeThreadsLock );
	numActiveThreads--;
	tptp_releaseWriteLock( &activeThreadsLock );
	tptp_postSemaphore( &threadFinishedSem );

	return (THREAD_USER_FUNC_RET_TYPE)0;
}

THREAD_USER_FUNC_RET_TYPE sendAndReceiveThread(LPVOID args)
{
	AgentController* ACProxy = (AgentController*)args;
	int i;
	char buffer[1024];

	// Prepare the test data
	for ( i = 0; i < 1024; i++ )
	{
		buffer[i] = i/8;
	}

	if ( ACProxy == NULL )
	{
		printf( "Internal error in test code.  ACProxy is NULL.\n" );
		return (THREAD_USER_FUNC_RET_TYPE)0;
	}

	//Get an Agent - StressAgent in this case
	Collector* stressAgent = new Collector("org.eclipse.tptp.StressAgent");
	
	printf( "\nGetting the Agent...\n" );

	int agentref = ACProxy->getAgent(stressAgent, TPTP_CONTROLLER_ACCESS | TPTP_CREATE_INSTANCE);
	if(agentref == -1)
	{
		printf( "Unable to get Agent org.eclipse.tptp.StressAgent\n" );
		return (THREAD_USER_FUNC_RET_TYPE)-1;
	}
	printf( "The Stress Agent ID: %d\n", stressAgent->getAgentID() );
		
	//Establish Data Path and Data Listener
	printf( "Establish the Data Channel Path with the Stress Agent...\n" );
	stressAgent->createDataConnection(TPTP_DATA_PATH_TWOWAY);
	TestDataProcessor* dataProcessor = new TestDataProcessor();
	int dataConnectionID = stressAgent->addDataListener(dataProcessor);

	/* Tell the main thread we're ready */
	tptp_postSemaphore( &threadStartedSem );
	/* Wait for all the other threads to be ready too */
	tptp_waitSemaphore( &startTestSem );
	/* Trigger any other threads to wake up also */
	tptp_postSemaphore( &startTestSem );


	for ( i = 0; i < numIterations; i++ )
	{
		if ( bTestReceiveData )
		{
			//Send RUN command - Starts the StressAgent
			printf( "R" );
			stressAgent->run();
		}

		if ( bTestSendData )
		{
			//Send Data to StressAgent		
			for(int j=0;j<100;j++)
			{
				// sent data through the path 
				stressAgent->sendData(buffer, 100);
			}
		}
		else
		{
			Sleep( 5000 );
		}
		
		if ( bTestReceiveData )
		{
			//Send STOP command - Stops the StressAgent
			//and returns the Elapsed Time since RUN command
			//Also sends bunch of data back
			printf( "S" );
			stressAgent->stop();
		}

	}

	// Release the time collector 
	printf( "\nReleasing the Agent...\n" );
	stressAgent->releaseAgent();
		
	delete stressAgent;

	tptp_getWriteLock( &activeThreadsLock );
	numActiveThreads--;
	tptp_releaseWriteLock( &activeThreadsLock );
	tptp_postSemaphore( &threadFinishedSem );

	return (THREAD_USER_FUNC_RET_TYPE)0;
}

THREAD_USER_FUNC_RET_TYPE fullCycleThread(LPVOID args)
{
	AgentController* ACProxy = (AgentController*)args;
	int i;
	char buffer[1024];

	// Prepare the test data
	for ( i = 0; i < 1024; i++ )
	{
		buffer[i] = i/8;
	}

	if ( ACProxy == NULL )
	{
		printf( "Internal error in test code.  ACProxy is NULL.\n" );
		return (THREAD_USER_FUNC_RET_TYPE)0;
	}

	/* Tell the main thread we're ready */
	tptp_postSemaphore( &threadStartedSem );
	/* Wait for all the other threads to be ready too */
	tptp_waitSemaphore( &startTestSem );
	/* Trigger any other threads to wake up also */
	tptp_postSemaphore( &startTestSem );


	for ( i = 0; i < numIterations; i++ )
	{
		//Get an Agent - StressAgent in this case
		Collector* stressAgent = new Collector("org.eclipse.tptp.StressAgent");
		
		printf( "\nGetting the Agent...\n" );

		int agentref = ACProxy->getAgent(stressAgent, TPTP_CONTROLLER_ACCESS | TPTP_CREATE_INSTANCE);
		if(agentref == -1)
		{
			printf( "Unable to get Agent org.eclipse.tptp.StressAgent\n" );
			return (THREAD_USER_FUNC_RET_TYPE)-1;
		}
		printf( "The Stress Agent ID: %d\n", stressAgent->getAgentID() );
		

		//Establish Data Path and Data Listener
		printf( "Establish the Data Channel Path with the Stress Agent...\n" );
		stressAgent->createDataConnection(TPTP_DATA_PATH_TWOWAY);
		TestDataProcessor* dataProcessor = new TestDataProcessor();
		int dataConnectionID = stressAgent->addDataListener(dataProcessor);

		if ( bTestReceiveData )
		{
			//Send RUN command - Starts the StressAgent
			printf( "R" );
			stressAgent->run();
		}

		if ( bTestSendData )
		{
			//Send Data to StressAgent		
			for(int j=0;j<100;j++)
			{
				// sent data through the path 
				int length = strlen(buffer);
				stressAgent->sendData(buffer, 100);
			}
		}
		else
		{
			Sleep( 5000 );
		}
		
		if ( bTestReceiveData )
		{
			//Send STOP command - Stops the StressAgent
			//and returns the Elapsed Time since RUN command
			//Also sends bunch of data back
			printf( "S" );
			stressAgent->stop();
		}

		// Release the time collector 
		printf( "\nReleasing the Agent...\n" );
		stressAgent->releaseAgent();
		
		delete stressAgent;
	}

	tptp_getWriteLock( &activeThreadsLock );
	numActiveThreads--;
	tptp_releaseWriteLock( &activeThreadsLock );
	tptp_postSemaphore( &threadFinishedSem );

	return (THREAD_USER_FUNC_RET_TYPE)0;
}

int processCommandLineOptions( int argc, char* argv[] )
{
	int i;

	for ( i = 1; i < argc; i++ )
	{

		if ( argv[i][0] == '-' )
		{
			switch ( argv[i][1] )
			{
				case '?':
					printf( "Usage: MultiThread [options]\n\n" );
					printf( "   -?       Show this message\n" );
					printf( "   -n<N>    Repeat each test for <N> iterations\n" );
					printf( "   -s<0|1>  Enable or disable the send data tests\n" );
					printf( "   -r<0|1>  Enable or disable the receive data tests\n" );
					printf( "   -g<N>    Start <N> threads for the Get and Release test\n" );
					printf( "   -f<N>    Start <N> threads for the Full Cycle test\n" );
					printf( "   -d<N>    Start <N> threads for the Send and Receive test\n" );
					return -1;
					break;

				case 'n':
					numIterations = atoi( &argv[i][2] );
					if ( numIterations == 0 )
					{
						printf( "Bad value for -n option\n\n" );
						printf( "   -n<N>  Repeat each test for <N> iterations\n" );
						printf( "          Specify a non-zero integer for <N>\n" );
						return -1;
					}
					break;

				case 's':
					if ( argv[i][2] == '0' )
					{
						bTestSendData = false;
					}
					else if ( argv[i][2] == '1' )
					{
						bTestSendData = true;
					}
					else
					{
						printf( "Bad value for -n option\n\n" );
						printf( "   -s<0|1>  Enable or disable the send data tests\n" );
						printf( "            Use either -s0 or -s1\n" );
						return -1;
					}
					break;

				case 'r':
					if ( argv[i][2] == '0' )
					{
						bTestReceiveData = false;
					}
					else if ( argv[i][2] == '1' )
					{
						bTestReceiveData = true;
					}
					else
					{
						printf( "Bad value for -n option\n\n" );
						printf( "   -r<0|1>  Enable or disable the receive data tests\n" );
						printf( "            Use either -r0 or -r1\n" );
						return -1;
					}
					break;

				case 'g':
					numGetAndRelease = atoi( &argv[i][2] );
					if ( (numGetAndRelease == 0) && (argv[i][2] != '0') )
					{
						printf( "Bad value for -g option\n\n" );
						printf( "   -g<N>  Start <N> threads for the Get and Release test\n" );
						printf( "          Specify an integer for <N>\n" );
						return -1;
					}
					break;

				case 'f':
					numFullCycle = atoi( &argv[i][2] );
					if ( (numFullCycle == 0) && (argv[i][2] != '0') )
					{
						printf( "Bad value for -f option\n\n" );
						printf( "   -f<N>  Start <N> threads for the Full Cycle test\n" );
						printf( "          Specify an integer for <N>\n" );
						return -1;
					}
					break;

				case 'd':
					numGetAndRelease = atoi( &argv[i][2] );
					if ( (numGetAndRelease == 0) && (argv[i][2] != '0') )
					{
						printf( "Bad value for -d option\n\n" );
						printf( "   -d<N>  Start <N> threads for the Send and Receive test\n" );
						printf( "          Specify an integer for <N>\n" );
						return -1;
					}
					break;

				default:
					break;
			}
		}
	}

	return 0;
}
