IO Completion Port 작성하기 ( 2010.10.06 )

IO Completion Port 작성 하기

1. MFC 프로젝트 - 다이얼로그 방식

프로젝트명 : IOCompletionPort 

생성 : 
자동 ( IOCompletionPortDlg.h / IOCompletionPortDlg.cpp / class CIOCompletionPort() {} /
        로 생김(자동으로 C가 붙음) )


2. 다이얼로그에  ListBox Control 추가

---- 흠

3. 버튼을 옮겨서 : 서버 시작 버튼을 만들고.....클릭시 함수처리부분에서 호출 및 사용될  class 정의


4.  class 정의 :  cIOCompletionPort

5. 클래스 마법사에서 ListBox 멤버 변수 작성 - IOCompletionPortDlg.cpp
6. ListBox에 문자 출력하는 함수 작성 - IOCompletionPortDlg.cpp
IOCompletionPortDlg.h

// 흠 흠 흠
#include "afxwin.h"
#include "resource.h"

#include "cIOCompletionPort.h"

#define	LISTEN_PORT		8000

///..........

public:
	void OutputMsg(char * szOutputString,...);
	CListBox m_ctOutput;
	boolean		m_bServerStarted;
	cIOCompletionPort m_IOCompletionPort;

	afx_msg void OnBnClickedStartserver();
	afx_msg void OnBnClickedCancel();

IOCompletionPortDlg.cpp

/// 사용자 추가
void CIOCompletionPortDlg::OutputMsg(char * szOutputString,...)
{

	char szOutStr[1024];
	va_list argptr;

	va_start(argptr, szOutputString) ;
	vsprintf( szOutStr, szOutputString, argptr );
	va_end(argptr);

	
	//  CListBox에 추가

	// 속성 - > 구성속성->일반 -> 멀티 바이트 문자 조합 MBCS
	//m_ctOutput.SetCurSel( m_ctOutput.AddString( szOutStr ) );

	m_ctOutput.InsertString(0, szOutStr) ;

}



void CIOCompletionPortDlg::OnBnClickedStartserver()
{
	// TODO: 여기에 컨트롤 알림 처리기 코드를 추가합니다.

		if ( ! m_bServerStarted  ) {
			OutputMsg("0.----서버 ");

			m_IOCompletionPort.SetMainDlg(this);
			bool bRet = m_IOCompletionPort.InitSocket();

			OutputMsg("1.----포트:%d", LISTEN_PORT );
			m_IOCompletionPort.BindandListen(LISTEN_PORT);

			OutputMsg("2.----가동 " );
			m_IOCompletionPort.StartServer();

			m_bServerStarted=true;
		} else {
			OutputMsg("****************서버 가동중 : 누르지 마세요 ******************");
		}

}


void CIOCompletionPortDlg::OnBnClickedCancel()
{
	// TODO: 여기에 컨트롤 알림 처리기 코드를 추가합니다.

	m_IOCompletionPort.DestroyThread();

	CDialogEx::OnCancel();
}

cIOCompletionPort.h

// 1. 패킷 사이즈
#define MAX_SOCKBUF		1024
// 2. 클라이언트 수
#define		MAX_CLIENT	1024
// 3. 쓰레스 수
#define		MAX_WORKERTHREAD	4


enum	enumOperation {
	OP_RECV,
	OP_SEND
};

/// WSAOVERLAPPED 구조체를 확장시켜 필요한 정보 추가
struct stOverlappedEx {
	WSAOVERLAPPED		m_wsaOverlapped;	// Overlapped I/O 구조체
	SOCKET				m_socketClient;		// client socket
	WSABUF				m_wsaBuf;			// Overlapped I/O 작업 버퍼
	char				m_szBuf[MAX_SOCKBUF];	// 데이타 버퍼
	enumOperation		m_eOperation;			// 작업 동작 종류
};
/// client 정보를 담기 위한 구조체
struct	stClientInfo {
	SOCKET				m_socketClient;		// 클라이언트와 연결되는 소켓
	stOverlappedEx		m_stRecvOverlappedEx;	// Recv Overlapped I/O 작업을 위한 변수
	stOverlappedEx		m_stSendOverlappedEx;	// Send Overlapped I/O 작업을 위한 변수

