// Include files
#include "iocpudpsend.h"

// Local definitions
#pragma warning( disable: 4127 )

#define SafeCloseHandle(P)  if (P!=NULL) { CloseHandle(P); (P)=NULL; } // 핸들를 안전하게 해제
#define SafeDelete(P)       if (P!=NULL) { delete(P);      (P)=NULL; } // "

// Global data

// cIocpUdpSend Constructor
cIocpUdpSend::cIocpUdpSend(void)
{
	// Critical Section를 초기화.
	InitializeCriticalSection( &mCs );

	mIocp                   = NULL;
	mSocket                 = INVALID_SOCKET;
	mIocpWorkerThreadNumber = 0;
	mIoContextPool          = NULL;
	mIocpBackendThread      = NULL;

	mIoContextFrontBuffer   = NULL;
	mIoContextBackBuffer    = NULL;

	mRunServer              = false;
	mEndServer              = false;
}

// ~cIocpUdpSend Destructor.
cIocpUdpSend::~cIocpUdpSend(void)
{
	// 종료
	Shutdown( );

	// Critical Section를 해제.
	DeleteCriticalSection( &mCs );
}

// Initialize Method
bool cIocpUdpSend::Initialize(unsigned short numWorkerThreads)
{
	// CreateIoCompletionPort를 이용하여 completion port 오브젝트를 생성할 때 일반적으로 신경써야 할 파
	// 라메터는 NumberOfConcurrentThreads 이다. 처음 3개의 파라메터는 값을 설정하지 않는다.
	//
	// NumberOfConcurrentThreads는 현재 completion port에 대하여 몇개의 스레드를 실행할 것인가를 결정한
	// 다. 가장 이상적인 경우는 스레드의 context switching을 방지하기 위하여 서비스에 대하여 하나의 프로
	// 세서당 하나의 스레드만 동작하도록 하는 것이다. NumberOfConcurrentThreads의 값을 0으로 설정하면 시
	// 스템은 프로세서에 적합한 스레드 개수를 할당하여 사용한다.
	// (하나의 CPU에 대하여 IOCP 하나당 하나의 스레드를 생성)
	
	mIocp = CreateIoCompletionPort( INVALID_HANDLE_VALUE, NULL, 0, 0 );
	if ( mIocp == NULL )
		return false;

	// CreateIoCompletionPort 오브젝트를 생성한 뒤에 작업스레드를 생성하고 소켓핸들을 지정할 수 있는데,
	// 이때 생성하는 작업스레드는, 스레드가 블록될 가능성이 있다면, NumberOfConcurrentThreads 의 갯수보
	// 다 많이 생성하는 것이 좋다. NumberOfConcurrentThreads만큼 스레드가 항상 실행되게 할 수 있기 때문
	// 에, 블록된 스레드가 있다면 다른 스레드가 대신 동작할 것이기 때문이다.
	//
	SYSTEM_INFO si;
	DWORD       threadId;

	// 시스템에 몇 개의 프로세서가 있는지 확인한다.
	GetSystemInfo( &si );

	mIocpWorkerThreadNumber = min( si.dwNumberOfProcessors * numWorkerThreads, IOCP_MAX_WORKER_THREAD );
	for ( int i = 0; i < mIocpWorkerThreadNumber; i++ )
	{
		mIocpWorkerThread[i] = CreateThread( NULL, 0, WorkerThreadStartingPoint, (LPVOID)this, 0, &threadId );
		if ( mIocpWorkerThread[i] == NULL )
			return false;
	}

	// I/O Context Pool를 생성한다.
	// 0xffff: UDP(65,535) 최대 패킷 사이즈
	// 0x0044: IPv4(60) + UDP(8) 의 해더 사이즈
	mIoContextPool = new cIoContextPool( MAX_UDP_PACKET_SIZE );
	if ( mIoContextPool == NULL )
		return false;

	// I/O Context Buffer를 생성한다.
	mIoContextFrontBuffer = (IoContextBuffer*)GlobalAlloc( GPTR, sizeof(IoContextBuffer) );
	mIoContextBackBuffer  = (IoContextBuffer*)GlobalAlloc( GPTR, sizeof(IoContextBuffer) );
	if ( !mIoContextFrontBuffer && !mIoContextBackBuffer )
		return false;

	// connect할 클라이언트 소켓을 생성한다.
	mSocket = WSASocket( AF_INET, SOCK_DGRAM, IPPROTO_UDP, NULL, 0, WSA_FLAG_OVERLAPPED );
	if ( mSocket == INVALID_SOCKET )
		return false;

	// Disable send buffering on the socket. Setting SO_SNDBUF to 0 causes winsock to stop bufferring
	// sends and perform sends directly from our buffers, thereby reducing CPU usage.
	//
	// SO_SNDBUF를 0으로하여 송신버퍼를 사용하지 않는다. 송신버퍼가 0이 아니면 불필요하게 이 송신버퍼를
	// 거쳐서 overlapped I/O를 호출했을 때 제공한 버퍼로 복사가 이루어진다.
	// 송신버퍼 공간이 없으면 overlapped I/O를 호출햇을 때 제공한 버퍼로 시스템에서 직접 데이터가 복사된
	// 다.
	//
	int zero = 0;
	if ( setsockopt( mSocket, SOL_SOCKET, SO_SNDBUF, (char*)&zero, sizeof(zero) ) == SOCKET_ERROR )
	{
		closesocket( mSocket );
		return false;
	}

	// 생성된 소켓 핸들을 completion port에 할당한다.
	if ( CreateIoCompletionPort( (HANDLE)mSocket, mIocp, (ULONG_PTR)NULL, 0 ) == NULL )
	{
		closesocket( mSocket );
		return false;
	}

	mEndServer = false;
	mRunServer = true;

	// Backend Thread를 생성한다.
	mIocpBackendThread = CreateThread( NULL, 0, BackendThreadStartingPoint, (LPVOID)this, 0, &threadId );
	if ( mIocpBackendThread == NULL )
		return false;

	return true;
}

