[TCP/IP 소켓 프로그래밍] (14) 쓰레드 동기화 (윈도우)

출처 : 열혈 TCP/IP 소켓 프로그래밍 윤성우 저

 

유저 모드와 커널 모드

유저 모드: 응용 프로그램이 실행되는 기본 모드, 하드웨어 접근이 제한되고 사용할 수 있는 메모리 공간에도 제약이 있다.

커널 모드: 운영체제가 실행될 때의 모드, 메모리, 하드웨어 접근에 제약이 없다.

 

운영체제는 일반 프로그램이 실행되는 모드는 유저 모드로 두어 컴퓨터 자원을 보호한다. 쓰레드와 같은 운영체제 자원을 생성할 시 유저 모드에서 커널 모드의 변환이 필요하며 생성된 쓰레드를 다시 커널모드에서 유저 모드로 전환하여 전달한다.

 

쓰레드 동기화, 유저 모드 vs 커널 모드

유저 모드에서 사용하는 동기화 기법은 운영체제의 힘을 빌리지 않기 때문에 빠르다.

커널 모드에서 사요하는 동기화 기법은 운영체제의 힘을 빌리기 떄문에 비용이 더 크다. 유저 모드 보다 더 많은 기능을 제공하고 데드락을 방지하기 위해 타임 아웃을 지정할 수 있다.

 

유저 모드 동기화 기법

CRTICAL_SECTION

유저 모드 동기화  기법중 하나로 유닉스의 뮤텍스처럼 임계영역 진입을 잠그고 푼다.

생성 및 삭제

void InitializeCriticalSection(LPCRITICAL_SECTION lpCriticalSection);
void DeleteCriticalSection(LPCRITICAL_SECTION lpCriticalSection);

획득 및 반납

void EnterCriticalSection(LPCRITICAL_SECTION lpCriticalSection);
void LeaveCriticalSection(LPCRITICAL_SECTION lpCriticalSection);

CRTICAL_SECTION은 운영체제 자원이 아니므로 커널 오브젝트가 아니다.

 

커널 모드 동기화 기법

Mutex

HANDLE CreateMutex(LPSECURITY_ATTRIBUTES lpMutexAttributes, BOOL bInitialOwner, LPCSTR lpName);

• lpMutexAttributes: 보안관련 속성 정보
• bInitialOwner: TRUE 전달 시 Non-signaled 상태의 Mutex 오브젝트를 생성하고 함수를 호출한 스레드가 뮤텍스를 소유하여 잠그게 된다. FALSE 전달 시 Signal 상태의 Mutex 오브젝트를 생성하고 소유자가 없으므로 누구든 잠글 수 있다.
• lpName: 이름을 부여

커널 오브젝트 소멸

BOOL CloseHandle(HANDLE hObject);

hObject: 소멸하고자하는 커널 오브젝트를 전달한다. Semaphore와 Event에 쓰이는 함수이다.

 

Mutex 반납

BOOL ReleaseMutex(HANDLE hMutex);

Mutex 획득

Mutex는 auto-reset 모드 커널 오브젝트이기 때문에 WaitForSingleObject로 획득 시 자동으로 non-signaled 상태가 되는 커널 오브젝트이다. non-signaled 상태의 mutex는 다른 쓰레드에 의해 소유될 수 없다. 획득한 쓰레드가 반납하여야 signal 상태로 복귀하게되고 다른 쓰레드가 mutex를 소유할 수 있게 된다.

WaitForSingleObject(hMutex, INFINITE);
// 임계영역 시작
// ...
// 임계영역 끝
ReleaseMutex(hMutex);

 Mutex를 이용한 증감 연산 동기화 문제 해결

#include <stdio.h>
#include <windows.h>
#include <process.h>

#define NUM_THREAD 50
unsigned WINAPI threadInc(void* arg);
unsigned WINAPI threadDes(void* arg);
long long num = 0;
HANDLE hMutex;

