Leviathan/Library/Internal/source/network/XIOCP.cpp

#define	WIN32_LEAN_AND_MEAN
#include <windows.h>
#include <process.h>

#include <vector>
#include <list>
#include <algorithm>
#include <cassert>

#include "../../include/network/XIOCP.h"
#include "../../include/network/XIOCPConnection.h"
#include "../../include/network/XIOCPAcceptor.h"
#include "../../include/network/XIOCPDatagram.h"

#include "../../include/network/XNetworkUtil.h"
#include "../../include/network/XIOCPStruct.h"
#include "../../include/toolkit/XMemoryPool.h"
#include "../../include/toolkit/ILock.h"
#include "../../include/toolkit/XThread.h"
#include "../../include/toolkit/XConsole.h"
#include "../../include/toolkit/XEnv.h"
#include "../../include/toolkit/XThreadMonitor.h"
#include "../../include/logging/FileLog.h"
#include "../../include/dump/XExceptionHandler.h"

volatile LONG g_nDisconnectCnt;
static void	(*s_pfInitFunc)( int ) = NULL;

struct IOCPTAG
{	
	HANDLE						hIOCP;
	
	std::vector< uintptr_t >	vThreadHandle;
	std::vector< unsigned int >	vThreadId;

	OverlappedAllocator			*pOverlappedAllocator;

	INetworkEventReceiver*		pReceiver;
	XIOCP*						pIOCPMgr;
	volatile long* 				plActiveThreadCount;
	volatile long*				plInstructionCount;
	volatile long*				plCurrentThreadCount;
	volatile bool* 				pbIsPaused;
	bool						bIsFinished;

	IOCPTAG( INetworkEventReceiver* receiver)
	{
		bIsFinished = false;
		hIOCP		= NULL;
		pReceiver	= receiver;

		pIOCPMgr = NULL;
		plActiveThreadCount = NULL;
		plInstructionCount = NULL;
		plCurrentThreadCount = NULL;
		pbIsPaused = NULL;

		pOverlappedAllocator = new OverlappedAllocator;
	}

	virtual ~IOCPTAG()
	{
		delete pOverlappedAllocator;
	}

private:
	IOCPTAG( const IOCPTAG& );
	IOCPTAG& operator=( const IOCPTAG& );

};

bool XIOCP::onAcceptEvent( IOCPTAG * pTag, int nThreadNum, XOVERLAPPED * pOverlapped, bool bIsSuccess )
{
	XIOCPAcceptor* pAcceptor = static_cast< XIOCPAcceptor* >( pOverlapped->pObj );

	SOCKET hFileHandle = reinterpret_cast< SOCKET >( pOverlapped->hFileHandle );
	char buf[sizeof(sockaddr_in)*2 + 32];
	s_memcpy( buf, sizeof( buf ), pOverlapped->pBuf, sizeof(sockaddr_in)*2 + 32 );

	pAcceptor->deleteFromPendingList( hFileHandle );

	// 이전의 AcceptEx() 가 무효화 되었으므로 재차 콜을 해 준다.
	pAcceptor->pendAcceptRequest();

	if( bIsSuccess )
	{
		// NIOCPConnection() 을 조립해 아래 receiver 에 주어야 한다.
		XIOCPConnection * pConnection = static_cast< XIOCPConnection* >( pTag->pReceiver->createConnection( nThreadNum, pAcceptor, XSocket( hFileHandle ) ) );
		if ( !pConnection )
			pConnection = static_cast< XIOCPConnection* >( pTag->pReceiver->createConnection( XSocket( hFileHandle ) ) );

		if( pConnection )
		{
			pConnection->SetID( XOBJ_CONNECTION );
			pConnection->onConnect( buf );

			bool bIsNeedAddObject = true;
			// Event Receiver 호출
			if( pTag->pReceiver ) 
				bIsNeedAddObject = pTag->pReceiver->onAccept( nThreadNum, pAcceptor, pConnection );

			// 이것이 꼭 event receiver 보다 늦게 호출 되어야 한다~!
			if( bIsNeedAddObject ) 
			{
				pTag->pIOCPMgr->AddObject( pConnection );
				if( pTag->pReceiver )
					pTag->pReceiver->onAccepted( nThreadNum, pConnection );
			}
		}
	}
	else
	{
		closesocket( hFileHandle );
	}	

	return true;
}


