[04주차] I/O 모델

 

 

한국공학대학교 정내훈 교수님 4-1 게임 서버 프로그래밍 수업 중 일부입니다.

 

---

 

 

- 게임 프로그램 특징

: 게임의 진행 속도는 CPU의 속도가 아니라 실제 시간.

: REAL TIME → 내가 명령을 입력할 때까지 기다리지 않음!!!! (적어도 렌더링은 계속 됨)

: INDEPENDENCE → 모든 객체가 독립적 행동 (게임 진행 정해져 있지 않음 + 플레이어 행동 안 기다림, non-blocking)

 

 

 

- Non-blocking I/O

 : 기존 Blocking 소켓은 I/O 작업을 수행할 때 해당 작업이 완료될 때까지 대기

 : Non-blocking → 작업 수행 시 작업이 완료되지 않아도 즉시 반환

 : WSARecv() 호출을 즉시 완료하지 못하는 경우

  1) 도착한 데이터 없으면,

  2) WSAEWOULDBLOCK 에러 내고 리턴

  3) 기다리지 않음

 : 만약 n:1 온라인 게임을 구현한다면? → 서버 문제 발생 (많은 클라이언트로 인한 잦은 WSARecv 실패로 성능 저하) : recv 뿐만 아니라 accept도 고려해야 함 + 멀티쓰레드 고려 필요

 

// server main loop (Non-blocking I/O)

while(true){
	// Process Client Packet
    ret = recv(client_A, buf);
    if (SUCCESS == ret) process_packet(client_A, buf);
    ret = recv(client_B, buf);
    if (SUCCESS == ret) process_packet(client_B, buf);
    
    //update world
    // do npc ai, do heal......
   }
   
   
   >> 실패한 RECV == CPU 낭비 !!!

 

 

 

- 네트워크 I/O 모델

 : 게임 서버에서의 다중 접속 → 정해지지 않은 순서, 수 천 개의 접속, 상대적 낮은 접속 당 대역폭 (bandwidth)

 : 블럭킹 방지와 비 규칙적인 입출력의 효율적 관리, 대규모 다중 접속 관리로 인해 네트워크 I/O 모델이 필요

  1) Non-blocking I/O

  2) Socket Thread

  3) Select

  4) WSAAsyncSelect

  5) WSAEventSelect

  6) Overlapped I/O(Event)

  7) Overlapped I/O(Callback)

  8) I/O Completion Port

 

 

 

-  Non-blocking I/O

// Server main loop

while (true) {
	for (SOCKET s : sockets){
    	recv(s, ...);          // 모든 패킷에 대한 처리
        if (success) 
        	/*패킷 처리*/;
    }
	// 다른 작업
}

 

 : 단점 → Busy Waiting ) 모든 소켓의 recv를 돌아가며 반복 호출해야 하며 많은 recv가 실패함

 : 잦은 시스템 call → CPU 낭비 → 성능저하

 

 

 

- Socket Thread

 : Thread를 통한 처리, 1:1 게임서버가 여러 개 있는 것과 비슷 (Blocking : 절대 실패하지 않는 대신 멈춘다)

while (!shutdown){
	new_sock = accept(sock, &addr, &len);
    thread t = thread { do_io, new_sock };
}

void do_io(mysock){
	while(true){
    	recv(mysock);
        process_packet();}}

 

 : 다중 소켓 처리 ok, non-blocking 불필요 → 효율적 API 호출

 : 과도한 thread 개수로 인한 OS Overhead 

  - context switching : 쓰레드 간 전환에서 OS는 현재 실행 중인 쓰레드 상태를 저장 & 새로운 쓰레드를 로드하는 과정 수행

         → 쓰레드 개수가 많아질수록 이 작업이 빈번하게 발생

  - Thread Control Block(TCB) : 여기엔 쓰레드의 상태, 우선 순위, 레지스터 값 등의 정보가 포함하는데 이게 많은 수록 관리하는 데 필요한 자원이 많아짐

  - Stack Memory : 각 쓰레드는 자체 스택을 가짐, 이는 쓰레드가 함수 호출 시 지역 변수 및 함수 호출 정보를 저장하는 데 사용

          → 과도한 쓰레드 개수로 인해 메모리 소비가 늘어나고 스택 공간을 관리하는데 추가적인 오버헤드가 발생

 : Coroutine을 사용한 처리 → 프로그래밍 방법은 thread - IO와 동일 + 오버헤드 전부 해결

 : C++ 23에서 제대로 구현 예정이라고 함 → 현재 참고 https://tango1202.github.io/cpp/modern-cpp-coroutine/

