윈도우즈 시스템 프로그래밍 15장 - 쓰레드 풀링(Pooling)

 

 

 

 

윈도우즈 시스템 프로그래밍이라는 책과 해당 책의 저자이신 윤성우님의 강의를 통해 공부한 내용을 정리하는 글입니다.

 

 

 

---

 

 

 

커널 오브젝트 생성은 부담스러운 작업이다. 커널 모드와 유저 모드 간의 이동도 빈번하고, 새로운 리소스를 등록하는 등의 작업이 동반되기 때문이다. 쓰레드의 생성도 마찬가지이다.

 

 

쓰레드 풀이란?

 

⦁ 쓰레드의 생성과 소멸은 시스템에 부담을 준다. 그러므로 할당된 일을 마친 쓰레드를 소멸시키지 않고, 쓰레드 풀에 저장해 뒀다가 필요할 때 다시 꺼내 쓰는 개념이다. 쓰레드의 생성과 소멸에 필요한 비용을 지불하지 않겠다는 뜻이다.

 

 

 

쓰레드 풀 구현의 모듈별 해석

 

쓰레드 풀 자료구조

typedef void (*WORK)(void);

DWORD AddWorkToPool(WORK work);
WORK GetWorkFromPool(void);
DWORD MakeThreadToPool(DWORD numOfThread);
void WorkerThreadFunction(LPVOID pParam);

typedef struct __WorkerThread
{
	HANDLE hThread;
	DWORD idThread;
}WorkerThread;

struct __ThreadPool
{
	WORK workList[WORK_MAX];

	WorkerThread workerThreadList[THREAD_MAX];
	HANDLE workerEventList[THREAD_MAX];

	DWORD idxOfCurrentWork;
	DWORD idxOfLastAddedWork;

	DWORD threadIdx;
}gThreadPool;

 

⦁ WORK는 쓰레드에게 일을 시키기 위한 작업의 기본 단위이다.

⦁ WorkerThread는 생성되는 쓰레드의 정보를 담기 위한 구조체이다.

⦁ 구조체 _ThreadPool이 쓰레드 풀에 해당되는 자료구조이다. 그 멤버 중 아래 멤버들은 Work에 관련된 멤버들이다.

WORK workList[WORK_MAX];
DWORD idxOfCurrentWork;        // 대기 1순위 Work Index
DWORD idxOfLastAddedWork;    // 마지막 추가 Work Index+1

workList는 Work를 등록하는 저장소이다.

idxOfLastAddedWork는 마지막에 추가된 Work index보다 1 많은 값을 유지하며 새로운 Work가 등록될 때 등록 위치를 가리킨다.

idxOfCurrentWork는 처리되어야 할 Work의 위치를 가리킨다.

 

아래 멤버들은 쓰레드와 관련된 멤버들이다.

WorkerThread workerThreadList[Thread_MAX];
HANDLE workerEventList[THREAD_MAX];
DWORD threadIdx;

workerThreadList에는 풀에 저장된 쓰레드 정보를 저장한다.

workerEventList는 각 쓰레드 별로 하나씩 할당되는 이벤트 동기화 오브젝트를 저장하는 배열이다.

threadIdx는 저장된 쓰레드의 개수 정보를 담는다.

 

쓰레드에 할당된 일이 없을 경우에는 WaitForSingleObject와 같은 함수를 호출하여 Blocked 상태를 만들어줘야 한다.

그러므로 쓰레드마다 하나씩 이벤트 오브젝트를 가져야하므로 이를 workerEventList에 저장해둔다.

 

idxOfCurrentWork와 idxOfLastAddedWork는 증가만 하지 감소는 하지 않으므로 THREAD_MAX만큼 일이 등록되면 더 이상 등록할 수 없다. 이는 자료구조나 STL을 활용하여 해결이 가능하다.

 

 

쓰레드 풀의 함수 관계

 

⦁ Work GetWorkFromPool(void);

쓰레드 풀에서 Work를 가져올 때 호출하는 함수이다.

 

⦁ DWORD AddWorkToPool(WORK work);

새로운 Work를 등록할 때 호출하는 함수이다.

 

⦁ DWORD MakeThreadToPool(DWORD numOfThread);

쓰레드 풀이 생성된 이후에 풀에 쓰레드를 생성(등록)하는 함수이다. 인자로 전달되는 수 만큼 쓰레드가 생성된다.

 

⦁ void WorkerThreadFunction(LPVOID pParam);

쓰레드가 생성되자마자 호출하는 쓰레드의 main 함수이다. 이 함수의 구성을 통해 Work를 어떻게 할당하고 처리하는지 Work가 없을 때의 쓰레드 상태를 알 수 있다.

 

 

⦁ 쓰레드 풀의 매커니즘과 함수와의 관계

 

① 전역으로 선언된 쓰레드 풀에 MakeThreadToPool 함수의 호출을 통해 쓰레드를 생성하여 등록시킨다. 이 쓰레드는 이벤트 오브젝트가 Signaled 상태가 되기를 기다리며 Blocked 상태가 된다.

② AddWorkToPool 함수 호출을 통해서 Work를 등록한다.

③ Work가 등록되면, 쓰레드 풀에서 Blocked 상태에 있는 모든 이벤트 오브젝트를 Signaled 상태로 변경한다.

