1.1절. Thread Pooling 이란

pool 의 사전적인 뜻을 찾아보면 연못, 저수지, 수영장 풀 등 "무엇을 담아놓는" 의 뜻을 가진다. 이대로 해석하자면 Thread Pooling 이란 쓰레드를 담아 놓는 용기(메모리가 될것이다) 를 뜻하며, 프로그래밍 측면에서 해석하자면, "미리 쓰레드를 할당시켜 놓는기법" 을 뜻한다.

그렇다면 쓰레드를 미리 할당시켜 놓는 이유에 대해서 생각해보자, 지금까지 이 사이트에서 다루었던 쓰레드프로그래밍 기법은 기본적으로 fork 방식과 매우 비슷하며, 쓰레드를 생성시켜야 될 필요가 있을때 pthread_create(3)등의 함수를 이용하여 새로운 작업쓰레드를 생성시키는 방식을 사용했다. 보통 쓰레드프로그래밍은 네트웍 프로그래밍시 주로 사용됨으로 accept(2) 로 연결을 기다리다가 연결이 만들어지면 accept 에서 넘어온 소켓 지시자를 인자로 하는 쓰레드를 생성했다.

이러한 방식 - 요청이 있을때 쓰레드를 생성시키는 - 의 쓰레드 프로그래밍기법은 대부분의 작업을 처리하기에 충분히 효율적이며, 빠르긴하지만 클라이언트로 부터의 연결과 종료가 매우 바쁘게 일어나는 서버의 경우, 계속적으로 쓰레드를 생성하고 종료해야 하는 비용을 무시할수 없게 된다. 쓰레드가 비록 fork()에 비해서 생성과 소멸시에 훨씬 적은 비용을 소모한다고는 하지만, 이건 어디까지나 상대적인 것으로 실상은 꽤 많은 시간과 비용을 소비하는 작업이다. 특히 Linux 에서의 Pthread 의 경우 clone(2)를 이용한 구현임으로 더욱더 많은 비용을 소비하게 된다.

Thread Pooling 은 이러한 반복적인 쓰레드의 생성/소멸에 의한 비효율적인 측면을 없애고자 하는 목적으로 만들어진 프로그래밍 기법이다.

1.2절. 예제

지금까지 Thread POOL 의 구현방법에 대해서 알아봤으니, 간단하게 구현해 보도록 하겠다. 이 코드는 지극히 기능구현에만 신경쓴 코드이다. 에러처리와 몇군데 뮤텍스잠금처리는 각자의 재량에 맡기겠다.

예제 : pool_echo.cc