#28. [모던 C++] 코루틴(coroutine)(C++20)

 

 

 

- Select

 : 여러 개의 소켓 데이터 도착 여부를 하나의 API로 검사 (효율적 API 호출)

 : 단점 → socket 개수의 한계 존재 (unix 64, linux 1024), socket 개수가 많아질 수록 성능 저하 증가

// Select
int select( __in int nfds, __inout fd_set* readfds, __inout fd_set* writefds,
		__inout fd_set* exceptfds, __in const struct timeval* timeout);
        
// nfds : 검사하는 소켓 개수 (window에선 무시)
// readfds : 읽기 가능 검사용 소켓 집합 포인터
// writefds : 쓰기 가능 검사용 소켓 집합 포인터
// exceptfds : 에러 검사용 소켓 집합 포인터
// timeout : select가 기다리는 최대 시간
// return value : 사용 가능한 소켓 개수


// 실제 코딩
FD_SET(sock1, &rfds);
FD_SET(sock2, &rfds);
select(0, &rfds, NULL, NULL, &time);

if (FD_ISSET(sock1, &rfds)) recv(sock1, buf, len, 0)
if (FD_ISSET(sock2, &rfds)) recv(sock2, buf, len, 0


// server
while (true) {
 	select(&all_sockets);
 	for (i : all_sockets)
		if (IS_READY(i)) {
 			recv(socket[i], buf, …);
			 process_packet(i, buf); 
 		}
	 다른 작업;
}

 

 

 

- WSAAsyncSelect

 : 소켓 이벤트를 특정 윈도우의 메시지로 받음

int WSAAsyncSelect( 
	__in SOCKET s,  // 소켓
	__in HWND hWnd,  // 메시지를 받을 윈도우 
	__in unsigned int wMsg,  // 메시지 번호
	__in long lEvent );  // 반응 event 선택

 

 : 클라이언트에 많이 쓰임, 윈도우 필요 / 윈도우 메세지 큐  사용 -> 성능 느림

 : 자동으로 non-blocking mode로 소켓 전환

 

 

 

- WSAEventSelect

 : 메시지를 처리할 윈도우가 필요 없음

 : WSAEVENT - 운영체제가 관리하는 플랙, true가 될 때까지 실행을 멈출 수 있음 (CPU 낭비 없이)

 : socket과 event의 array를 만들어서 WSAWaitForMultipleEvents() 리턴값에서 socket 추출 (소켓 수 64개 제한)

 

int WSAEventSelect( 
	__in SOCKET s, 
	__in WSAEVENT hEventObject, 
	__in long lNetworkEvents );

// API 대기 상대 검출
DWORD WSAWaitForMultipleEvents(
	DWORD nCount,
	const WSAEVENT *lphEvents,
	BOOL bWaitAll,
	DWORD dwMiliseconds);
    
int WSAEnumNetworkEvents(SOCKET s,
	WSAEVENT hEventObject;,
	LPWSANETWORKEVENTS lpNetworkEvents);

lpNetworkEvents->lNetworkEvents;

 

 

: 동작

정내훈 교수님 강의자료

 

 

- Overlapped I/O

 = Asynchronous I/O or 비동기 I/O

 : IOCP도 이를 이용함

 : 사용 방법이 select style의 I/O 모델과 다르다

  - I/O 요청을 먼저 하고 종료를 나중에 확인 → 요청 후 즉시 리턴 & Non-blocking과 다르게 거의 실패 X

  - I/O 요청 후 기다리지 않고 다른 일 가능, 여러 개의 I/O 요청을 동시에 가능

// Select의 while 부분과 비교해보면 알 것

for (sock : all_sockets)
	recv(sock, buf[sock]);       // 먼저 다 받음
while (true) {
	sock = wait_for_completed_recv();
	process_pakcet(sock, buf[sock]);
	recv(sock, buf[sock], ...);
}

 

 

 : Select I/O 모델들은 recv/send의 가능 여부만을 검사 후 I/O 진행 한다면, 이는 I/O 요청 후 실패 완료 통보

 : (non-blocking 차이점) I/O 호출과 완료의 분리 (Overlapped IO는 요청이 커널에 남아서 수행됨. 버퍼가 커널에 등록됨, 실패
없음)

 

 : send와 recv 호출 시 패킷 송수신의 완료응 기다리지 않고 종료함 ( 송수신의 시작만을 지시 )

 : 동시다발적으로 recv, sdnd 가능하나, 하나의 socket은 하나의 recv만 가능

 

 : OS 입장

 - 동기식 : 데이터 도착시 버퍼에 저장해 두고 Recv 요청이 올 때까지 기다림, 오면 복사해서 보냄

 - 비동기식 : 데이터 도착시 저장된 Recv 요청이 있다면, 요청된 버퍼에 데이터를 저장하고 즉시 완료함

 

SOCKET WSASocket(int af, int type, int protocol,
				LPWSAPROTOCOL_INFO lpProtocolInfo, GROUP g, DWORD dwFlags)
                
// af : address family 
// type : socket type
// protocol : 사용할 프로토콜 종류
// lpProtocolInfo : 프로토콜 정보 (NULL)
// g : 예약
// dwFlags : WSA_FLAG_OVERLAPPED   -> 이렇게 하고 언오버랩 써도 잘돌아감


typedef struct WSAOVERLAPPED {
	DWORD Internal;
	DWORD InternalHigh;
	DWORD Offset;
	DWORD OffsetHigh;
	WSAEVENT hEvent;
} WSAOVERLAPPED, FAR *LPWSAOVERLAPPED;

// DWORD type 부분은 0으로 초기화 후 사용
// hEvent : I/O가 완료되었음을 알려주는 Eevent 핸들
// -> LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine (callback 함수, 뒤에서 설명)

// Overlapped I/O Event model
// LPWSAOVERLAPPED lpOverlapped 구조체의 WSAEVENT hEvent 사용

WSAWaitForMultipleEvents() // 완료 확인(recv했다, 몇바이트 읽었다를 알려주진 않음)
WSAGetOverlappedResult()  // 정보 얻기(recv완료가 되었다면, 성공/실패, 성공했다면 몇 바이트 했는지 알려줌)

BOOL WSAGetOverlappedResult(
	SOCKET s,
	LPWSAOVERLAPPED lpOverlapped,
	LPDWORD lpcbTransfer,
	BOOL fWait,
	LPDWORD lpdwFlags);
}