	/// 생성자에서 멤버 변수들을 초기화
	stClientInfo() 
	{
		m_socketClient = INVALID_SOCKET;
		ZeroMemory( &m_stRecvOverlappedEx, sizeof(m_stRecvOverlappedEx) );
		ZeroMemory( &m_stSendOverlappedEx, sizeof(m_stSendOverlappedEx) );
	}
};


class CIOCompletionPortDlg;

class cIOCompletionPort
{
public:
	cIOCompletionPort(void);
	~cIOCompletionPort(void);
	bool InitSocket(void);

	bool BindandListen(int nPort);
	bool StartServer(void);
	bool CreateWorkerThread(void);
	bool CreateAccepterThread(void);
	stClientInfo * GetEmptyClientInfo(void);
	bool BindIOCompletionPort(stClientInfo * pClientInfo);

	bool BindRecv(stClientInfo * pClientInfo);
	bool SendMsg(stClientInfo * pClientInfo, char * pMsg, int nLen);
	void WorkerThread(void);
	void AccepterThread(void);
	void SetMainDlg(CIOCompletionPortDlg * pMainDlg);
	void DestroyThread(void);
	void CloseSocket(stClientInfo * pClientInfo, bool bIsForce=false);
private:
	// 1. 클라이언트 정보 저장 구조체
	stClientInfo *	m_pClientInfo;

	// 2. 클라이언트 접속을 받기위한 리슨 소켓
	SOCKET			m_socketListen;

	// 3. 접속 되어 있는 클라이언트 수
	int				m_nClientCnt;

	// 4. 메인 윈도우 포인터
	CIOCompletionPortDlg *		m_pMainDlg;

	// 5. 작업 스레드 핸들
	HANDLE			m_hWorkerThread[MAX_WORKERTHREAD];

	// 6. 접속 스레드 핸들
	HANDLE			m_hAccepterThread;

	// 7. CompletionPort 객체 핸들
	HANDLE			m_hIOCP;

	// 8. 작업 스레드 동작 플래그
	bool			m_bWorkerRun;

	// 9. 접속 스레드 동작 플래그
	bool			m_bAccepterRun;

	// 10. 소켓 버퍼
	char			m_szBuf[1024];

};

cIOCompletionPort.cpp

#include "StdAfx.h"

#include "cIOCompletionPort.h"

//// 흠
#include "IOCompletionPortDlg.h"

/// 쓰레드 만들기 
/// WSARecv , WSASend의 Overlapped I/O 작업을 위한 
unsigned int WINAPI	CallWorkerThread(LPVOID p)
{
	cIOCompletionPort * pOverlappedEvent = (cIOCompletionPort *)p;

	pOverlappedEvent->WorkerThread();

	return 0;
}
unsigned int WINAPI	CallAccepterThread(LPVOID p)
{
	cIOCompletionPort * pOverlappedEvent = (cIOCompletionPort *)p;

	pOverlappedEvent->AccepterThread();

	return 0;
}
/// 
cIOCompletionPort::cIOCompletionPort(void)
{
	/// 모든 멤버 변수들의 초기화
	m_pMainDlg = NULL;
	m_bWorkerRun	= true;
	m_bAccepterRun	= true;
	m_nClientCnt = 0;
	m_hAccepterThread = NULL;
	m_hIOCP	= NULL;
	m_socketListen = INVALID_SOCKET;
	ZeroMemory(m_szBuf, 1024);
	for ( int i=0; i < MAX_WORKERTHREAD; i++ ) {
		m_hWorkerThread [i] = NULL;
	}
	m_pClientInfo = new stClientInfo[MAX_CLIENT];
}


