[리눅스 네트워크 프로그래밍] 병행 처리 서버

<서영진 님의 사물인터넷을 위한 리눅스 프로그래밍 with 라즈베리파이 서적을 참고해서 작성하였습니다 :-)>

 

⭐ 병행 처리 서버

앞서 서버 코드를 빌드한 후 2대의 클라이언트가 접속한 경우라면

첫 번째 클라이언트는 서버에 문제없이 접속되지만,

두 번째 클라이언트가 접속하면, 두 번째 클라이언트는 첫 번째 클라이언트의 연결이 끊어질 때까지 대기 상태로 잇께 된다

 

여러 클라이언트를 동시에 병렬로 처리하기 위해서는 멀티 프로세스나 멀티 스레드 기반의 서버가 필요하다

멀티 프로세스 서버의 경우 클라이언트의 접속이 들어올 때마다 새로운 프로세스를 만들어서 처리할 수 있다

BUT 🚨 프로세스를 만드는데 시간이 많이 걸리고, 프로세스 간 통신 (IPC) 이 멀티 스레드에 비해 복잡하기 때문에 요즘은 멀티 스레드를 더 많이 쓴다

 

멀티 스레드 서버도 클라이언트의 요청이 들어올 때마다 새로운 스레드를 만들어서 처리한다

멀티 프로세스보다 쉽게 스레드를 만들 수 있고, 서로 통신이 간단하기 때문에 많이 사용된다

 

다중 클라이언트의 통신을 처리하는 방법으로 멀티 프로세스와 멀티 스레드 방식 이외에도 I/O 멀티 플랙싱 (Multiplexing) 방식이 있다

멀티 플랙싱 방식은 클라이언트의 요구사항이 많지 않은 경우 주로 이용되는데, 이는 하나의 소켓 디스크립털르 효율적으로 사용할 수 있게 해준다

I/O 멀티 플랙싱은 하나의 프로세스 또는 스레드에서 사용하는 다중 처리 방법으로 구분된다

구분	내용	비고
폴링 (Polling)	루프를 이용해서 처리해야 할 일이 있는지 주기적으로 지켜보다가 처리해야 할 작업들이 생기면 순차적으로 돌아가면서 처리하는 방법	poll
셀렉트 (Selecting)	지정된 파일 디스크립터의 변화를 확인해서 변화 (입력, 출력, 에러)가 감지되면 해당 작업을 처리하는 방법	select
인터럽트 (Interrupt)	프로세스가 어떤 작업을 처리하는 도중에 특정한 이벤트가 발생하면 시그널 등을 이용해서 해당 이벤트를 처리하는 방법	시그널, epoll

 

폴링 (polling) 방식은 커널이 주변 장치에게 일을 시킨 후 루프를 돌면서 그 일이 완료되었는지 주기적으로 검사

인터럽트 (interrupt) 방식은 애플리케이션이 주변 장치에게 일을 시킨 후 프로세스는 다른 일을 하며 디바이스가 일을 완료한 후 시그널이나 다른 인터럽트로 알리는 방식

select 와 poll 은 파일 디스크립터를 기반으로 처리한다

select 은 최대 1024 개의 파일 디스크립터만 처리할 수 있지만, poll 은 더 많은 파일 디스크립터를 처리할 수 있다

 

select와 poll이 모두 이벤트를 감지하기 위해 파일 디스크립터를 순차적으로 검색하고, 이벤트가 발생하면 파일 디스크립터의 집합(fd_set)을 모두 변경해야 하기 때문에 속도가 느리다

 

이러한 문제를 해결하기 위해 만들어진 것이 epoll

epoll 은 파일 디스크립터 수의 제한을 받지 않으며, 리눅스 커널에서 파일 디스크립터를 관리하므로 fd_set 복사가 필요 없으므로 가장 빠른 속도를 제공한다

 

➕ 인터럽트와 트랩

시스템에서 발생하는 비동기적인 사건을 알리는 메커니즘

인터럽트 : 현재 수행 중인 프로세스와 관계없이 하드웨어적인 사건을 알리는 메커니즘