// s : socket
// lpOverlapped : WSARecv에 넣었던 구조체
// lpbTransfer : 전송된 데이터 크기
// fWait : 대기 여부
// lpdwFlags : Recv의 lpFlag 결과


// 1. WSACreateEvent()를 사용해서 이벤트 생성
// 2. WSAOVERLAPPED 구조체 변수선언, 0으로 초기화, hEvent에 1의 이벤트
// 3. WSASend(), WSARecv() -> 2의 구조체를 WSAOVERLAPPED에
//  – 중복 사용 불가능!! 호출 완료 후 재사용
//  - lpCompletionROUTINE -> NULL
// 4. WSAWaitForMultipleEvents()함수로 완료 이벤트 감지
// 5. WSAGetOverlappedResult()함수로 정보 획득


// 단점 : 시스템 호출이 빈번, 최대 동접 64 ==> callback mmodel

 

// Overlapped I/O Callback model
// 이벤트 제한 개수 없으며 사용 편리
// LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine 함수 사용
// Overlapped I/O가 끝난후 lpCompletionRoutine 호출

void CALLBACK CompletionROUTINE(
	DWORD dwError,
	DWORD cbTransferred,
	LPWSAOVERLAPPED lpOverlapped,
	DWORD dwFlags);
}

// dwError : 작업의 성공 유무
// cbTransferred : 전송된 바이트 수
// lpOverlapped : WSASend, WSARev에서 사용한 구조체
// dwflags : WSASend, WSARecv에서 사용한 flag