cIOCompletionPort::~cIOCompletionPort(void)
{
	// 윈속 사용 해재
	WSACleanup();
	// 다 사용한 객체 삭제
	if ( m_pClientInfo ) 
	{
		delete[] m_pClientInfo;
		m_pClientInfo = NULL;
	}
}


bool cIOCompletionPort::InitSocket(void)
{
	WSADATA	wsaData;

	// 윈속 버젼 2.2
	int nRet = WSAStartup( MAKEWORD(2,2) , &wsaData);
	if ( 0 != nRet ) {

		m_pMainDlg->OutputMsg("[에러]WSAStartup() 실패:%d:",WSAGetLastError() );

		return false;
	}
	m_socketListen = WSASocket(AF_INET, SOCK_STREAM,
					IPPROTO_TCP, NULL, NULL, WSA_FLAG_OVERLAPPED );

	if ( INVALID_SOCKET == m_socketListen ) {

		m_pMainDlg->OutputMsg("[에러]WSASocket() 실패:%d:",WSAGetLastError() );

		return false;

	}
	m_pMainDlg->OutputMsg(" InitSocket() 성공" );
	///
	return true;
}




bool cIOCompletionPort::BindandListen(int nPort)
{

	SOCKADDR_IN		stServerAddr;

	stServerAddr.sin_family = AF_INET;

	stServerAddr.sin_port = htons(nPort);

	stServerAddr.sin_addr.S_un.S_addr = htonl(INADDR_ANY);

	int nRet = bind( m_socketListen, (SOCKADDR *)&stServerAddr, sizeof(SOCKADDR_IN) );
	if ( 0 != nRet ) {

		m_pMainDlg->OutputMsg("[에러] bind() 실패:%d:",WSAGetLastError() );

		return false;

	}
	nRet = listen( m_socketListen, 5 ) ;
	if ( 0 != nRet ) {

		m_pMainDlg->OutputMsg("[에러] listen() 실패:%d:",WSAGetLastError() );

		return false;

	}
	m_pMainDlg->OutputMsg(" BindandListen() 성공" );
	///
	return true;
}





bool cIOCompletionPort::CreateWorkerThread(void)
{

	unsigned int uiThreadId = 0;

	/// Waiting Thread Queue에 대기 상태로 넣을 쓰레드들 생성
	/// 권장하는 개수 : cpu *2 +1

	for ( int i =0; i < MAX_WORKERTHREAD; i++ ) 
	{

		m_hWorkerThread[i] = (HANDLE)_beginthreadex(NULL, 0,
							&CallWorkerThread,
							this,
							CREATE_SUSPENDED,
							&uiThreadId);
		if ( m_hWorkerThread[i] == NULL ) 
		{
			m_pMainDlg->OutputMsg("[에러] CreateWorkerThread() 실패:%d:",GetLastError() );

			return false;
		}
		ResumeThread( m_hWorkerThread[i] );
	}


	m_pMainDlg->OutputMsg(" CreateWorkerThread() 성공" );
	///
	return true;
}


bool cIOCompletionPort::CreateAccepterThread(void)
{

	unsigned int uiThreadId = 0;

	/// 클라이언트 접속 요청을 받은 쓰레드 생성
	m_hAccepterThread = (HANDLE)_beginthreadex(NULL, 0,
						&CallAccepterThread,
						this,
						CREATE_SUSPENDED,
						&uiThreadId);
	if ( m_hAccepterThread == NULL ) 
	{
		m_pMainDlg->OutputMsg("[에러] CreateAccepterThread() 실패:%d:",GetLastError() );

		return false;
	}
	ResumeThread( m_hAccepterThread );


	m_pMainDlg->OutputMsg(" CreateAccepterThread() 성공" );
	///
	return true;
}
////
bool cIOCompletionPort::BindIOCompletionPort(stClientInfo * pClientInfo)
{

	HANDLE hIOCP;

	/// socket 과 pClientInfo를  CompletionPort객체와 연결 시킨다.
	hIOCP = CreateIoCompletionPort( (HANDLE)pClientInfo->m_socketClient,
			m_hIOCP,
			reinterpret_cast( pClientInfo ),
			0);
	if ( NULL == hIOCP || m_hIOCP != hIOCP ) 
	{
		m_pMainDlg->OutputMsg("[에러] CreateIoCompletionPort() 실패:%d:",GetLastError() );

		return false;
	}


	m_pMainDlg->OutputMsg(" BindIOCompletionPort() 성공" );
	///
	return true;
}