트랩 : 현재 수행 중인 프로그램이 유발한 소프트웨어적인 사건의 발생을 알리는 메커니즘

 

☑️ select 함수

I/O 멀티 플랙싱을 위한 대표적인 함수가 select

select() 함수는 파일 디스크립터 변화를 확인하는데, 파일 디스크립터를 통해 데이터 송수신이 가능한 상태인지 확인하는 것

변화를 감지해서 하나의 전성로를 동시에 여러 사용자가 사용할 수 있도록 효율성을 극대화할 수 있다

 

select() 함수를 사용하기 위해서는 파일 디스크립터의 범위를 설정해야 한다

효율적인 수행을 위해 확인해야 하는 파일 디스크립터의 범위를 제한하는 것이다

타임아웃 시간을 설정할 수 있는데 select() 함수가 호출할 때 이벤트 발생에 대해 얼마만큼 기다릴 것인지를 지정할 수 있다

int select (int nfds, fd_set *readfds, fd_set *writefds, fd_Set *exceptfds, struct timeval *timeout);

select() 함수의 첫 번째 인자 (nfds) 는 상태를 지켜볼 파일 디스크립터의 제일 높은 번호

일반적으로 파일 디스크립터의 "가장 큰 디스크립터 번호 + 1" 을 이용

실행 중에 파일이나 소켓을 열었다가 닫으면 파일 디스크립터는 중간 번호가 비게 되는데, 이때 nfds 에 들어갈 숫자는 전체 개수가 아닌 열린 파일 디스크립터의 제일 높은 번호이므로 주의해라

 

마지막 인자 (timeout) 은 함수 호출 후, 무한 대기 상태에 빠지지 않도록 타임아웃을 설정할 수 있다

timeval 구조체를 사용하면 된다

인자	내용	비고
readfds	수신할 데이터가 있는지 확인	입력 버퍼에 데이터 존재
writefds	데이터 전송이 가능한 상태인지 확인	출력 버퍼에 충분한 여유 공간 존재
exceptfds	소켓이 예외 상황이 발생했는지 확인	OOB 메시지 전송 등의 사항

함수 호출에 성공하면 0 이상의 값 반환, 0이 반환되면 timeout 으로 설정된 시간에 타임아웃이 되었다는 의미

오류가 발생하면 -1

 

select() 함수는 fd_set 자료형 사용한다

보고 싶은 파일 디스크립터의번호를 fd_set 자료형에 마크하는데, 기본적으로 시그널 마스크처럼 하나의 비트가 해당 파일 디스크립터 번호와 일치하는 형태로 되어 있다

fd_set 자료형을 설정한 후 select() 함수에 등록해서 실행하면, 이벤트가 발생할 때까지 select() 함수가 대기하고 있다가, 해당 이벤트가 발생했을 때 변화가 생긴 파일 디스크립터를 확인할 수 있다

 

POSIX 에서 사용하는 timespec 구조체를 사용하는 pselect() 함수도 있다

타임아웃 시간을 나노초로 설정할 수 있고, 파일 디스크립터 조사 중 시그널이 발생하는 것을 블록할 수 있도록 지원한다

// select_server.c
// TCP 서버에서 select() 함수를 이용하여 병렬 처리하기
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>

#define SERVER_PORT 5100 		/* 서버의 포트 번호 */