int main(int argc, char* argv[])
{
	HANDLE tHandles[NUM_THREAD];
	int i;

	hMutex = CreateMutex(NULL, FALSE, NULL);
	printf("sizeof long long: %d\n", sizeof(long long));
	for (i = 0; i < NUM_THREAD; ++i)
	{
		if (i % 2)
			tHandles[i] = (HANDLE)_beginthreadex(NULL, 0, threadInc, NULL, 0, NULL);
		else
			tHandles[i] = (HANDLE)_beginthreadex(NULL, 0, threadDes, NULL, 0, NULL);
	}

	WaitForMultipleObjects(NUM_THREAD, tHandles, TRUE, INFINITE);
	printf("end of main num %d", num);
	return 0;
}

unsigned WINAPI threadInc(void* arg)
{
	WaitForSingleObject(hMutex, INFINITE);
	for (int i = 0; i < 1000000; ++i)
	{
		++num;
	}
	ReleaseMutex(hMutex);
	return 0;
}

unsigned WINAPI threadDes(void* arg)
{
	WaitForSingleObject(hMutex, INFINITE);
	for (int i = 0; i < 1000000; ++i)
	{
		--num;
	}
	ReleaseMutex(hMutex);
	return 0;
}

 

Semaphore

HANDLE CreateSemaphore(LPSECURITY_ATTRIBUTES lpSemaphoreAttributes, LONG lInitialCount, LONG lMaximumCount, LPCSTR lpName);

• lpSemalphoreAttributes: 보안 속성정보
• lInitialCount: 세마포어 초기값 지정. lMaximumCount 보다 크면 안되고 0 이상 값
• lMaximumCount: 최대 세마포어 값 지정
• lpName: 이름 지정.

0 일때 non-signaled 상태 0 보다 크면 signaled 상태에 있다는 뜻이다.

 

세마포어 반납

BOOL ReleaseSemaphore(HANDLE hSemaPhore, LONG lReleaseCount, LPLONG lpPreviousCount);

• hSemaphore 반납할 Semaphore 오브젝트 핸들 전달
• lReleaseCount: 세마포어 값을 증가시키고 최대를 넘는다면 FALSE를 반환한다.
• lpPreviousCount: 변경 이전의 세마포어 값을 저장할 수 있도록 주소를 전달한다.

획득방법은 Mutex와 같다.

WaitForSingleObject(hSemaphore, INFINITE);
// 임계영역 시작
// ...
// 임계영역 끝
ReleaseSemaphore(hSemaphore, 1 NULL);

 

Event

Event 동기화 오브젝트는 auto-reset모드와 manual-reset모드를 선택하여 생성할 수 있다.

HANDLE CreateEvent(LPSECURITY_ATTRIBUTES lpEventAttributes, BOOL bManualReset, BOOL bInitialState, LPCTSTR lpName);

• lpEventAttributes: 보안관련 설정
• bManualReset: TRUE 전달 시 manual-reset 모드 Event, FALSE 전달 시 auto-reset 모드 Event 오브젝트 생성
• bInitalState: TRUE 전달 시 signaled 상태의 Event, FALE 전달시 non-signaled 상태의 Event 생성
• lpName: 이름을 부여

Event 오브젝트가 manual-reset 모드라면 다음 두 함수를 이용하여 signal 상태값을 변경해야한다.

BOOL ResetEvent(HANDLE hEvent); // to Non-signaled state
BOOL SetEvent(HANDLE hEvent); // to Signaled state

 

사용 예제

#include <stdio.h>
#include <windows.h>
#include <process.h>

#define STR_LEN 100
unsigned WINAPI NumberOfA(void* arg);
unsigned WINAPI NumberOfOthers(void* arg);

static char str[STR_LEN];
static HANDLE hEvent;