bool XIOCP::onConnectionEvent( IOCPTAG * pTag, int nThreadNum, XOVERLAPPED * pOverlapped, int nSize )
{
	XIOCPConnection * pConnection = static_cast< XIOCPConnection* >( pOverlapped->pObj );

	// _oprint( "IOCP EVENT : 0x%08X\n", pConnection );

	switch( pOverlapped->cFlag )
	{
	case XIOCP_RECV:
		{
			pConnection->onRecvCompletionEvent( nSize );

			if( nSize < 1 )
			{	
				pConnection->decreaseQueryCount();
				pConnection->onDisconnect( 1 );
				break;
			}

			// Event Receiver 호출
			if( pTag->pReceiver ) pTag->pReceiver->onRead( nThreadNum, pConnection );

			// 정상적인 Recv 였다면..
			if( nSize > 0 && pConnection->IsConnected() )
			{
				// 이전의 WSARecv() 가 무효화 되었으므로 재차 콜을 해 준다.
				if( !pConnection->pendRecvRequest() ) 
				{
					pConnection->decreaseQueryCount(); // ???
					break;
				}
			}

			// 쿼리 카운트 감소
			pConnection->decreaseQueryCount();

			// 다른 쓰레드에서 disconnect 이벤트가 발생했다면 연결 끊어짐 처리
			if( pConnection->CheckDisconnectSignal() || !pConnection->IsConnected() )
			{
				pConnection->onDisconnect( 2 );
				break;
			}			
		
		}
		break;

	case XIOCP_SEND:
		{
			// Event Receiver 호출
			if( pTag->pReceiver ) pTag->pReceiver->onWrite( nThreadNum, pConnection );

			// 보내진 데이터를 Send Queue 에서 제거한다
			pConnection->onSendCompletionEvent( nSize );

			// 쿼리 카운트 감소
			pConnection->decreaseQueryCount();

			// nSize 가 1 이하이면 연결 끊어진것임
			if( nSize < 1 )
			{
				pConnection->onDisconnect( 3 );
				break;
			}

			// 다른 쓰레드에서 disconnect 이벤트가 발생했다면 연결 끊어짐 처리
			if( pConnection->CheckDisconnectSignal() )
			{
				pConnection->onDisconnect( 4 );
				break;
			}
		}
		break;

	case XIOCP_CONNECT:
		{
			if( nSize >= 0 )
			{
				pConnection->onConnectCompletionEvent( true );

				// Event Receiver 호출
				if( pTag->pReceiver ) 
					pTag->pReceiver->onConnect( nThreadNum, pConnection );
			}
			else
			{
				pConnection->onConnectCompletionEvent( false );

				// Event Receiver 호출
				if( pTag->pReceiver ) 
					pTag->pReceiver->onCantConnect( nThreadNum, pConnection );
			}
		}
	}

	return true;
}

bool XIOCP::onDatagramEvent(IOCPTAG *pTag, int nThreadNum, XOVERLAPPED *pOverlapped, int nSize)
{
	XIOCPDatagram *pDatagram = static_cast<XIOCPDatagram *>(pOverlapped->pObj);

	switch(pOverlapped->cFlag)
	{
		case XIOCP_DATAGRAMRECV:
			{
				pDatagram->OnRecvFromCompletionEvent(nSize);

				if(pTag->pReceiver) pTag->pReceiver->onRead(nThreadNum, pDatagram, nSize);

				pDatagram->OnPendRecvFromRequest();

				if(pDatagram->IsOpened()) while(!pDatagram->PendRecvFromRequest());

				pDatagram->DecreaseQueryCount();
			}
			
			break;
		case XIOCP_DATAGRAMSEND:
			{
				if(pTag->pReceiver) pTag->pReceiver->onWrite(nThreadNum, pDatagram, nSize);

				pDatagram->OnSendToCompletionEvent(nSize);

				pDatagram->DecreaseQueryCount();
			}

			break;
	}

	return true;
}

