'epoll 클라이언트'에 해당되는 글 1건

epoll 뿐만 아닌 다중 입출력의 설명과 코드를 아래의 note에서 확인하실 수 있습니다. 

https://reakwon.tistory.com/233

epoll

epoll은 poll과 비슷한 방식으로 동작하지만 전체 파일 디스크립터가 아닌 이벤트가 발생한 객체만 되돌려줍니다. 그리고 두 가지 이벤트 트리거 방식을 선택하여 동작할 수 있는데, Level Trigger 방식과 Edge Trigger 방식입니다. 이 둘을 줄여서 LT, ET라고 하겠습니다. 

이번 글에서는 epoll에 대한 개념을 설명한 후에 Level Trigger 방식의 에코 서버, Edge Trigger 방식의 에코서버의 예제코드를 작성해보도록 하겠습니다. 서버의 소스코드나 클라이언트 소스코드가 길게 느껴질 수 있는데, 한줄 한줄 읽어보면 진짜 별거없습니다!

Level Trigger

LT방식은 어떤 일이 일어났을 때 상태에 따라서 트리거를 지속 시킵니다. 예를 들어서 입력 버퍼가 채워져있는 상태를 1, 아닌 상태를 0으로 놓고, 입력 버퍼가 채워져있는 상태는 계속 1인 상태이기 때문에 이벤트가 지속된다는 뜻입니다.

위의 그림에서도 볼 수 있듯이 1인 상태는 버퍼가 채워져있는 상태로 정의하고, 빨간색 부분에 대해서 지속적으로 이벤트 발생을 알려줍니다.

Edge Trigger

반면 ET방식은 사건이 발생한 시점에 딱 한번 트리거 됩니다. 그러니까 버퍼가 채워지는 그 시점에만 이벤트가 발생한다는 뜻입니다.

이러한 LT, ET 방식인지에 따라서 잠시 후에 나올 epoll_wait 함수가 다음 이벤트까지 기다리는 시점이 달라집니다. 이제 epoll을 다루기 위한 세가지 함수를 소개합니다.

1. epoll_create 

#include <sys/epoll.h>
int epoll_create(int size);

epol l객체를 생성합니다. 커널 2.6.8 버전부터 이 size라는 인자는 무시되며 그냥 0보다 큰 값으로 설정만 해주면 됩니다. 실패시에는 -1을 반환하고 정상 반환일때는 0보다 큰 값이 반환됩니다. 이 반환 값을 파일 디스크립터입니다. 그래서 close로 닫을 수 있습니다. 이렇게 나오는 epoll의 파일 디스크립터는 제어, 이벤트 대기시에 사용이 됩니다.

2. epoll_ctl

#include <sys/epoll.h>
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

epoll_ctl 함수는 epfd라는 epoll 파일 디스크립터에의해 참조되는 관심 리스트(interest list)에 fd를 더할지, 변경할지, 삭제할지를 도와주는 함수입니다. event는 그 fd에 대해서 관심있게 주시할 event를 설정해줍니다. poll과 유사합니다. event에는 아까 소개한 ET 방식도 설정할 수 있습니다. operation을 뜻하는 op는 아래의 표와 같이 세가지가 있습니다.

epoll_event 구조체는 아래와 같이 정의되어있습니다.

typedef union epoll_data {
        void        *ptr;
        int          fd;
        uint32_t     u32;
        uint64_t     u64;
} epoll_data_t;

struct epoll_event {
        uint32_t     events;      /* Epoll events */
        epoll_data_t data;        /* User data variable */
};

events의 설정할 수 있는 이벤트들은 poll과 비슷합니다.

3. epoll_wait

#include <sys/epoll.h>
int epoll_wait(int epfd, struct epoll_event *events,
                      int maxevents, int timeout);

epfd에 대해서 이벤트가 발생함을 기다리는 역할을 합니다. Event에는 발생한 event들이 있습니다. 최대 기다릴 events수를 maxevents에 넣을 수 있습니다. Timeout에 따라서 epoll_wait이 계속 기다릴지, 특정 시간 동안 기다릴지, 바로 반환할 지를 정해줄 수 있습니다.

• timeout = -1 : 이벤트 발생시까지 무한히 대기합니다.
• timeout = 0 : 곧 장 반환합니다.
• timeout > 0 : ms만큼 대기하다가 반환됩니다.

예제 코드 - Level Trigger Epoll

아래는 Level Trigger 방식의 epoll 에코 서버의 예제입니다.

server 