#include <map> #include <iostream> #include <pthread.h> #include <unistd.h> #include <stdlib.h> #include <time.h> #include <sys/socket.h> #include <sys/stat.h> #include <signal.h> #include <arpa/inet.h> #include <sys/un.h> #include <stdio.h> #include <stdlib.h> #include <string.h> // 최대 쓰레드 POOL 크기 #define MAX_THREAD_POOL 256 using namespace std; // 전역 쓰레드 구조체 typedef struct _ph { int sockfd; // 현재 사용중인 소켓 fd int index_num; // 인덱스 번호 } ph; // 전역쓰레드 구조체로써 // 현재 쓰레드 상황을 파악함 struct schedul_info { int client_num; // 현재 연결된 클라이언트수 int current_sockfd; // 가장최근에 만들어진 소켓지시자 multimap<int, ph> phinfo; }; // 각 쓰레드별 조건변수 pthread_cond_t *mycond; // 쓰레드 동기화를 위한 조건변수 pthread_cond_t async_cond = PTHREAD_COND_INITIALIZER; // 각 쓰레드별 조건변수의 크리티컬세션 지정을 위한 // 뮤텍스 pthread_mutex_t mutex_lock= PTHREAD_MUTEX_INITIALIZER; // 쓰레드 동기화용 조건변수의 크리티컬세션 지정을 위한 // 뮤텍스 pthread_mutex_t async_mutex = PTHREAD_MUTEX_INITIALIZER; // 클라이언트와의 통신용 쓰레드 void *thread_func(void *data); // 현재 클라이언트 상태 모니터용 쓰레드 // 한마디로 디버깅용 void *mon_thread(void *data); schedul_info s_info; // 메인 함수 int main(int argc, char **argv) { int i; ph myph; int status; int pool_size = atoi(argv[2]); pthread_t p_thread; struct sockaddr_in clientaddr, serveraddr; int server_sockfd; int client_sockfd; int client_len; // 풀사이즈 검사 if ((pool_size < 0) || (pool_size > MAX_THREAD_POOL)) { cout << "Pool size Error" << endl; exit(0); } // Make Socket if ((server_sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) { perror("error : "); exit(0); } // Bind bzero(&serveraddr, sizeof(serveraddr)); serveraddr.sin_family = AF_INET; serveraddr.sin_addr.s_addr = htonl(INADDR_ANY); serveraddr.sin_port = htons(atoi(argv[1])); if (bind (server_sockfd, (struct sockaddr *)&serveraddr, sizeof(serveraddr)) == -1) { perror("bind error : "); exit(0); } // Listen if (listen(server_sockfd, 5) == -1) { perror("listen error : "); exit(0); } // 쓰레드 갯수만큼 조건변수 생성 mycond = (pthread_cond_t *)malloc(sizeof(pthread_cond_t)*pool_size); // 쓰레드 전역변수 초기화 s_info.client_num = 0; s_info.current_sockfd = 0; // 쓰레드 POOL 새성 for (i = 0; i < pool_size; i++) { memset((void *)&myph, 0x00, sizeof(myph)); myph.index_num = i; s_info.phinfo.insert(pair<int, ph>(0, myph)); // 조건변수를 이용해서 쓰레드간 동기화를 실시한다. pthread_mutex_lock(&async_mutex); if (pthread_create(&p_thread, NULL, thread_func, (void *)&i) < 0) { perror("thread Create error : "); exit(0); } pthread_cond_wait(&async_cond, &async_mutex); pthread_mutex_unlock(&async_mutex); } // 디버깅용 쓰레드 생성 pthread_create(&p_thread, NULL, mon_thread, (void *)NULL); // MAIN THREAD accept wait client_len = sizeof(clientaddr); // 클라이언트 ACCEPT 처리를 위한 // MAIN 쓰레드 while(1) { multimap<int, ph>::iterator mi; client_sockfd = accept(server_sockfd, (struct sockaddr *)&clientaddr, (socklen_t *)&client_len); if (client_sockfd > 0) { // 만약 쓰레드풀이 가득찼다면 클라이언트 연결을 // 종료시킨다. mi = s_info.phinfo.begin(); if (mi->first == 1) { //error message send to client_sockfd cout << "SOCKET IS FULL" << endl; close(client_sockfd); } // 그렇지 않다면 연결을 받아들이고 // 클라이언트 전역 변수를 세팅한다. // 세팅후 해당 처리쓰레드에게 시그널을 보내어서 // 처리하게 한다. else { ph tmpph; int psockfd; int pindex_num; s_info.current_sockfd = client_sockfd; tmpph.sockfd = client_sockfd; tmpph.index_num = mi->second.index_num; s_info.phinfo.erase(mi); s_info.phinfo.insert(pair<int, ph>(1,tmpph)); s_info.client_num ++; cout << "SEND SIGNAL " << mi->second.index_num << endl; pthread_cond_signal(&mycond[mi->second.index_num]); } } else { cout << "ACCEPT ERROR " << endl; } } pthread_join(p_thread, (void **)status); } void *thread_func(void *data) { char buf[255]; int mysocket; int mynum = *((int *)data); multimap<int, ph>::iterator mi; // 쓰레드 동기화용 조건변수 pthread_mutex_lock(&async_mutex); pthread_cond_signal(&async_cond); pthread_mutex_unlock(&async_mutex); cout << "Thread create " << mynum << endl; while(1) { // MAIN 쓰레드로 부터 신호를 기다린다. // 신호가 도착하면 쓰레드 전역변수로 부터 // 현재 처리해야할 소켓지정값을 가져온다. pthread_mutex_lock(&mutex_lock); pthread_cond_wait(&mycond[mynum], &mutex_lock); mysocket = s_info.current_sockfd; pthread_mutex_unlock(&mutex_lock); memset(buf, 0x00, 255); // 데이타를 처리한다. // 만약 quit 문자열을 만나면 // 쓰레드 전역변수를 세팅한다음 연결종료 한다. while(1) { read(mysocket, buf, 255); if (strstr(buf, "quit") == NULL) { write(mysocket, buf, 255); } else { mi = s_info.phinfo.begin(); while(mi != s_info.phinfo.end()) { cout << "search " << mi->second.index_num << endl; if (mi->second.index_num == mynum) { ph tmpph; tmpph.index_num = mynum; tmpph.sockfd = 0; s_info.phinfo.erase(mi); s_info.phinfo.insert(pair<int, ph>(0, tmpph)); s_info.client_num --; close(mysocket); break; } mi ++; } break; } memset(buf, 0x00, 255); } } } void *mon_thread(void *data) { cout << "moniter thread" << endl; while(1) { sleep(10); multimap<int, ph>::iterator mi; mi = s_info.phinfo.begin(); cout << "size " << s_info.phinfo.size() << endl; while(mi != s_info.phinfo.end()) { cout << mi->first << " : " << mi->second.index_num << " : " << mi->second.sockfd << endl; mi ++; } } }

이 프로그램은 2개의 인자를 받아들이며, 클라이언트의 입력을 되돌려주는 일을한다 (echo 서버). 첫번째 인자는 서비스할 PORT 번호이고, 두번째 인자는 쓰레드 생성갯수이다. 프로그램은 인자의 정보를 이용해서 PORT 를 열고 클라이언트를 받아들인다. 클라이언트가 연결하면, Thread Pool 에 남는 공간이 있는지를 확인하고, 남는 공간이 있다면 클라이언트와 통신하게 된다.

단지 쓰레드를 미리 생성시키고 나서, 이것을 스케쥴링하기 위한 코드가 몇줄 추가되었을 뿐 특별히 복잡한 코드는 아닐거라고 생각된다.