unsigned __stdcall XIOCP::IOCPWorkerThread( void* pArg )
{
	XError::Debug();

	IOCPTAG * pTag = static_cast< IOCPTAG* >( pArg );

	InterlockedIncrement( pTag->plActiveThreadCount );

	unsigned nThreadNum = 0;
	for( nThreadNum = 0;
		 nThreadNum < pTag->vThreadId.size() && GetCurrentThreadId() != pTag->vThreadId[nThreadNum];
		 nThreadNum++ );	// do nothing (nThreadNum 을 얻기 위한 루프임)

	char buf[1024];
	s_sprintf( buf, _countof( buf ), "IOCP %02d", nThreadNum );
	XSetThreadName( -1, buf );

	if( s_pfInitFunc )
	{
		s_pfInitFunc( nThreadNum );
	}

	int nRtn;
	DWORD dwNumberOfBytes;
	ULONG_PTR ulpCompletionKey;
	XOVERLAPPED *pOverlapped;

	while( true ) 
	{
		try
		{
			// Active thread count 조정  ( 대기모드이므로 1 감소 )
			InterlockedDecrement( pTag->plActiveThreadCount );

			nRtn = GetQueuedCompletionStatus( pTag->hIOCP,
											&dwNumberOfBytes,
											&ulpCompletionKey,
											(OVERLAPPED**)&pOverlapped,
											INFINITE );

			while( *pTag->pbIsPaused )
			{
				Sleep( 100 );
			}

			// Active thread count 조정  ( 처리모드이므로 1 증가 )
			InterlockedIncrement( pTag->plActiveThreadCount );

			// Instruction count 조정  ( 처리모드이므로 1 증가 )
			InterlockedIncrement( pTag->plInstructionCount );

			// 쓰레드 종료 시그널이 도착했다면 중지
			if( ulpCompletionKey == XIOCP_EVENT_STOP || pTag->bIsFinished )
			{
				break;
			}

			// 커넥션 끊어짐 이벤트라면..
			if( ulpCompletionKey == XIOCP_EVENT_CONNECTION_CLOSED )
			{
				// _oprint( "DISCONNECT EVENT : 0x%08X\n", static_cast< XIOCPConnection* >( pOverlapped->pObj ) );

				// 이벤트 리시버에 disconnect 이벤트 알려주자
				pTag->pReceiver->onDisconnect( nThreadNum, static_cast< XIOCPConnection* >( pOverlapped->pObj ) );

				InterlockedIncrement( &g_nDisconnectCnt );

				continue;
			}

			// GetQueuedCompletionStatus() 에러라면..
			if( !nRtn )
			{
				int nErrorCode = WSAGetLastError();
				// { ConnectEx 실패의 경우
				if( nErrorCode == WSAECONNREFUSED )
				{	
					if( pTag->pReceiver ) pTag->pReceiver->onCantConnect( nThreadNum, static_cast< IConnection* >( pOverlapped->pObj ) );
					continue;
				}
				// }

				// { 디스커넥트

				if(	nErrorCode == WSAENETDOWN ||
					nErrorCode == WSAENETUNREACH ||
					nErrorCode == WSAENETRESET ||
					nErrorCode == WSAECONNABORTED ||
					nErrorCode == WSAECONNRESET ||
					nErrorCode == WSAETIMEDOUT ||
					nErrorCode == WSAEHOSTDOWN ||
					nErrorCode == WSAEHOSTUNREACH ||
					nErrorCode == WSAEDISCON ||
					nErrorCode == WSA_OPERATION_ABORTED ||

					nErrorCode == ERROR_SEM_TIMEOUT ||
					nErrorCode == ERROR_NETNAME_DELETED ||
					nErrorCode == ERROR_CONNECTION_ABORTED ||
					nErrorCode == ERROR_OPERATION_ABORTED ||

					nErrorCode == ERROR_HOST_UNREACHABLE )	// IP 스푸핑? 패킷 손상? 뭐 여튼 가끔 발생하여 추가...;;
				{
					if( pOverlapped->cFlag == XIOCP_RECV || pOverlapped->cFlag == XIOCP_SEND || pOverlapped->cFlag == XIOCP_CONNECT )
					{
						onConnectionEvent( pTag, nThreadNum, pOverlapped, -1 );
						//_IOCPImpl::onDisconnecEvent( static_cast< XIOCPConnection * >( pOverlapped->pObj ) );
						// _oprint( "GQCS ERR : 0x%08X (%d)\n", static_cast< XIOCPConnection * >( pOverlapped->pObj ), pOverlapped->cFlag );
					}
					else if( pOverlapped->cFlag == XIOCP_DATAGRAMRECV || pOverlapped->cFlag == XIOCP_DATAGRAMSEND )
					{
						onDatagramEvent(pTag, nThreadNum, pOverlapped, 0);
					}
					else
						//XIOCP_ACCEPT관련 에러처리..
						if( pOverlapped->cFlag == XIOCP_ACCEPT )
					{
						// TODO : acceptex 다시 해주자
						onAcceptEvent( pTag, nThreadNum, pOverlapped, false );
						// 로그는 남기도록 한다.
					}

					continue;
				}

				if( pOverlapped->cFlag == XIOCP_ACCEPT )
				{
					// TODO : acceptex 다시 해주자
					onAcceptEvent( pTag, nThreadNum, pOverlapped );
					// 로그는 남기도록 한다.
				}
				else if( pOverlapped->cFlag == XIOCP_RECV || pOverlapped->cFlag == XIOCP_SEND || pOverlapped->cFlag == XIOCP_CONNECT )
				{
					onConnectionEvent( pTag, nThreadNum, pOverlapped, -1 );
					//_IOCPImpl::onDisconnecEvent( static_cast< XIOCPConnection * >( pOverlapped->pObj ) );
					// _oprint( "GQCS ERR : 0x%08X (%d)\n", static_cast< XIOCPConnection * >( pOverlapped->pObj ), pOverlapped->cFlag );
				}
				//else if( pOverlapped->cFlag == XIOCP_DATAGRAMRECV || pOverlapped->cFlag == XIOCP_DATAGRAMSEND )
				//{
				//	onDatagramEvent(pTag, nThreadNum, pOverlapped, 0);
				//}
				// }

				// 기타에러임
				std::string strWinSockError = XNetworkUtil::GetWin32ErrorInfo( nErrorCode );
				// 개행 문자 제거
				if( strWinSockError.size() > 2 )
					strWinSockError.erase( strWinSockError.end() - 2, strWinSockError.end() );
				std::string strWindowsError = XNetworkUtil::GetWin32ErrorInfo( GetLastError() );
				// 개행 문자 제거
				if( strWindowsError.size() > 2 )
					strWindowsError.erase( strWindowsError.end() - 2, strWindowsError.end() );
				strWindowsError.erase( std::remove( strWindowsError.begin(), strWindowsError.end(), '\n' ), strWindowsError.end() );
				FileLogHandler::GetFileLogHandler()->LogStringEx( NULL, "IOCP", "ERROR :[%d] %s / %s (%d)\n", nErrorCode, strWindowsError.c_str(), strWinSockError.c_str(), pOverlapped->cFlag );
				_cprint( "IOCP ERROR :[%d] %s / %s (%d)\n", nErrorCode, strWindowsError.c_str(), strWinSockError.c_str(), pOverlapped->cFlag );
				//throw XException( strError );
				continue;
			}

			// 처리
			switch( pOverlapped->pObj->GetID() )
			{
			case XOBJ_ACCEPTOR:
				{
					onAcceptEvent( pTag, nThreadNum, pOverlapped );
					break;
				}

			case XOBJ_CONNECTION:
				{	
					onConnectionEvent( pTag, nThreadNum, pOverlapped, dwNumberOfBytes );
					break;       
				}
			case XOBJ_DATAGRAM:
				{
					onDatagramEvent( pTag, nThreadNum, pOverlapped, dwNumberOfBytes );
					break;
				}
			case XOBJ_NULL:
				break;

			default:
				throw XException( "IOCPWorker : INVALID OBJECT" );
				break;
			}
		}
		catch( std::exception & ex )
		{	
			assert( 0 );
			pTag->pReceiver->onError( ex.what() );
		}
		catch( const char* str )
		{
			assert( 0 );
			pTag->pReceiver->onError( str );
		}
		catch( ... )
		{
			assert( 0 );
			pTag->pReceiver->onError( "IOCP : Unknown\n" );
		}
	}

	// Active thread count 조정  ( 종료모드이므로 1 감소 )
	InterlockedDecrement( pTag->plActiveThreadCount );
	InterlockedDecrement( pTag->plCurrentThreadCount );	

	return 0L;
}