// 1) 누가/언제 Callback을 호출하는가?
//   OS에서 I/O 작업이 완료되면
//   해당 작업에 대한 완료를 알리기 위해 지정된 콜백 함수가 호출

// 2) 멀티쓰레드로 동시에 Callback이 실행되는가?
//    No!, 콜백 함수가 단일 쓰레드에서 순차적으로 실행

// 3) 운영체제가 실행중인 프로그램을 강제로 멈추고 Callback을 실행하는가?
//    No!, I/O 작업이 완료되면 해당 작업에 대한 완료를 알리기 위해 등록된 콜백 함수를 호출

// Callback을 사용한 프로그램의 구조
// 할 일이 있으면 하고, 없으면 SleepEx를 호출한다

 

 

 

 

- 채팅 서버 ( Overlapped 1: N Chat Server )

 : 클라이언트에 다른 클라이언트의 메시지도 전달해 주어야 함

 : 메시지에 클라이언트 ID 포함해야 함 → Packet 단위로 Send/Recv 필요 (size + id + message)

 : 더이상 send와 recv가 차례로 호출되지 않음 (정해진 순서 없이 동시다발적 발생)

 : Send와 Recv가 더이상 OVERLAPPED, WSABUF, BUF를 공유할 수 없음

 

 [ 서버 ]

#include <iostream>
#include <unordered_map>   

#include <WS2tcpip.h>

#pragma comment (lib, "WS2_32.LIB")

constexpr short PORT = 4000;
constexpr int BUFSIZE = 256;

bool b_shutdown = false;

class SESSION;
std::unordered_map<LPWSAOVERLAPPED, int> g_session_map;
std::unordered_map<int, SESSION> g_players;

void CALLBACK send_callback(DWORD err, DWORD sent_size, LPWSAOVERLAPPED pover, DWORD recv_flg);
void CALLBACK recv_callback(DWORD err, DWORD recv_size, LPWSAOVERLAPPED pover, DWORD recv_flg);
void print_error(const char* msg, int err_no);

class EXP_OVER {
public:
	WSAOVERLAPPED over;  // WSAOVERLAPPED 구조체
	WSABUF wsabuf[1];  // WSABUF 배열
	char buf[BUFSIZE];  // 데이터 버퍼
	EXP_OVER(int s_id, char* msg, int m_size)
	{
		ZeroMemory(&over, sizeof(over));
		wsabuf[0].buf = buf;
		wsabuf[0].len = m_size + 2;

		buf[0] = m_size + 2;
		buf[1] = s_id;
		memcpy(buf + 2, msg, m_size);
	}
};

class SESSION {
	char buf[BUFSIZE];  // 데이터 버퍼
	WSABUF wsabuf[1];  // WSABUF 배열
	SOCKET client_s;  // 클라이언트 소켓
	WSAOVERLAPPED wsaover;  // WSAOVERLAPPED 구조체

public:
	SESSION(SOCKET s, int my_id) :client_s(s) {
		g_session_map[&wsaover] = my_id;
		wsabuf[0].buf = buf;
		wsabuf[0].len = BUFSIZE;
	}

	SESSION() {
		std::cout << "ERROR";
		exit(-1);
	}
	~SESSION() { closesocket(client_s); }
	
    // 수신 함수 - 비동기
	void do_recv()
	{
		DWORD recv_flag = 0;
		ZeroMemory(&wsaover, sizeof(wsaover));

		int res = WSARecv(client_s, wsabuf, 1, nullptr, &recv_flag, &wsaover, recv_callback);
		if (res != 0) {
			int err_no = WSAGetLastError();
			if (WSA_IO_PENDING != err_no)
				print_error("WSARecv", WSAGetLastError());
		}
	}

	// 송신 함수 - 비동기
	void do_send(int s_id, char* msg, int recv_size)
	{
		auto b = new EXP_OVER(s_id, msg, recv_size);
		int res = WSASend(client_s, b->wsabuf, 1, nullptr, 0, &b->over, send_callback);
		if (res != 0) {
			print_error("WSARecv", WSAGetLastError());
		}
	}