//epoll_server-level.c
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/epoll.h>

#FD_MAX 1024
#define PORT 12346
#define BUF_SIZE 1

const char *welcome_message = "Welcome!\n";

void err_exit(const char *err){
        perror(err);
        exit(1);
}

void clear_fd(const int epoll_fd, const int fd){
        epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
        close(fd);
}
int main(void){
        int socket_fd, accepted_fd;
        struct sockaddr_in host_addr, client_addr;
        socklen_t size;
        int epoll_fd, i, n, ret; 
        char buffer[BUF_SIZE] = {0,};
        struct epoll_event events[FD_MAX];
        int pos = 0;

        //STREAM socket 생성
        socket_fd=socket(PF_INET,SOCK_STREAM,0);
        if(socket_fd < 0) err_exit("socket error ");

        //server ip와 bind하기 위해 주소 설정
        memset(&host_addr, 0, sizeof(host_addr));
        host_addr.sin_family=AF_INET;
        host_addr.sin_port=htons(PORT); 
        host_addr.sin_addr.s_addr = 0;  //자신의 주소로 자동으로 설정 

        //socket과 host_addr과 bind
        if(bind(socket_fd,(struct sockaddr *)&host_addr,sizeof(struct sockaddr)) < 0)
                err_exit("bind error ");

        //접속 대기
        if(listen(socket_fd,3) < 0)
                err_exit("listen error ");

        for(i = 0; i < FD_MAX; i++)
                events[i].data.fd = -1;


        //그냥 0보다 크면 된다.
        epoll_fd = epoll_create(1024); 
        if(epoll_fd < 0) err_exit("epoll_create error ");

        struct epoll_event event;
        event.data.fd = socket_fd;
        event.events = EPOLLIN;

        if(epoll_ctl(epoll_fd, EPOLL_CTL_ADD, socket_fd, &event) < 0)
                err_exit("epoll_ctl error ");

        while(1){

                ret = epoll_wait(epoll_fd, events, FD_MAX, -1);
                //-1은 이벤트가 발생할때까지 무한정 대기
                if(ret == -1) err_exit("epoll_wait error ");
                //ret는 이벤트가 발생한 entry의 갯수, events는 발생한 events의 배열이 저장된다.
                for(i = 0; i < ret; i++){
                        //accept할 것이 있는가?
                        if(events[i].data.fd == socket_fd && events[i].events & EPOLLIN){
                                size = sizeof(struct sockaddr_in);

                                //Client가 connect할때까지 기디린다. 
                                accepted_fd = 
                                        accept(socket_fd,(struct sockaddr *)&client_addr,&size);

                                if(accepted_fd  < 0)
                                        err_exit("accept error ");

                                struct epoll_event client;
                                client.data.fd = accepted_fd;
client.events = EPOLLIN;


                                if(epoll_ctl(epoll_fd, EPOLL_CTL_ADD, accepted_fd, &client) < 0)
                                        err_exit("epoll_ctl error ");
                                //Client의 접속 정보를 출력하고 접속 잘됐다고 메시지 전송
                                printf("Client Info : IP %s, Port %d\n", 
                                                inet_ntoa(client_addr.sin_addr),
                                                ntohs(client_addr.sin_port));

                                n = send(accepted_fd, welcome_message, strlen(welcome_message), 0);
                                if(n < 0) err_exit("send error ");
                                continue;
                        } 

                        if(events[i].events & EPOLLIN){

                                n = recv(events[i].data.fd, &buffer[pos], BUF_SIZE, 0);

                                if(n < 0) err_exit("recv error ");

                                //n == 0일 경우에 epoll_fd 관심 리스트에서 비워준다. 
                                if(n == 0) {
                                        clear_fd(epoll_fd, events[i].data.fd);
                                        pos = 0;
                                        continue;
                                }

                                if(buffer[pos] == '\0'){
                                        printf("rcv msg : %s\n", buffer);
                                        //quit라는 메시지를 받으면 종료
                                        if(!strcmp(buffer,"quit")){
                                                clear_fd(epoll_fd, events[i].data.fd);
                                                pos = 0;
                                                continue;
                                        }

                                        n = send(events[i].data.fd, buffer, pos, 0);
                                        if(n < 0) err_exit("send error ");
                                        pos = 0;
                                } else 
                                        pos++;
                        }

                }

        }
        printf("end\n");
        shutdown(socket_fd, SHUT_RDWR);

        return 0;

}

client

//client.c
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <arpa/inet.h> 

#define BUF_SIZE 128