int main(int argc, char* argv[])
{
	HANDLE hThread1, hThread2;

	// manual-reset 모드의 non-signaled 상태 event 객체 생성
	hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
	hThread1 = (HANDLE)_beginthreadex(NULL, 0, NumberOfA, NULL, 0, NULL);
	hThread2 = (HANDLE)_beginthreadex(NULL, 0, NumberOfOthers, NULL, 0, NULL);

	fputs("Input string: ", stdout);
	fgets(str, STR_LEN, stdin);
	SetEvent(hEvent);  // signaled
	WaitForSingleObject(hThread1, INFINITE);
	WaitForSingleObject(hThread2, INFINITE);
	ResetEvent(hEvent); // non-signaled
	CloseHandle(hEvent);
	return 0;
}

unsigned WINAPI NumberOfA(void* arg)
{
	int i, cnt = 0;
	WaitForSingleObject(hEvent, INFINITE);
	for (i = 0; str[i] != 0; ++i)
	{
		if (str[i] == 'A')
			++cnt;
	}
	printf("Num of A: %d\n", cnt);
	return 0;
}

unsigned WINAPI NumberOfOthers(void* arg)
{
	int i, cnt = 0;
	WaitForSingleObject(hEvent, INFINITE);
	for (i = 0; str[i] != 0; ++i)
	{
		if (str[i] != 'A')
			++cnt;
	}
	printf("Num of Others: %d\n", cnt);
	return 0;
}

 

Event 객체는 생성시 non-signal 상태에 있었다. 그리고 SetEvent 호출로 Signal 상태가 되어 두 쓰레드는 작업을 시작하게 된다. 작업이 끝난 뒤에도 manual-reset 모드인 Event 객체는 signale 상태에 있기 때문에 ResetEvent 호출로 Non-signale 상태로 만들어 준다.

	SetEvent(hEvent);  // signaled
	WaitForSingleObject(hThread1, INFINITE);
	WaitForSingleObject(hThread2, INFINITE);
	ResetEvent(hEvent); // non-signaled
	CloseHandle(hEvent);

 

윈도우 기반 멀티 스레드 서버 구현

 

참고 : 열혈 TCP/IP 소켓 프로그래밍

 

서버

#include "../network_header.h"
#include "../ErrorHandling.h"

unsigned WINAPI HandleClnt(void* arg);
void SendMsg(char* msg, int len);

int clnt_cnt = 0;
int clnt_socks[MAX_CLNT];
HANDLE hMutex;
HANDLE hThread;

int main(int argc, char* argv[])
{
    // 윈속 라이브러리 초기화
    WSADATA wsaData;
    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
        ErrorHandling("WSAStartup() error");

    SOCKET serv_sock, clnt_sock;
    SOCKADDR_IN serv_adr, clnt_adr;
    int clnt_adr_sz;
   
    hMutex = CreateMutex(NULL, FALSE, NULL);
    serv_sock = socket(PF_INET, SOCK_STREAM, 0);

    memset(&serv_adr, 0, sizeof(serv_adr));
    serv_adr.sin_family = AF_INET;
    serv_adr.sin_addr.s_addr = htonl(INADDR_ANY);
    serv_adr.sin_port = htons(PORT);

    if (bind(serv_sock, (SOCKADDR*)&serv_adr, sizeof(serv_adr)) == -1)
        ErrorHandling("bind() error");

    if (listen(serv_sock, 5) == -1)
        ErrorHandling("listen() error");

    while (1)
    {
        clnt_adr_sz = sizeof(clnt_adr);
        clnt_sock = accept(serv_sock, (SOCKADDR*)&clnt_adr, &clnt_adr_sz);
        WaitForSingleObject(hMutex, INFINITE);
        clnt_socks[clnt_cnt++] = clnt_sock;
        ReleaseMutex(hMutex);

        hThread = (HANDLE)_beginthreadex(NULL, 0, HandleClnt, &clnt_sock, 0, NULL);
        printf("Connected client IP: %s\n", inet_ntoa(clnt_adr.sin_addr));
    }
    CloseHandle(hMutex);
    closesocket(serv_sock);

    // 윈속 라이브러리 해제
    WSACleanup();
    return 0;
}