XIOCP::XIOCP( INetworkEventReceiver * pReceiver ) : INetworkEvent( pReceiver )
{
	m_nThreadNum = 0;
	m_lActiveThreadCount = 0;
	m_lInstructionCount = 0;
	m_lCurrentThreadCount = 0;
	m_bIsPaused = false;

	m_pTag = new IOCPTAG( pReceiver );
	m_pTag->pIOCPMgr = this;
	m_pTag->plActiveThreadCount = &m_lActiveThreadCount;
	m_pTag->plInstructionCount = &m_lInstructionCount;
	m_pTag->plCurrentThreadCount = &m_lCurrentThreadCount;
	m_pTag->pbIsPaused = &m_bIsPaused;

	ENV().Bind( "iocp.active", (int*)&m_lActiveThreadCount );
	ENV().Bind( "iocp.instruction", (int*)&m_lInstructionCount );

	ENV().Bind( "iocp.dis_count", (int*)&g_nDisconnectCnt );
}

XIOCP::~XIOCP()
{
	if( m_pTag->hIOCP ) DeInit();

	delete m_pTag;
}

bool XIOCP::Init()
{
	m_pTag->hIOCP = CreateIoCompletionPort( INVALID_HANDLE_VALUE, NULL, 0, 0 );

	if( m_pTag->hIOCP )
	{
		m_nThreadNum = 0;
		m_lCurrentThreadCount = 0;
		m_lActiveThreadCount = 0;
		m_lInstructionCount = 0;
		m_bIsPaused = false;

		return true;
	}

	XError::XSetLastError( "CreateIoCompletionPort() error" );

	return false;
}