bool cIOCompletionPort::StartServer(void)
{


	m_hIOCP = CreateIoCompletionPort( INVALID_HANDLE_VALUE,
								NULL,
								NULL,
								0);
	if ( NULL == m_hIOCP ) 
	{
		m_pMainDlg->OutputMsg("[에러] CreateIoCompletionPort() 실패:%d:",GetLastError() );

		return false;
	}

	
	bool bRet = CreateWorkerThread();
	if ( false == bRet ) 
	{
		m_pMainDlg->OutputMsg("[에러] CreateWorkerThread() 실패:%d:",GetLastError() );

		return false;
	}

	bRet = CreateAccepterThread();
	if ( false == bRet ) 
	{
		m_pMainDlg->OutputMsg("[에러] CreateAccepterThread() 실패:%d:",GetLastError() );

		return false;
	}

	m_pMainDlg->OutputMsg(" StartServer() 성공" );
	///
	return true;
}

	
bool cIOCompletionPort::BindRecv(stClientInfo * pClientInfo)
{
	DWORD	dwFlag = 0;
	DWORD	dwRecvNumBytes = 0;


	// Overlapped I/O Setting
	pClientInfo->m_stRecvOverlappedEx.m_wsaBuf.len = MAX_SOCKBUF;
	pClientInfo->m_stRecvOverlappedEx.m_wsaBuf.buf = 
		pClientInfo->m_stRecvOverlappedEx.m_szBuf;
	pClientInfo->m_stRecvOverlappedEx.m_eOperation = OP_RECV;

	
	//// 입력 버퍼 클리어 ???????????????
	//ZeroMemory(pClientInfo->m_stRecvOverlappedEx.m_szBuf, 1024);

	int nRet = WSARecv( pClientInfo->m_socketClient,
						&(pClientInfo->m_stRecvOverlappedEx.m_wsaBuf),
						1,
						&dwRecvNumBytes,
						&dwFlag,
						(LPWSAOVERLAPPED)&(pClientInfo->m_stRecvOverlappedEx),
						NULL);
	/// socket_error 이면 client socket이 끊어 진걸로 처리한다.
	if ( nRet ==  SOCKET_ERROR && ( ERROR_IO_PENDING != WSAGetLastError()  )  ) 
	{
		m_pMainDlg->OutputMsg("[에러] BindRecv WSARecv() 실패 WSAGetLastError:%d:",WSAGetLastError() );

		return false;
	}
	m_pMainDlg->OutputMsg("[알림] BindRecv WSARecv() 성공");
	return true;
}