unsigned WINAPI HandleClnt(void* arg)
{
    int clnt_sock = *((int*)arg);
    int str_len = 0, i;
    char msg[BUF_SIZE];

    while ((str_len = recv(clnt_sock, msg, sizeof(msg), 0)) != 0)
        SendMsg(msg, str_len);

    WaitForSingleObject(hMutex, INFINITE);
    for (i = 0; i < clnt_cnt; ++i)
    {
        if (clnt_sock == clnt_socks[i])
        {
            while (i++ < clnt_cnt - 1)
            {
                clnt_socks[i] = clnt_socks[i + 1];
            }
            break;
        }
    }
    clnt_cnt--;
    ReleaseMutex(hMutex);
    closesocket(clnt_sock);
    return NULL;
}

void SendMsg(char* msg, int len)
{
    int i;
    WaitForSingleObject(hMutex, INFINITE);
    for (i = 0; i < clnt_cnt; i++)
        send(clnt_socks[i], msg, len, 0);
    ReleaseMutex(hMutex);
}

클라이언트

#include "../network_header.h"
#include "../ErrorHandling.h"
#define BUF_SIZE 100
#define NAME_SIZE 20

unsigned WINAPI SendMsg(void* arg);
unsigned WINAPI RecvMsg(void* arg);
void error_handling(char* msg);

char name[NAME_SIZE] = "[DEFAULT]";
char msg[BUF_SIZE];

int main(int argc, char* argv[])
{
	// 윈속 라이브러리 초기화
	WSADATA wsaData;
	if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
		ErrorHandling("WSAStartup() error");

	SOCKET sock;
	SOCKADDR_IN serv_addr;
	HANDLE snd_thread, rcv_thread;

	sock = socket(PF_INET, SOCK_STREAM, 0);

	memset(&serv_addr, 0, sizeof(serv_addr));
	serv_addr.sin_family = AF_INET;
	serv_addr.sin_addr.s_addr = inet_addr(DEFAULT_NETWORK);
	serv_addr.sin_port = htons(PORT);

	if (connect(sock, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) == -1)
		ErrorHandling("connect() error");

	snd_thread = (HANDLE)_beginthreadex(NULL, 0, SendMsg, (void*)&sock, 0, NULL);
	rcv_thread = (HANDLE)_beginthreadex(NULL, 0, RecvMsg, (void*)&sock, 0, NULL);

	DWORD wr;
	wr = WaitForSingleObject(snd_thread, INFINITE);
	if (wr == WAIT_FAILED)
		ErrorHandling("Send Thread failed\n");

	wr = WaitForSingleObject(rcv_thread, INFINITE);
	if (wr == WAIT_FAILED)
		ErrorHandling("Receive Thread failed\n");
	closesocket(sock);

	// 윈속 라이브러리 해제
	WSACleanup();
	return 0;
}

unsigned WINAPI SendMsg(void* arg)   // send thread main
{
	int sock = *((int*)arg);
	char name_msg[NAME_SIZE + BUF_SIZE];
	while (1)
	{
		fgets(msg, BUF_SIZE, stdin);
		if (!strcmp(msg, "q\n") || !strcmp(msg, "Q\n"))
		{
			closesocket(sock);
			exit(0);
		}
		sprintf(name_msg, "%s %s", name, msg);
		send(sock, name_msg, strlen(name_msg), 0);
	}
	return TRUE;
}

unsigned WINAPI RecvMsg(void* arg)   // read thread main
{
	int sock = *((int*)arg);
	char name_msg[NAME_SIZE + BUF_SIZE];
	int str_len;
	while (1)
	{
		str_len = recv(sock, name_msg, NAME_SIZE + BUF_SIZE - 1, 0);
		if (str_len == -1)
			return FALSE;
		name_msg[str_len] = 0;
		fputs(name_msg, stdout);
	}
	return TRUE;
}