int main(int argc, char **argv)
{
    int ssock;
    socklen_t clen;
    int n;
    struct sockaddr_in servaddr, cliaddr;
    char mesg[BUFSIZ];
    fd_set readfd; 			/* select( ) 함수를 위한 자료형 */
    int maxfd, client_index, start_index;
    int client_fd[5] = {0}; 		/* 클라이언트의 소켓 FD 배열 */

    /* 서버 소켓 디스크립터 연다. */
    if((ssock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        perror("socket()");
        return -1;
    }

    memset(&servaddr, 0, sizeof(servaddr)); 	/* 운영체제에 서비스 등록 */
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(SERVER_PORT);

    if(bind(ssock, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0) {
        perror("bind()");
        return -1;
    }

    if(listen(ssock, 8) < 0) { 		/* 클라이언트의 소켓들을 위한 큐 생성 */
        perror("listen()");
        return -1;
    }

    FD_ZERO(&readfd); 			/* fd_set 자료형을 모두 0으로 초기화 */
    maxfd = ssock; 			/* 현재 최대 파일 디스크립터의 번호는 서버 소켓의 디스크립터 */
    client_index = 0;

    do {
        FD_SET(ssock, &readfd); 	/* 읽기 동작 감지를 위한 fd_set 자료형 설정 */

        /* 클라이언트의 시작 주소부터 마지막 주소까지 fd_set 자료형에 설정 */
        for(start_index = 0; start_index < client_index; start_index++) {
            FD_SET(client_fd[start_index], &readfd);
            if(client_fd[start_index] > maxfd)
                maxfd = client_fd[start_index]; 	/* 가장 큰 소켓의 번호를 저장 */
        }
        maxfd = maxfd + 1;

        /* select( ) 함수에서 읽기가 가능한 부분만 조사 */
        select(maxfd, &readfd, NULL, NULL, NULL); 	/* 읽기가 가능해질 때까지 블로킹 */
        if(FD_ISSET(ssock, &readfd)) { 			/* 읽기가 가능한 소켓이 서버 소켓인 경우 */
            clen = sizeof(struct sockaddr_in);		/* 클라이언트의 요청 받아들이기 */
            int csock = accept(ssock, (struct sockaddr*)&cliaddr, &clen);
            if(csock < 0) {
                perror("accept()");
                return -1;
            } else {
                /* 네트워크 주소를 문자열로 변경 */
                inet_ntop(AF_INET, &cliaddr.sin_addr, mesg, BUFSIZ);
                printf("Client is connected : %s\n", mesg);

                /* 새로 접속한 클라이언트의 소켓 번호를 fd_set에 추가 */
                FD_SET(csock, &readfd);
                client_fd[client_index] = csock;
                client_index++;
                continue;
            }
            if (client_index == 5) break;
        }

        /* 읽기 가능했던 소켓이 클라이언트였던 경우 */
        for(start_index = 0; start_index < client_index; start_index++) {
            /* for 루프를 이용해서 클라이언트들을 모두 조사 */
            if(FD_ISSET(client_fd[start_index], &readfd)) {
                memset(mesg, 0, sizeof(mesg));

                /* 해당 클라이언트에서 메시지를 읽고 다시 전송(Echo) */
                if((n = read(client_fd[start_index], mesg, sizeof(mesg))) > 0) {
                    printf("Received data : %s", mesg);
                    write(client_fd[start_index], mesg, n);
                    close(client_fd[start_index]); 	/* 클라이언트 소켓을 닫는다. */

                    /* 클라이언트 소켓을 지운다. */
                    FD_CLR(client_fd[start_index], &readfd);
                    client_index--;
                }
            }
        }
    } while(strncmp(mesg, "q", 1));

    close(ssock); 					/* 서버 소켓을 닫는다. */

    return 0;
}

/*
클라이언트와의 통신을 위한 소켓을 생성 후, 운영체제에 서비스를 등록하고, 클라이언트의 접속을 처리할 수 있는 대기 큐를 생성하는 기본 작업을 수행함
TCP 서버와 다른 부분은 select() 함수에서 사용할 fd_set 변수형을 선언한 부분임

select() 함수를 사용하더라도 일반적인 클라이언트 통신 처리는 기본 TCP 서버와 비슷함
select() 함수를 이용해서 서버 소켓이나 클라이언트 소켓에서 읽을 데이터가 있는지 확인하고 이를 각각 처리하는 것

fd_set 자료형을 FD_ZERO() 매크로를 통해 모든 비트를 0으로 초기화하고, 서버 솤켓의 파일 디스크립터에 대한 번호를 FD_SET() 매크로를 통해 설정한다
select() 함수를 이용하여 현재의 fd_set 자료형에 대해 읽기가 가능한 이벤트가 있는지 확인하는데, 가장 마지막 인자(timeout) 가 NULL 로 설정되어 있으므로 읽기가 가능할 때까지 무한 대기

서버에 ㅡㄹ라이언트로부터접속이 들어오면 select() 함수에서 빠져나와 fd_set 에 어떤 비트가 설정되어 있는지 확인 가능
FD_ISSET() 매크로를 사용하고, 만약 서버 소켓에서 이벤트가 발생했으면 accept() 함수를 통해 클라이언트의 접속을 받아들이고, fd_set에 FD_SET() 매크로를 통해 새로 접속한 클라이언트의 소켓에 대한 파일 디스크립터를 추가한다

서버는 다시 select() 함수로 돌아가서 서버나 클라이언트에서 데이터가 오는지 확인하고, 새로운 클라이언트의 요청이 들어오면 이전 작업을 다시 수행한다
이미 접속한 클라이언트에서 메시지가 들어오는 경우라면 다음 작업을 수행한다
for 루프와 FD_ISSET() 매크로를 이용해서 이전에 접속된 클라이언트들의 소켓의 배열에서 어떤 클라이언트로부터 데이터가 들어왔는지 확인하고, 데이터가 들어왔으면
read() 함수를 통해 데이터를 읽은 후 다시 write() 함수를 통해 클라이언트로 동일한 ㅁ네시지를 보낸다

TCP 클라이언트를 이용해서 서버에 접속할 수 있다
*/

gcc -o select_server select_server.c 
./select_server &

 gcc -o tcp_client tcp_client.c
./tcp_client 127.0.0.1

 

☑️ epoll API

epoll 은 리눅스 환경에서 사용할 수 있도록 만든 I/O 통지 기법으로 입력과 출력을 담당할 포트를 지정하여 처리하기 때문에 select 와 같이 전체 파일 디스크립터에 대해 순차적으로 감시하지 않고, 전체 파일 디스크립터도 복사하지 않는다

 

운영체제가 직접 파일 디스크립터의 정보를 담은 저장소를 관리하고 관찰 대상을 위한 저장소의 생성을 요청하면, 그 저장소에 해당하는 파일 디스크립터 (epoll_fd) 를 반환한다

무언가 이벤트가 발생하면 epoll_fd 를 통해 확인할 수 있다

epoll은 주로 epoll_create(), epoll_ctl(), epoll_wait() 함수를 사용해서 병행 서버를 만들 수 있다

// epoll_server.c
// TCP 서버 코드를 epoll 의 함수를 이용하여 병렬 처리할 수 있도록 변경하기
#include <stdio.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <sys/epoll.h>

#define SERVER_PORT 5100 			/* 서버의 포트 번호 */
#define MAX_EVENT 32

/* 파일 디스크립터를 넌블로킹 모드로 설정 */
void setnonblocking(int fd)
{
    int opts = fcntl(fd, F_GETFL);
    opts |= O_NONBLOCK;
    fcntl(fd, F_SETFL, opts);
}

int main(int argc, char **argv)
{
    int ssock, csock;
    socklen_t clen;
    int n, epfd, nfds = 1; 			/* 기본 서버 추가 */
    struct sockaddr_in servaddr, cliaddr;
    struct epoll_event ev;
    struct epoll_event events[MAX_EVENT];
    char mesg[BUFSIZ];

    /* 서버 소켓 디스크립터를 연다. */
    if((ssock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        perror("socket()");
        return -1;
    }

    setnonblocking(ssock); 			/* 서버를 넌블로킹 모드로 설정 */

    memset(&servaddr, 0, sizeof(servaddr)); 	/* 운영체제에 서비스 등록 */
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(SERVER_PORT);
    if(bind(ssock, (struct sockaddr*)&servaddr, sizeof(servaddr)) < 0) {
        perror("bind()");
        return -1;
    }

    if(listen(ssock, 8) < 0) { 			/* 클라이언트의 소켓들을 위한 큐 생성 */
        perror("listen()");
        return -1;
    }

    /* epoll_create() 함수를 이용해서 커널에 등록 */
    epfd = epoll_create(MAX_EVENT);
    if(epfd == -1) {
        perror("epoll_create()");
        return 1;
    }

    /* epoll_ctl() 함수를 통해 감시하고 싶은 서버 소켓을 등록 */
    ev.events = EPOLLIN;
    ev.data.fd = ssock;
    if(epoll_ctl(epfd, EPOLL_CTL_ADD, ssock, &ev) == -1) {
        perror("epoll_ctl()");
        return 1;
    }

    do {
        epoll_wait(epfd, events, MAX_EVENT, 500);
        for(int i = 0; i < nfds; i++) {
            if(events[i].data.fd == ssock) { 	/* 읽기가 가능한 소켓이 서버 소켓인 경우 */
                clen = sizeof(struct sockaddr_in);

                /* 클라이언트의 요청 받아들이기 */
                csock = accept(ssock, (struct sockaddr*)&cliaddr, &clen);
                if(csock > 0) {
                    /* 네트워크 주소를 문자열로 변경 */
                    inet_ntop(AF_INET, &cliaddr.sin_addr, mesg, BUFSIZ);
                    printf("Client is connected : %s\n", mesg);
                    setnonblocking(csock); 	/* 클라이언트를 넌블로킹 모드로 설정 */

                    /* 새로 접속한 클라이언트의 소켓 번호를 fd_set에 추가 */
                    ev.events = EPOLLIN | EPOLLET;
                    ev.data.fd = csock;
                    epoll_ctl(epfd, EPOLL_CTL_ADD, csock, &ev);
                    nfds++;
                    continue;
                }
            } else if(events[i].events & EPOLLIN) { 	/* 클라이언트의 입력 */
                if(events[i].data.fd < 0) continue; 	/* 소켓이 아닌 경우의 처리 */
                memset(mesg, 0, sizeof(mesg));

                /* 해당 클라이언트에서 메시지를 읽고 다시 전송(Echo) */
                if((n = read(events[i].data.fd, mesg, sizeof(mesg))) > 0) {
                    printf("Received data : %s", mesg);
                    write(events[i].data.fd, mesg, n);
                    close(events[i].data.fd); 		/* 클라이언트 소켓을 닫고 지운다. */
                    epoll_ctl(epfd, EPOLL_CTL_DEL, events[i].data.fd, NULL);
                    nfds--;
                }
            }
        }
    } while(strncmp(mesg, "q", 1));

    close(ssock); 				/* 서버 소켓을 닫는다. */

    return 0;
}

/*
TCP 서버와 같이 소켓을 생성한 후 bind()와 listen() 함수를 이용해서 서비스를 운영체제에 등록한다
epoll_create() 함수를 이용해서 감시하고 싶은 파일 디스크립터의 숫자를 넣어줬는데, 서버 + 클라이언트의 수로 설정하면 된다
epoll_ctl() 함수에 서버 소켓을 추가하였다. 서버 소켓은 클라이언트의 접속을 감시할 것이므로 입력 (EPOLLIN) 으로 이벤트 설정

서버에 클라이언트가 접속되면 epoll_wait() 함수에서 넘어가게 되는데, 이벤트가 발생한 소켓이 서버이면 accpet() 함수를 통해서 받아들이고,
epoll_ctl() 함수를 통해 다시 커널에 클라이언트의 소켓 디스크립터를 감시할 수 있도록 추가한다

클라이언트 소켓으로부터 이벤트가 발생하면 클라이언트에서 온 데이터를 읽어서 다시 클라이언트로 전송(에코) 하고 클라이언트한테 소켓을 닫은 후
epoll_ctl() 함수를 통해 커널에 추가한 클라이언트 소켓의 감시를 삭제한다

클라이언트는 tcl_client 로 사용
epoll 이 select 에 비해 속도는 빠르지만, 특정 플랫폼에서는 사용할 수 없다

gani@gani:~/raspi/NetworkProgramming $ gcc -o epoll_server epoll_server.c 
gani@gani:~/raspi/NetworkProgramming $ ./epoll_server &
gani@gani:~/raspi/NetworkProgramming $ gcc -o tcp_client tcp_client.c 
gani@gani:~/raspi/NetworkProgramming $ ./tcp_client 
gani@gani:~/raspi/NetworkProgramming $ ./tcp_client 127.0.0.1
*/

gcc -o epoll_server epoll_server.c 
./epoll_server &
gcc -o tcp_client tcp_client.c 
./tcp_client 
./tcp_client 127.0.0.1