bool cIOCompletionPort::SendMsg(stClientInfo * pClientInfo, char * pMsg, int nLen)
{
	DWORD	dwRecvNumBytes = 0;

	//전송될 메시지를 복사
	CopyMemory( pClientInfo->m_stSendOverlappedEx.m_szBuf, pMsg, nLen );


	// Overlapped I/O Setting 정보
	pClientInfo->m_stSendOverlappedEx.m_wsaBuf.len = nLen;

	pClientInfo->m_stSendOverlappedEx.m_wsaBuf.buf =
							pClientInfo->m_stSendOverlappedEx.m_szBuf;

	pClientInfo->m_stSendOverlappedEx.m_eOperation = OP_SEND;

	int nRet = WSASend( pClientInfo->m_socketClient, 
						&(pClientInfo->m_stSendOverlappedEx.m_wsaBuf),
						1,
						&dwRecvNumBytes,
						0,
						(LPWSAOVERLAPPED)&(pClientInfo->m_stSendOverlappedEx),
						NULL);
	if ( nRet == SOCKET_ERROR ) {
		m_pMainDlg->OutputMsg("[에러] SendMsg WSASend() nRet:%s:","SOCKET_ERROR" );
	}
	/// socket_error 이면 client socket이 끊어 진걸로 처리한다.
	if ( nRet ==  SOCKET_ERROR && ( WSAGetLastError() != ERROR_IO_PENDING )  ) 
	{
		m_pMainDlg->OutputMsg("[에러] SendMsg WSASend() 실패 WSAGetLastError:%d:",WSAGetLastError() );

		return false;
	}
			
	m_pMainDlg->OutputMsg("[알림] SendMsg WSASend() 성공");

	return true;
}
// 할당
stClientInfo * cIOCompletionPort::GetEmptyClientInfo(void)
{
	for(int i = 0; i < MAX_CLIENT; i++ ) 
	{
		if(INVALID_SOCKET == m_pClientInfo[i].m_socketClient) 
		{
			return &m_pClientInfo[i];
		}
	}
	return NULL;
}



// 사용자 접속 받는 쓰레드
void cIOCompletionPort::AccepterThread(void)
{
	SOCKADDR_IN		stClientAddr;

	int nAddrLen = sizeof(SOCKADDR_IN);
	while ( m_bAccepterRun)
	{
		// 접속 받을 구조체 인덱스 얻기
		stClientInfo * pClientInfo = GetEmptyClientInfo();
		if ( NULL == pClientInfo )
		{
			m_pMainDlg->OutputMsg("[에러] NULL == pClientInfo :%s:","Client FULL");

			return ;
		}
		// 클라이언트 접속 요청까지 대기
		pClientInfo->m_socketClient = accept ( m_socketListen,
							(SOCKADDR *)&stClientAddr, &nAddrLen );
		if ( INVALID_SOCKET == pClientInfo->m_socketClient ) {
			continue;
		}
		// I/O Completion Port객체와 소켓을 연결 시킨다.
		bool bRet = BindIOCompletionPort( pClientInfo );
		if ( false == bRet ) {
			return;
		}
		// Recv Overlapped I/O 작업을 요청한다
		bRet = BindRecv(pClientInfo);
		if ( false == bRet ) {
			return;
		}

		m_pMainDlg->OutputMsg("[클라이언트 접속] ip(%s) SOCKET(%d)",
			inet_ntoa( stClientAddr.sin_addr) ,
			pClientInfo->m_socketClient);

		m_nClientCnt ++;

	}
}