// Shutdown Method
void cIocpUdpSend::Shutdown(DWORD maxWait)
{
	// 안전하게 종료한다.
	mEndServer = true;
	mRunServer = false;

	// Backend Thread가 종료될때까지 대기
	if ( mIocpBackendThread != NULL )
	{
		WaitForSingleObjectEx( mIocpBackendThread, maxWait, TRUE );
		CloseHandle( mIocpBackendThread );
		mIocpBackendThread = NULL;
	}

	// 안전하게 종료한다.
	SOCKET sockTemp = mSocket;;

	mSocket = INVALID_SOCKET;

	if ( sockTemp != INVALID_SOCKET )
	{
		closesocket( sockTemp );
	}

	// Cause worker threads to exit
	if ( mIocp != NULL )
	{
		for ( int i = 0; i < mIocpWorkerThreadNumber; i++ )
		{
			PostQueuedCompletionStatus( mIocp, 0, 0, IOCP_SHUTDOWN );
		}
	}

	// Make sure worker threads exits.
	for ( int i = 0; i < mIocpWorkerThreadNumber; i++ )
	{
		if ( WaitForSingleObject( mIocpWorkerThread[i], 60000 ) != WAIT_OBJECT_0 )
		{
			DWORD exitCode;
			GetExitCodeThread( mIocpWorkerThread[i], &exitCode);
			if ( exitCode == STILL_ACTIVE )
			{
				TerminateThread( mIocpWorkerThread[i], 0 );
			}
		}
		CloseHandle( mIocpWorkerThread[i] );
		mIocpWorkerThread[i] = NULL;
	}

	// Overlapped I/O Model Pool및 Socket Context Pool를 해제한다.
	if ( mIoContextFrontBuffer )
	{
		GlobalFree( mIoContextFrontBuffer );
		mIoContextFrontBuffer = NULL;
	}
	if ( mIoContextBackBuffer )
	{
		GlobalFree( mIoContextBackBuffer );
		mIoContextBackBuffer = NULL;
	}
	// Overlapped I/O Model Pool를 해제한다.
	SafeDelete( mIoContextPool );

	// completion port를 해제한다.
	SafeCloseHandle( mIocp );
}

// GetIoPoolUsage Method
void cIocpUdpSend::GetIoPoolUsage(SIZE_T& quotaPagedPoolUsage, SIZE_T& quotaNonePagedPoolUsage, SIZE_T& workingSetSize)
{
	if ( mIoContextPool != NULL )
	{
		mIoContextPool->GetProcessMemoryInfo( quotaPagedPoolUsage, quotaNonePagedPoolUsage, workingSetSize );
	}
}

// QueueRequest Method
BOOL cIocpUdpSend::QueueRequest(ULONG_PTR completionKey, LPOVERLAPPED overlapped, DWORD bytesTransferred)
{
	return PostQueuedCompletionStatus( mIocp, bytesTransferred, completionKey, overlapped );
}

// SendExec Method
bool cIocpUdpSend::SendExec(PerIoContext* perIoContext)
{
	WSABUF wsaBuf;
	DWORD  sendBytes = 0;
	DWORD  flags     = 0;
	int    addrlen   = sizeof(SOCKADDR_IN);
	int    retcode;

	if ( perIoContext->offset > perIoContext->length )
	{
		PostQueuedCompletionStatus( mIocp, 0, (ULONG_PTR)NULL, (LPOVERLAPPED)perIoContext );
		return false;
	}

	wsaBuf.len = perIoContext->offset;
	wsaBuf.buf = perIoContext->buffer;

	retcode = WSASendTo( perIoContext->socket,
						 &wsaBuf,
						 1,
						 &sendBytes,
						 flags,
						 (SOCKADDR*)&perIoContext->addr,
						 addrlen,
						 &(perIoContext->wsaOverlapped),
						 NULL );

	if ( (retcode == SOCKET_ERROR) && (WSAGetLastError() != WSA_IO_PENDING) )
	{
		PostQueuedCompletionStatus( mIocp, 0, (ULONG_PTR)NULL, (LPOVERLAPPED)perIoContext );
		return false;
	}
	return true;
}