④ 모든 이벤트 오브젝트가 Signaled 상태가 되므로, 모든 쓰레드가 Running 상태가 된다. 그러나 Work를 할당받은 하나의 쓰레드를 제외하고 나머지는 다시 Blocked 상태가 된다. 이는 코드의 간결성을 위하여 다소 비효율적으로 작성된 부분이다.

⑤ Running 상태로 남아있는 하나의 쓰레드는 GetWorkFromPool 함수 호출로 Work를 할당받아 실행된다.

 

 

 

전역으로 선언된 쓰레드 풀 접근 동기화

 

쓰레드 풀에 해당하는 gThreadPool은 전역으로 선언되었다. 이는 둘 이상의 쓰레드에 의해 참조되는 메모리 영역이다. 그러므로 동기화가 필요하다. 아래 함수들은 뮤텍스 기반 동기화 함수들을 래핑(Rapping)한 함수이다.

void InitMutex(void);        // CreateMutex Rapping
void DeInitMutex(void);    // CloseHandle Rapping
void AcquireMutex(void);  // WaitForSingleObject Rapping
void ReleaseMutex(void);  // ReleaseMutex Rapping

 

 

 

쓰레드 풀 구현 소스코드

 

⦁ ThreadPooling.cpp

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

#define WORK_MAX 10000
#define THREAD_MAX 50

typedef void (*WORK)(void);

DWORD AddWorkToPool(WORK work);
WORK GetWorkFromPool(void);
DWORD MakeThreadToPool(DWORD numOfThread);
void WorkerThreadFunction(LPVOID pParam);

typedef struct __WorkerThread
{
	HANDLE hThread;
	DWORD idThread;
}WorkerThread;

struct __ThreadPool
{
	WORK workList[WORK_MAX];

	WorkerThread workerThreadList[THREAD_MAX];
	HANDLE workerEventList[THREAD_MAX];

	DWORD idxOfCurrentWork;
	DWORD idxOfLastAddedWork;

	DWORD threadIdx;
}gThreadPool;

static HANDLE mutex = NULL;

void InitMutex(void)
{
	mutex = CreateMutex(NULL, FALSE, NULL);
}

void DeInitMutex(void)
{
	BOOL ret = CloseHandle(mutex);
}

void AcquireMutex(void)
{
	DWORD ret = WaitForSingleObject(mutex, INFINITE);
	if (ret == WAIT_FAILED)
		_tprintf(_T("Error Occur! \n"));
}

void ReleaseMutex(void)
{
	BOOL ret = ReleaseMutex(mutex);
	if (ret == 0)
		_tprintf(_T("Error Occur! \n"));
}

DWORD AddWorkToPool(WORK work)
{
	AcquireMutex();
	if (gThreadPool.idxOfLastAddedWork >= WORK_MAX)
	{
		_tprintf(_T("AddWorkToPool fail! \n"));
		return NULL;
	}

	gThreadPool.workList[gThreadPool.idxOfLastAddedWork++] = work;

	for (DWORD i = 0; i < gThreadPool.threadIdx; i++)
		SetEvent(gThreadPool.workerEventList[i]);

	ReleaseMutex();
	return 1;
}

WORK GetWorkFromPool()
{
	WORK work = NULL;

	AcquireMutex();

	if (!(gThreadPool.idxOfCurrentWork < gThreadPool.idxOfLastAddedWork))
	{
		ReleaseMutex();
		return NULL;
	}

	work = gThreadPool.workList[gThreadPool.idxOfCurrentWork++];
	ReleaseMutex();

	return work;
}

DWORD MakeThreadToPool(DWORD numOfThread)
{
	InitMutex();
	DWORD capacity = WORK_MAX - (gThreadPool.threadIdx);

	if (capacity < numOfThread)
		numOfThread = capacity;

	for (DWORD i = 0; i < numOfThread; i++)
	{
		DWORD idThread;
		HANDLE hThread;

		gThreadPool.workerEventList[gThreadPool.threadIdx] = CreateEvent(NULL, FALSE, FALSE, NULL);

		hThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)WorkerThreadFunction, (LPVOID)gThreadPool.threadIdx, 0, &idThread);

		gThreadPool.workerThreadList[gThreadPool.threadIdx].hThread = hThread;

		gThreadPool.workerThreadList[gThreadPool.threadIdx].idThread = idThread;

		gThreadPool.threadIdx++;
	}

	return numOfThread;
}

void WorkerThreadFunction(LPVOID pParam)
{
	WORK workFunction;
	HANDLE event = gThreadPool.workerEventList[(DWORD)pParam];

	while (1)
	{
		workFunction = GetWorkFromPool();
		if (workFunction == NULL)
		{
			WaitForSingleObject(event, INFINITE);
			continue;
		}
		workFunction();
	}
}

void TestFunction()
{
	static int i = 0;
	i++;

	_tprintf(_T("Good Test --%d : Processing thread: %d--\n\n"), i, GetCurrentThreadId());
}

int _tmain(int argc, TCHAR* argv[])
{
	MakeThreadToPool(3);

	for (int i = 0; i < 100; i++)
	{
		AddWorkToPool(TestFunction);
	}

	Sleep(50000);
	return 0;
}

 

Work에 해당하는 함수는 TestFunction이다. 이 함수에 선언된 static i도 동기화가 필요하다.