bool XIOCP::DeInit()
{
	if( !m_pTag->hIOCP )
	{
		XError::XSetLastError( "invalid IOCP handle" );
		return false;
	}

	if( !CloseHandle( m_pTag->hIOCP ) )
	{
		throw XException( "Can't destroy IOCP handle" );
	}

	m_pTag->hIOCP = NULL;
	
	return true;
}

XOVERLAPPED * XIOCP::allocOverlapped()
{
	return m_pTag->pOverlappedAllocator->allocOverlapped();
}

void XIOCP::freeOverlapped( XOVERLAPPED * ptr )
{
	return m_pTag->pOverlappedAllocator->freeOverlapped( ptr );
}

struct OverlappedAllocator* XIOCP::getOverlappedAllocator()
{
	return m_pTag->pOverlappedAllocator;
}

void XIOCP::Pause()
{
	ENV().Set( "iocp.paused", 1 );
	m_bIsPaused = true;
}

void XIOCP::Resume()
{
	ENV().Remove( "iocp.paused" );
	m_bIsPaused = false;
}

bool XIOCP::AddObject( IBaseObject * pObj )
{
	if( !m_pTag->hIOCP ) return false;

	XIOCPAcceptor*		pAcceptor = NULL;
	XIOCPConnection*	pConnection = NULL;
	XIOCPDatagram*		pDatagram = NULL;
	
	HANDLE				hFileHandle;

	// 객채 얻어옴
	switch( pObj->GetID() )
	{
		case XOBJ_CONNECTION:
			pConnection = static_cast< XIOCPConnection * >( pObj );
			hFileHandle = reinterpret_cast< HANDLE >( pConnection->GetSocketHandle() );
			pConnection->m_hIOCP = m_pTag->hIOCP;
			break;
		case XOBJ_ACCEPTOR:
			pAcceptor	= static_cast< XIOCPAcceptor * >( pObj );
			hFileHandle = reinterpret_cast< HANDLE >( pAcceptor->GetSocketHandle() );
			break;
		case XOBJ_DATAGRAM:
			pDatagram = static_cast< XIOCPDatagram *>( pObj );
			hFileHandle = reinterpret_cast< HANDLE >( pDatagram->GetSocketHandle() );
			break;
		default:
			throw XException( "XIOCP::AddObject error #1" );
			break;
	}

	// 기본 IOCP 핸들에 등록
	if( CreateIoCompletionPort( hFileHandle,  m_pTag->hIOCP, (ULONG_PTR)hFileHandle, 0 ) == NULL )
	{
		return false;
	}
	
	// IO 미리 요청
	switch( pObj->GetID() )
	{
		case XOBJ_CONNECTION:
			// ReadFile
			if( pConnection->IsConnected() ) pConnection->pendRecvRequest();
			break;

		case XOBJ_ACCEPTOR:
			// AcceptEx 를 미리 요청해 놓는다.
			pAcceptor->pendAcceptRequest();
			break;
		
		case XOBJ_DATAGRAM:
			// RecvFrom을 미리 요청해 놓는다.
			if( pDatagram->IsOpened() ) pDatagram->PendRecvFromRequest();
			break;
	}

	return true;
}