	void print_message(DWORD recv_size)
	{
		int my_id = g_session_map[&wsaover];
		std::cout << "Client[" << my_id << "] Sent : ";
		for (DWORD i = 0; i < recv_size; ++i)
			std::cout << buf[i];
		std::cout << std::endl;
	}

	void broadcast(int m_size)
	{
		for (auto& p : g_players)
			p.second.do_send(g_session_map[&wsaover], buf, m_size);
	}
};

void print_error(const char* msg, int err_no)
{
	// 출력 메세지를 저장할 buf
	WCHAR* msg_buf;
	FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM,
		NULL, err_no, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT), reinterpret_cast<LPWSTR>(&msg_buf), 0, NULL);
	std::cout << msg;
	std::wcout << L" : 에러 : " << msg_buf;
	while (true);   // 디버깅
	LocalFree(msg_buf);
}

void CALLBACK send_callback(DWORD err, DWORD sent_size, LPWSAOVERLAPPED pover, DWORD recv_flg)
{
	if (err != 0)
		print_error("WSASend", WSAGetLastError());;

	auto b = reinterpret_cast<EXP_OVER*>(pover); // EXP_OVER 객체로 형변환
	delete b;  // 메모리 해제
}

void CALLBACK recv_callback(DWORD err, DWORD recv_size, LPWSAOVERLAPPED pover, DWORD recv_flg)
{
	if (err != 0) {
		print_error("WSARecv", WSAGetLastError());
	}

	int my_id = g_session_map[pover];  // 현재 세션 ID 가져옴
	if (recv_size == 0) {
		// b_shutdown = true;
		// (여기있던 코드) 유저하나가 나갔다고 서버 날리는 짓 
		g_players.erase(my_id);  // 해당 클라이언트 제거
		return;
	}

	g_players[my_id].print_message(recv_size);
	g_players[my_id].broadcast(recv_size);
	g_players[my_id].do_recv();
}

int main()
{

	std::wcout.imbue(std::locale("korean"));

	WSADATA WSAData;
	WSAStartup(MAKEWORD(2, 0), &WSAData);
	SOCKET server_s = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, nullptr, 0, WSA_FLAG_OVERLAPPED);
	SOCKADDR_IN server_a;
	server_a.sin_family = AF_INET;
	server_a.sin_port = htons(PORT);
	server_a.sin_addr.S_un.S_addr = htons(INADDR_ANY);
	bind(server_s, reinterpret_cast<sockaddr*>(&server_a), sizeof(server_a));
	listen(server_s, SOMAXCONN);

	int addr_size = sizeof(server_a);
	int id = 0;
	while (b_shutdown == false) {
		SOCKET client_s = WSAAccept(server_s, reinterpret_cast<sockaddr*>(&server_a), &addr_size, nullptr, 0);
		g_players.try_emplace(id, client_s, id);  // 새로운 클라이언트 세션 생성
		g_players[id++].do_recv();  // 수신 시작
	}

	g_players.clear();  // 모든 세션 제거
	closesocket(server_s);
	WSACleanup();
}

 

 

 [ 클라이언트 ]

#include <iostream>
#include <WS2tcpip.h>

#pragma comment (lib, "WS2_32.LIB")

constexpr short PORT = 4000;
constexpr char SERVER_ADDR[] = "127.0.0.1";
constexpr int BUFSIZE = 256;


bool bshutdown = false;

WSABUF wsabuf[1];  // WSABUF 배열
SOCKET server_s;  // 서버 소켓
char buf[BUFSIZE];  // 버퍼
WSAOVERLAPPED wsaover;  // WSAOVERLAPPED구조체

void CALLBACK send_callback(DWORD, DWORD, LPWSAOVERLAPPED, DWORD);

void read_n_send()
{
	std::cout << "Enter Message : ";
	std::cin.getline(buf, BUFSIZE);
	wsabuf[0].buf = buf;
	wsabuf[0].len = static_cast<int>(strlen(buf) + 1);

	if (wsabuf[0].len == 1) {
		bshutdown = true;
		return;
	}

	ZeroMemory(&wsaover, sizeof(wsaover));

	WSASend(server_s, wsabuf, 1, nullptr, 0, &wsaover, send_callback);
}