void cIOCompletionPort::WorkerThread(void)
{
	// CompletionKey를 받을 포인터 변수
	stClientInfo * pClientInfo = NULL;

	// 함수 호출 성공여부
	BOOL bSuccess = TRUE;

	// Overlapped I/O작업에서 전송된 데이타 크기
	DWORD dwIoSize = 0;

	// I/O 작업을 위해 요청한 Overlapped 구조체를 받을 포인터
	LPOVERLAPPED lpOverlapped = NULL;

	while ( m_bWorkerRun ) 
	{
		/**
		이 함수로 인해 쓰래들들은 WaitingThread Queue에 대기상태로 들어간다
		완료된 Overlapped I/O 작업이 발생하면 IOCP Queue에서 완료된 작업을 가져와 뒤처리
		그리고 PostQueuedCompletionStatus()함수에 의해 사용자 메시지가 도착되면 쓰레드 종료
		**/
		bSuccess = GetQueuedCompletionStatus( m_hIOCP,
				&dwIoSize,							// 실제 전송된 바이트
				(LPDWORD)&pClientInfo,				// Completionkey
				&lpOverlapped,						// Overlappped I/O 객체
				INFINITE);							// 대기할 시간(무한대기)
		
		// 클라이언트가 접속 끊었을 때
		//
		//  FALSE == bSuccess
		//
		if ( FALSE == bSuccess && 0 == dwIoSize ) 
		{
			m_pMainDlg->OutputMsg("[클라이언트] SOCKET(%d) 접속 끊김",	pClientInfo->m_socketClient);
			CloseSocket(pClientInfo);
			continue;
		}

		// 사용자 스레드 종료 메시지 처리
		//
		//  TRUE == bSuccess
		//

		if ( TRUE == bSuccess && 0 == dwIoSize && NULL == lpOverlapped )
		{
			//
			// WorkerThread 종료
			//
			m_bWorkerRun = false ;
			continue;
		}

		if ( NULL == lpOverlapped ) {
			continue;
		}
		stOverlappedEx * pOverlappedEx =(stOverlappedEx *)lpOverlapped;

		// Overlapped I/O Recv 작업 결과 뒤 처리
		// 
		//	OP_RECV
		//
		if ( OP_RECV == pOverlappedEx->m_eOperation ) 
		{
			pOverlappedEx->m_szBuf[dwIoSize] = NULL;
			m_pMainDlg->OutputMsg("[수신] ( %d ) bytes , msg : %s ",dwIoSize,pOverlappedEx->m_szBuf);

			// 클라이언트에 메시지를 에코한다.
			//BindRecv( pClientInfo );

			//SendMsg(pClientInfo, pOverlappedEx->m_szBuf, dwIoSize );

			//pOverlappedEx->m_eOperation = OP_SEND;
			//BindRecv( pClientInfo );

			SendMsg( pClientInfo, pOverlappedEx->m_szBuf, dwIoSize);

		}
		// Overlapped I/O Send 작업 결과 뒤 처리
		// 
		//	OP_SEND
		//
		else if ( OP_SEND == pOverlappedEx->m_eOperation ) 
		{
			m_pMainDlg->OutputMsg("[송신] ( %d ) bytes , msg : %s ",dwIoSize,pOverlappedEx->m_szBuf);
			
			//// 입력 버퍼 클리어 ???????????????
			ZeroMemory(pOverlappedEx->m_szBuf, 1024);

			BindRecv( pClientInfo );

		}
		else 
		{
			m_pMainDlg->OutputMsg("[클라이언트] SOCKET(%d) 예외 상황 ",pClientInfo->m_socketClient);
		}
		lpOverlapped = NULL;
	}
}

void cIOCompletionPort::SetMainDlg(CIOCompletionPortDlg * pMainDlg)
{
	/// .h에서 .cpp로 옮겨 놓았는데? 문제 없나?
	m_pMainDlg = pMainDlg;
}


void cIOCompletionPort::DestroyThread(void)
{
	for(int i=0; i< MAX_WORKERTHREAD; i++)
	{
		// WaitingThreadQueue에서 대기중인 쓰레드에 사용자 종료 메시지 보내기
		PostQueuedCompletionStatus( m_hIOCP,0,0,NULL);
	}
	for(int i=0; i< MAX_WORKERTHREAD; i++)
	{
		CloseHandle( m_hWorkerThread[i] );
		WaitForSingleObject ( m_hWorkerThread[i], INFINITE );
	}

	m_bAccepterRun = false;
	// Accepter Thread 종료
	closesocket( m_socketListen );
	// Thread 종료
	WaitForSingleObject( m_hAccepterThread, INFINITE );


}


void cIOCompletionPort::CloseSocket(stClientInfo * pClientInfo, bool bIsForce)
{
	struct linger stLinger = {0,0};

	if ( true ) {
		// timeout=0으로 설정되어 강제 종료. 주의 : 데이타 손실 가능성
		// right now !!!
		stLinger.l_onoff = 1;
	}
	// 데이타 송수신 모두 중단
	shutdown( pClientInfo->m_socketClient, SD_BOTH );

	// 소켓 옵션
	setsockopt( pClientInfo->m_socketClient, SOL_SOCKET, SO_LINGER,
			(char *)&stLinger, sizeof(stLinger) );
	// 소켓 연결 종료
	closesocket(pClientInfo->m_socketClient);
	pClientInfo->m_socketClient = INVALID_SOCKET;
}