bool XIOCP::DelObject( IBaseObject * pObj )
{
	//XIOCPAcceptor*		pAcceptor;
	//XIOCPConnection*	pConnection;
	//XIOCPDatagram*		pDatagram;
	//
	//HANDLE				hFileHandle;

	//switch( pObj->GetID() )
	//{
	//	case XOBJ_CONNECTION:
	//		pConnection = static_cast< XIOCPConnection * >( pObj );
	//		hFileHandle = reinterpret_cast< HANDLE >( pConnection->GetSocketHandle() );
	//		break;
	//	case XOBJ_ACCEPTOR:
	//		pAcceptor	= static_cast< XIOCPAcceptor * >( pObj );
	//		hFileHandle = reinterpret_cast< HANDLE >( pAcceptor->GetSocketHandle() );
	//		break;
	//	case XOBJ_DATAGRAM:
	//		pDatagram = static_cast< XIOCPDatagram *>( pObj );
	//		hFileHandle = reinterpret_cast< HANDLE >( pDatagram->GetSocketHandle() );
	//		break;
	//	default:
	//		throw XException( "XIOCP::DelObject error #1" );
	//		break;
	//}

	//::CloseHandle( hFileHandle );

	return true;
}

bool XIOCP::StartThreadPool( unsigned nThreadNum, void (*init_func)( int ), bool bRegisterMonitoring )
{
	if( !m_pTag->hIOCP )
	{
		XError::XSetLastError( "Invalid IOCP handle" );
		return false;
	}

	m_pTag->vThreadHandle.reserve( nThreadNum*2 );

	unsigned	nErrCnt = 0;
	unsigned 	dwThreadID;

	unsigned	i;

	s_pfInitFunc = init_func;

	for( i = 0; i < nThreadNum; i++ )
	{	
		InterlockedIncrement( &m_lCurrentThreadCount );

		unsigned 	hThread;
		if( (hThread = (unsigned)_beginthreadex( NULL, 0, IOCPWorkerThread, m_pTag, CREATE_SUSPENDED, &dwThreadID )) > 0 )
		{
			if( !hThread ) {
				nErrCnt++;
				if( nErrCnt < nThreadNum ) {
					i--;
					continue;
				}
				else break;
			}

			m_pTag->vThreadHandle.push_back( hThread );
			m_pTag->vThreadId.push_back( dwThreadID );

			if( bRegisterMonitoring )
			{
				char szThreadName[64];
				s_sprintf( szThreadName, sizeof( szThreadName ), "IOCPWorker_%d", i );
				XThreadMonitor::AddWatchingThread( dwThreadID, std::string( szThreadName ) );
			}
		}
	}

	if( i != nThreadNum )
	{
		XError::XSetLastError( "Can't create thread" );
		return false;
	}

	for( i = 0; i < nThreadNum; i++ )
	{
		ResumeThread( (HANDLE)m_pTag->vThreadHandle[i] );
	}

	m_pTag->bIsFinished = false;

	m_nThreadNum = nThreadNum;

	ENV().Bind( "iocp.total", (int*)&m_nThreadNum );

	return true;
}