// SendPost Method
bool cIocpUdpSend::SendPost(PerIoContext* perIoContext)
{
	cCSLock        lock( &mCs );
	PerIoContext** buffer = mIoContextBackBuffer->buffer;
	long&          offset = mIoContextBackBuffer->offset;

	if ( offset < MAX_IO_CONTEXT_BUFFER_LEN )
	{
		buffer[ offset ] = perIoContext;
		offset++;
		return true;
	}

	mIoContextPool->ReleaseIoContext( perIoContext );
	return false;
}

/*-- IoContextPresent Method
*/
void cIocpUdpSend::IoContextPresent( )
{
	IoContextBuffer* temp;

	// Double Buffering 처리.
	EnterCriticalSection( &mCs );

		temp                  = mIoContextBackBuffer;
		mIoContextBackBuffer  = mIoContextFrontBuffer;
		mIoContextFrontBuffer = temp;

	LeaveCriticalSection( &mCs );

	// 발신된 데이터 처리.
	PerIoContext** buffer = mIoContextFrontBuffer->buffer;
	long&          offset = mIoContextFrontBuffer->offset;

	while ( offset > 0 )
	{
		SendExec( (*buffer) );
		(*buffer) = NULL;

		buffer++;
		offset--;

		Sleep( 10 );
	}
}

// SendComplete Method
bool cIocpUdpSend::SendComplete(ULONG_PTR     /*completionKey*/,
							    PerIoContext* perIoContext,
							    DWORD         /*bytesTransferred*/)
{
	// 사용이 완료된 I/O Context는 회수한다.
	mIoContextPool->ReleaseIoContext( perIoContext );
	return true;
}

// CallbackComplete Method
bool cIocpUdpSend::CallbackComplete(ULONG_PTR     /*completionKey*/,
								    PerIoContext* perIoContext,
								    DWORD         /*bytesTransferred*/)
{
	// 사용이 완료된 I/O Context는 회수한다.
	mIoContextPool->ReleaseIoContext( perIoContext );
	return true;
}

// WorkerThread Method
DWORD cIocpUdpSend::WorkerThread( )
{
	DWORD         bytesTransferred;
	ULONG_PTR     completionKey;
	OVERLAPPED*   overlapped;
	BOOL          retValue;

	PerIoContext* perIoContext;

	while ( true )
	{
		// completion port에 할당된 모든 소켓의 I/O 완료를 기다린다.
		retValue = GetQueuedCompletionStatus( mIocp, &bytesTransferred, &completionKey, &overlapped, INFINITE );

		// Shutdown
		if ( overlapped == IOCP_SHUTDOWN )
			return 0;

		perIoContext = (PerIoContext*)overlapped;

		if ( retValue == FALSE || bytesTransferred == 0 )
		{
			if ( perIoContext != NULL )
			{
				// 사용이 완료된 I/O Context는 회수한다.
				perIoContext->offset = max( perIoContext->offset, bytesTransferred );
				mIoContextPool->ReleaseIoContext( perIoContext );
			}
			continue;
		}

		switch ( perIoContext->requestType )
		{
		case IOCP_REQUEST_WRITE:
			SendComplete( completionKey, perIoContext, bytesTransferred );
			break; // Send 완료 핸들러 함수.

		case IOCP_REQUEST_CALLBACK:
			CallbackComplete( completionKey, perIoContext, bytesTransferred );
			break; // Callback 완료 핸들러 함수.
		}
	}
	return 0;
}

// BackendThread Method
DWORD cIocpUdpSend::BackendThread( )
{
	while ( mEndServer != true )
	{
		IoContextPresent( );
		Sleep( 50 );
	}

	IoContextPresent( );
	Sleep( 1000 );
	return 0;
}

// WorkerThreadStartingPoint Method ( Worker Thread 시작 포인트 )
DWORD cIocpUdpSend::WorkerThreadStartingPoint(void* ptr)
{
	cIocpUdpSend* iocpUdpSend = (cIocpUdpSend*)ptr;
	return iocpUdpSend->WorkerThread( );
}

// BackendThreadStartingPoint Method ( Backend Thread 시작 포인트 )
DWORD cIocpUdpSend::BackendThreadStartingPoint(void* ptr)
{
	cIocpUdpSend* iocpUdpSend = (cIocpUdpSend*)ptr;
	return iocpUdpSend->BackendThread( );
}