void err_exit(const char *err){
        perror(err);
        exit(1);
}

int main(int argc, char *argv[])
{
        int sockfd = 0, n = 0;
        uint16_t port;
        char buffer[BUF_SIZE] = {0,};
        struct sockaddr_in server_addr; 

        if(argc != 3){
                printf("\n Usage: %s server_ip port \n",argv[0]);
                return 1;
        } 

        port = atoi(argv[2]);

        //Internet용 socket을 연다.
        if((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
                err_exit("socket error ");

        //server_addr의 구조체를 깔끔히 0으로 도배 
        memset(&server_addr, 0, sizeof(server_addr)); 
        server_addr.sin_family = AF_INET;
        server_addr.sin_port = htons(port); 
        //127.0.0.1은 본인의 IP이다. 
        inet_aton(argv[1], (struct in_addr*)&server_addr.sin_addr);

        //client는 서버의 ip, port, protocol 설정후 connect로 서버에게
        //바로 연결한다.
        if(connect(sockfd, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0)
                err_exit("connect error ");


        while(1){
                //서버로부터 에코된 메시지를 받음
                n = recv(sockfd, buffer, BUF_SIZE, 0);

                if(n < 0) err_exit("recv error ");

                printf("%s\n", buffer);

                while(1){
                        printf("message:");
                        fgets(buffer, BUF_SIZE, stdin);
                        if(strlen(buffer) >= 100)
                                printf("message too large\n");
                        else break;

                };
                // \n을 \0로 처리 
                buffer[strlen(buffer)-1] = '\0';

                //서버에게 메시지 전송
                n = send(sockfd, buffer, strlen(buffer)+1, 0);
                if(n < 0) err_exit("send error ");

        }

        shutdown(sockfd, SHUT_RDWR);
        return 0;
}

Level Trigger 방식의 결과화면입니다.

예제 코드 - Edge Trigger Epoll

위 LT방식은 ET방식으로 바꿔볼까요? event에다가 EPOLLET를 추가하시면 ET방식으로 동작하게 됩니다. accepted_fd를 EPOLLET event를 같이 추가해서 실행해보세요.

struct epoll_event client;
client.data.fd = accepted_fd;
client.events = EPOLLIN | EPOLLET;

그 후 실행하게 되면 에코가 되지 않습니다.

왜 이런 결과가 나올까요?

우리가 구현한 방식은 한 글자씩 버퍼에서 읽어옵니다. 이는 epoll_wait이 반환된 후에 읽을 수가 있죠. LT방식일때는 버퍼에 데이터가 남아있으면 epoll_wait에 의해서 반환되게 됩니다. 하지만 ET방식은 버퍼가 채워지는 이벤트 순간 한번만 epoll_wait이 반환되기 때문에 버퍼에서 한글자만 읽고 대기하게 되는 겁니다.

그럼 여기서 질문, select와 poll은 어떤 방식을 쓰고 있는 걸까요?

우리가 현상을 보게 되면 버퍼가 남이있다면 select와 poll은 return됩니다. 이런 결과를 봐서 우리는 select와 poll은 LT방식인 것을 알 수 있습니다. 이러한 문제를 해결하기 위해서는 non-blocking 방식의 read를 해야합니다.즉, 더 이상 버퍼에 남아있는 데이터가 없어서 오류를 발생할때까지 읽어야합니다. 이때 남아있는 데이터가 없으면 EAGAIN 에러를 발생하게 됩니다. 그래서 이런 방식으로 수정해야합니다.

• 비차단 모드로 설정 

//NON-BLOCKING 모드로 전환
int flags = fcntl(accepted_fd, F_GETFL);
flags |= O_NONBLOCK;
if(fcntl(accepted_fd, F_SETFL, flags) < 0) 
    err_exit("fcntl error ");

• read에서 EAGAIN 오류를 만날때까지 읽기

while(1){
    n = recv(events[i].data.fd, &buffer[pos], BUF_SIZE, 0);

    if(n < 0) 
        if(errno == EAGAIN) break;
    if(n == 0) {
        clear_fd(epoll_fd, events[i].data.fd);
        pos = 0;
        break;
    }

    pos++;
}

전체 풀 소스코드는 아래와 같습니다. 

//epoll_server-edge.c
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/epoll.h>
#include <errno.h>
#include <fcntl.h>

#define PORT 12346
#define BUF_SIZE 1

const char *welcome_message = "Welcome!\n";

void err_exit(const char *err){
        perror(err);
        exit(1);
}

void clear_fd(const int epoll_fd, const int fd){
        epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, NULL);
        close(fd);
}
int main(void){
        int socket_fd, accepted_fd;
        struct sockaddr_in host_addr, client_addr;
        socklen_t size;
        int epoll_fd, i, n, ret; 
        char buffer[BUF_SIZE] = {0,};
        struct epoll_event events[FD_MAX];
        int pos = 0;

        //STREAM socket 생성
        socket_fd=socket(PF_INET,SOCK_STREAM,0);
        if(socket_fd < 0) err_exit("socket error ");

        //server ip와 bind하기 위해 주소 설정
        memset(&host_addr, 0, sizeof(host_addr));
        host_addr.sin_family=AF_INET;
        host_addr.sin_port=htons(PORT); 
        host_addr.sin_addr.s_addr = 0;  //자신의 주소로 자동으로 설정 

        //socket과 host_addr과 bind
        if(bind(socket_fd,(struct sockaddr *)&host_addr,sizeof(struct sockaddr)) < 0)
                err_exit("bind error ");

        //접속 대기
        if(listen(socket_fd,3) < 0)
                err_exit("listen error ");

        for(i = 0; i < FD_MAX; i++)
                events[i].data.fd = -1;


        //그냥 0보다 크면 된다.
        epoll_fd = epoll_create(1024); 
        if(epoll_fd < 0) err_exit("epoll_create error ");

        struct epoll_event event;
        event.data.fd = socket_fd;
        event.events = EPOLLIN;
        //event.events = EPOLLIN | EPOLLET;

        if(epoll_ctl(epoll_fd, EPOLL_CTL_ADD, socket_fd, &event) < 0)
                err_exit("epoll_ctl error ");

        while(1){

                ret = epoll_wait(epoll_fd, events, FD_MAX, -1);
                //-1은 이벤트가 발생할때까지 무한정 대기
                if(ret == -1) err_exit("epoll_wait error ");
                //ret는 이벤트가 발생한 entry의 갯수, events는 발생한 events의 배열이 저장된다.
                for(i = 0; i < ret; i++){
                        //accept할 것이 있는가?
                        if(events[i].data.fd == socket_fd && events[i].events & EPOLLIN){
                                size = sizeof(struct sockaddr_in);

                                //Client가 connect할때까지 기디린다. 
                                accepted_fd = 
                                        accept(socket_fd,(struct sockaddr *)&client_addr,&size);

                                if(accepted_fd  < 0)
                                        err_exit("accept error ");

                                struct epoll_event client;
                                client.data.fd = accepted_fd;
                                client.events = EPOLLIN | EPOLLET;

                                //NON-BLOCKING 모드로 전환
                                int flags = fcntl(accepted_fd, F_GETFL);
                                flags |= O_NONBLOCK;
                                if(fcntl(accepted_fd, F_SETFL, flags) < 0) 
                                        err_exit("fcntl error ");


                                if(epoll_ctl(epoll_fd, EPOLL_CTL_ADD, accepted_fd, &client) < 0)
                                        err_exit("epoll_ctl error ");
                                //Client의 접속 정보를 출력하고 접속 잘됐다고 메시지 전송
                                printf("Client Info : IP %s, Port %d\n", 
                                                inet_ntoa(client_addr.sin_addr),
                                                ntohs(client_addr.sin_port));

                                n = send(accepted_fd, welcome_message, strlen(welcome_message), 0);
                                if(n < 0) err_exit("send error ");
                                continue;
                        } 

                        if(events[i].events & EPOLLIN){
                                while(1){
                                        n = recv(events[i].data.fd, &buffer[pos], BUF_SIZE, 0);

                                        if(n < 0) 
                                                if(errno == EAGAIN) break;
                                        if(n == 0) {
                                                clear_fd(epoll_fd, events[i].data.fd);
                                                pos = 0;
                                                break;
                                        }

                                        pos++;
                                }

                                //quit라는 메시지를 받으면 종료
                                if(!strcmp(buffer,"quit") || n == 0){
                                        clear_fd(epoll_fd, events[i].data.fd);
                                        pos = 0;
                                        continue;
                                }
                                printf("rcv msg : %s\n", buffer);

                                n = send(events[i].data.fd, buffer, pos, 0);
                                if(n < 0) err_exit("send error ");
                                pos = 0;
                        }

                }

        }
        printf("end\n");
        shutdown(socket_fd, SHUT_RDWR);

        return 0;

}

이상으로 epoll에 관한 설명을 마칩니다.