bool XIOCP::IncreaseThreadPool( unsigned nThreadNum )
{
	if( !m_pTag->hIOCP )
	{
		XError::XSetLastError( "Invalid IOCP handle" );
		return false;
	}

	unsigned	nErrCnt = 0;
	unsigned 	dwThreadID;

	unsigned	i;

	nThreadNum += m_nThreadNum;

	for( i = m_nThreadNum; i < nThreadNum; i++ )
	{
		uintptr_t 	hThread;
		if( (hThread = _beginthreadex( NULL, 0, IOCPWorkerThread, m_pTag, CREATE_SUSPENDED, &dwThreadID )) > 0 )
		{
			if( !hThread ) {
				nErrCnt++;
				if( nErrCnt < nThreadNum ) {
					i--;
					continue;
				}
				else break;
			}

			m_pTag->vThreadHandle.push_back( hThread );
			m_pTag->vThreadId.push_back( dwThreadID );
		}
	}

	if( i != nThreadNum )
	{
		XError::XSetLastError( "Can't create thread" );
		return false;
	}

	for( i = m_nThreadNum; i < nThreadNum; i++ )
	{
		ResumeThread( (HANDLE)m_pTag->vThreadHandle[i] );
	}

	return true;
}


bool XIOCP::EndThreadPool()
{
	if( !m_pTag->hIOCP ) return false;

		unsigned nErrCnt = 0;
	std::vector< uintptr_t >::iterator it;
	std::vector< uintptr_t > & vThreadHandle = m_pTag->vThreadHandle;

	m_pTag->bIsFinished = false;

	// { 모든 쓰레드에게 종료 시그널(XIOCP_EVENT_STOP)을 전송한다
	for( it = vThreadHandle.begin() ; it != vThreadHandle.end(); ++it )
	{
		 if( !PostQueuedCompletionStatus( m_pTag->hIOCP, 0, static_cast< ULONG_PTR >( XIOCP_EVENT_STOP ), NULL ) ) nErrCnt++;
	}
	if( nErrCnt )
	{
		return false;
	}
	// }


	// { 모든 쓰레드가 마칠때까지 대기한다
	while( m_lCurrentThreadCount ){
		Sleep( 100 );
	}


	for( it = vThreadHandle.begin() ;
		 it != vThreadHandle.end(); ++it ) {

		 CloseHandle( (HANDLE)*it );
	}

	// 쓰래드 핸들 벡터 초기화
	vThreadHandle.clear();
	
	return true;
}

struct _OverlappedAllocatorData
{
	_OverlappedAllocatorData() : overlappedHeap( sizeof(XOVERLAPPED), 512, 8 )	{}

	XMemoryPool											overlappedHeap;
	XSpinLock											heapLock;
};

OverlappedAllocator::OverlappedAllocator()
{
	if( ENV().GetInt( "iocp.useheap", 0 ) == 0 )
	{
		m_bUseHeap = false;
	}
	else
	{
		m_bUseHeap = true;
	}

	m_pData = new _OverlappedAllocatorData;
}

OverlappedAllocator::~OverlappedAllocator()
{
	delete m_pData;
}

XOVERLAPPED * OverlappedAllocator::allocOverlapped()
{	
	return static_cast< XOVERLAPPED * >( m_pData->overlappedHeap.Alloc() );
}

void OverlappedAllocator::freeOverlapped( XOVERLAPPED * ptr )
{
	if( !HasOverlappedIoCompleted( ptr ) ) 
	{
		//assert( 0 );
	}

	m_pData->overlappedHeap.Free( ptr ); 
}

void OverlappedAllocator::Lock()
{
	m_pData->heapLock.Lock();
}

void OverlappedAllocator::UnLock()
{
	return m_pData->heapLock.UnLock();
}

bool OverlappedAllocator::IsLocked() const
{
	return m_pData->heapLock.IsLocked();
}