void CALLBACK recv_callback(DWORD err, DWORD recv_size, LPWSAOVERLAPPED pwsaover, DWORD sendflag)
{
	int p_size = 0;   // 현재 처리한 데이터 크기
	while (recv_size > p_size) {  // 받은 데이터가 처리되지 않았다면 반복
		int m_size = buf[0 + p_size];  // 메시지 크기 읽기
		std::cout << "Player [" << static_cast<int>(buf[1 + p_size]) << "] : ";  // 플레이어 ID 출력
		for (DWORD i = 0; i < m_size; ++i)
			std::cout << buf[i + p_size + 2];  // 메시지 출력
		std::cout << std::endl;
		p_size = p_size + m_size;  // 처리한 데이터 크기 갱신
	}
	read_n_send();
}

void CALLBACK send_callback(DWORD err, DWORD sent_size, LPWSAOVERLAPPED pwsaover, DWORD sendflag)
{
	wsabuf[0].len = BUFSIZE;
	DWORD recv_flag = 0;
	ZeroMemory(pwsaover, sizeof(*pwsaover));
	WSARecv(server_s, wsabuf, 1, nullptr, &recv_flag, pwsaover, recv_callback);
}

int main()
{
	std::wcout.imbue(std::locale("korean"));

	WSADATA WSAData;
	WSAStartup(MAKEWORD(2, 0), &WSAData);

	server_s = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, nullptr, 0, WSA_FLAG_OVERLAPPED);
	SOCKADDR_IN server_a;
	server_a.sin_family = AF_INET;
	server_a.sin_port = htons(PORT);
	inet_pton(AF_INET, SERVER_ADDR, &server_a.sin_addr);

	connect(server_s, reinterpret_cast<sockaddr*>(&server_a), sizeof(server_a));

	read_n_send();   // 메시지 읽고 전송 시작
	while (false == bshutdown) {
		SleepEx(0, TRUE);  // 이벤트를 기다림
	}

	closesocket(server_s);
	WSACleanup();
}

 

 

- EXP_OVER에서 생성자 부분의 m_size + 2 같은 코드는 상당히 위험한 코드.

- 서버에서 클라로 보내는 패킷이 뭉쳐서 왔을 때의 처리 없고, 임의의 위치에서 잘려 오기도 하는데 이를 이어주는 코드도 필요.

 

- Avatar(나) / PC(Playing Character. 다른 사람)

- 서버의 객체 변화 상태는 BroadCast

- 패킷 종류의 다양화 → 프로토콜 정의 필요 (종류 - Object 추가/이동/삭제, Size/Type/Data(id, 좌표, 아바타여부 등)

 

 

---

 

 

- IOCP 멀티쓰레드

 : GetQueuedCompletionStatus를 별도의 쓰레드로 (GQCS루프를 worker thread)

 : Worker Thread 작성 - Core 개수 / thread::hardware_concurrency() - 쓰레드 개수

// 모두 전역
unordered_map <int, CLIENT> clients;
SOCKET g_s_socket;
HANDLE g_h_iocp;

 

 : g_h_iocp, g_s_socket는 초기화된 후 변하지 않기에 No DataRace

 : Clients 컨테이너 자체가 수정 → DataRace

  → concurrent_unordered_map <int, atomic_shared_ptr <CLIENT>> clients;

 

class CLIENT
{
 	int m_id; // no data race
 	EXP_OVER m_recv_over; // no data race
 	unsigned char m_prev_recv;  // no data race
 	SOCKET m_s;  // no data race -> data race
 	bool m_inuse;  // data race
 	char m_name[MAX_NAME];  // data race
 	short m_x, m_y;  // data race
};

// no data race - 오직 한 쓰레드에서만 접근할 때, 값이 정해진 후 다른 아무도 값을 변경 X

 

 : 재사용 문제

 → clients 재사용시 m_socket이 DataRace. 값이 valid한지를 나타내는 변수가 필요 (m_state)

 →  m_state : ST_FREE(미사용), ST_ALLOC(Accpet됨, 아직 cs_login 받지 않음, 이 클라이언트는 broadcasting에서 제외), ST_INGAME (게임 참여중)