Select-based Non-blocking I/O Processing

I/O처리의 blocking 문제를 해결하는 방법으로 select() 함수를 고려할 수 있다. select() 함수는 여러개의 file descriptor들을 관찰하여 읽을 데이타가 준비되어 있는지, 쓸수 있는 상황인지, 에러가 발생한 상황인지를 검사하여 관찰 대상들에게서 변화가 발생하면 그 file descriptor들이 무엇인지 리턴하게 된다. 만일 관찰 대상으로부터 변화가 없다면 무한정 blocking 상태가 될 수 있는데, 이때에 timeout을 지정해 둠으로써 특별한 변화가 발생하지 않는다고 하여도 무조건 리턴하게금 설정할 수 있다. 결론적으로 select() 함수를 사용함으로써 특정 file descriptor들을 주기적으로 관찰할 수 있으며, select()함수 리턴후에 변화가 발생한 file descriptor들에 대해서만 지정된 작업을 수행할 수 있고, 이후에 반복해서 select를 사용하는 방식으로 non-blocking I/O processing이 가능케 된다.

select() 함수 사용하기

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

exceptfds 의 경우는 일반적인 예외상황을 말하는 것이 아니라, TCP기반의 out-of-band 데이타의 수신 여부를 말한다는 것에 주의하라.

File Descriptor 설정

select() 함수는 file descriptor들을 동시에 관찰할 수 있다. 여기에서 관찰할 대상은 위 함수 프로토타입에서 볼 수 있듯이, read, write, execpt 의 세가지 부류가 대상이 된다. 특별히 관찰할 대상을 설정하는 데이터 타입으로 fd_set 이라는 놈을 사용하는데, 이놈은 bit 단위로 file descriptor들을 인식한다. 즉, fd_set 은 bit들의 array로써 표현될 수 있고, 첫번째 bit가 1로 세팅되어 있다면, file descriptor 0 번이 관찰대상이라는 뜻이 된다. 다음은 fd_set을 쉽게 다루기 위한 macro 함수들이다.

FD_ZERO (fd_set *fdset);	fdset 포인터가 가리키는 변수의 모든 비트들을 0으로 초기화한다.
FD_SET (int fd, fd_set *fdset);	fdset 포인터가 가리키는 변수에 fd로 전달되는 파일 디스크립터 정보를 설정한다.
FD_CLR (int fd, fd_set *fdset);	fdset 포인터가 가리키는 변수에서 fd로 전달되는 파일 디스크립터 정보를 삭제한다.
FD_ISSET (int fd, fd_set *fdset);	fdset 포인터가 가리키는 변수가 fd로 전달되는 파일 디스크립터 정보를 지니고 있는지 확인한다.

관찰 대상의 범위 설정

사실 select() 함수의 가장 큰 문제는 바로 이 부분이다. select() 함수의 첫번째 파라미터인 n 은 바로 검사 대상이 되는 file descriptor들의 관찰 범위의 최대값이다. 즉, 만일 관찰하고자 하는 file descriptor가 5 라면, select()는 0~5번까지를 모조리 관찰 대상으로 인식하게 되고, 따라서 n = 6 (0~5) 이 될 수 밖에 없다. 관찰할 필요가 없는 5개(0~4)의 file descriptor들도 관찰의 대상이 된다는 것이다. 물론 fd_set 자체가 bit-wise data set 이기 때문에 효율문제에서 큰 걱정을 하진 않는다 하더라도, select() 함수는 기본적으로 stateless 함수로써, 매번 호출될 때마다 kernel 영역과 user 영역 사이에 모든 file descriptor 정보 리스트와 부가적인 정보를 복사해야 하고, 미리 언급한 것처럼 마지막 file descriptor의 번호가 큰 경우라면 당연히 overhead가 발생할 수 밖에 없는 문제를 가지고 있다. 이 문제를 해결하기 위하여 poll 방식을 사용할 수도 있고, epoll/kqueue 를 사용할 수도 있다.

Timeout 설정

select() 함수 역시 관찰 대상으로부터 변화가 발생하지 않으면 역시 blocking 상태에 빠지게 된다. 따라서 non-blocking I/O processing을 지원한다는 것이 뻥이 된다. 그러나 select() 함수는 마지막 파라미터로 timeout을 지정할 수 있기 때문에 변화가 발생하지 않는다 하여도 무조건 리턴함으로써 다른 처리를 할 수 있게 된다. timeout 설정에 사용되고 있는 struct 는 다음과 같다.

 
struct timeval
{
   long tv_sec;   /*seconds */
   long tv_usec;   /* microseconds */
}

예를 들어, 3.5초를 지정하려면, tv_sec = 3, tv_usec = 500000 을 지정하면 된다. 만일 무한정 기다리기를 원한다면 NULL 값을 지정하면 된다.

함수 호출후 Return 처리

select() 함수가 정상적으로 리턴되었다는 것은 딱 두가지 경우중 하나이다. 하나는 관찰 대상인 file descriptor들의 변화가 발생했다는 것이고, 또 하나는 timeout이 발생하여 무조건 리턴된 경우이다. 그러나 만일 비정상적인 리턴이었다면 그것은 select() 함수의 오류를 말하는 것이다. 따라서 이러한 함수 호출후 리턴값에 따른 처리가 필요하다.

• -1 : 오류 발생
• 0 : timeout
• 0 보다 큰 수 : 변화가 발생한 file descriptor들의 수

결과적으로 select() 함수를 호출후 각 관찰 대상 readfds, writefds, exceptfds을 점검하여 1로 세팅된 bit가 있다면 그 위치의 file descriptor에 변화가 있었다는 뜻이 된다.

select() 함수 호출 예제

#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/time.h>

#define BUFSIZE 30

int main(int argc, char **argv)
{
   fd_set reads, temps;
   int result;

   char message[BUFSIZE];
   int str_len;
   struct timeval timeout;

   FD_ZERO(&reads);
   FD_SET(0, &reads);  // standard input 설정

   /* 여기에 설정해 버리면 select()함수가 정상리턴된후 다시 호출될 때에 remained time이 사용되는 문제가 발생한다.
   timeout.tv_sec = 5;
   timeout.tv_usec = 100000;
   */

   while(1)
   {
      temps = reads;
      timeout.tv_sec = 5;
      timeout.tv_usec = 0;

      result = select(1, &temps, 0, 0, &timeout);
      if(result==-1) { // 오류
         puts("select 오류 발생");
         exit(-1);
      }
      else if(result==0) { // timeout
         puts("시간이 초과되었습니다 : select ");
      }
      else { // 변경사항 발생
         if(FD_ISSET(0, &temps)) {
            str_len = read(0, message, BUFSIZE);
            message[str_len] = 0;
            fputs(message, stdout);
         }
      }
   }

Select-based Multiplexing Server Programming

이제 select() 함수를 이용하여 어떻게 socket 프로그래밍에서 multiplexing server를 구현할 수 있는 지를 보자. socket 자체도 file descriptor 이기 때문에 select를 사용하여 실제로 read, write, except 관련 상황의 관찰이 가능하다. 아래의 내용은 내가 대학원 시절에 시간이 남아돌아(?) 3가지 방식의 서버 프로그래밍 기법을 영어로 작성한 것이다. 아래의 내용은 Server Programming For Multi-Users에서 multiplexing 관련 내용만 포함한 것이고, multiprocessing이나 multithreading server 구현에도 관심이 있다면 원문(http://blog.naver.com/wisereign/30023041919)을 참조하기 바란다.

Multiplexing Server Programming

Now, the next way is the very multiplexing method. The word “multiplex” means to put many things just into one thing. That is, as you see the previous example, the multiprocessing server needs as many cloned-processes forked as numbers of clients to communicate to them. In result, the resources like cpu and memory might be consumed more than we can expect in some cases. With the above reason, I’ll introduce a way to keep from wasting server-side resources by means of using select () function. Also, this way’s compiled and executed successfully on both Windows and Linux machine. So, you can use it on any environments without change in the code. Just make sure that you need to keep some lists of clients to manage all accesses to the server efficiently and effectively. You might make client class to include some information like ip, port, socket descriptor, client alias, involved group, access time – because you’d need to deal with some abnormal client socket like no response – and buffer-related information like size and data itself. In addition, I haven’t used the expanded Windows socket functions in this example. And I haven’t given you knowledge about nonblocking socket. I’ve just tried to suggest a simple instance to you and the above things are up to you.

Source Code

//----------------------------------------------
// Multiplexing Server using select()
// ------------------------
// Author : Woo-Hyun, Kim
// Date : Feb. 12, 2004
// Email : woorung@empal.com             
//----------------------------------------------
#include <iostream.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
//----------------------------------------------
#ifdef _WIN32  // on Windows
#include <winsock2.h>
#else  // on Linux
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <sys/wait.h>
#include <sys/time.h> // for using struct timeval as the timeout parameter of select();
#include <signal.h>
#endif
//----------------------------------------------
// for half-close like shutdown() on Windows
#ifdef _WIN32
#define SHUT_RD SD_RECEIVE
#define SHUT_WR SD_SEND
#define SHUT_RDWR SD_BOTH
#endif
//-----------------------------------------------
#define WAIT_QUEUE 10
#define SRV_PORT "2020"
#define BUF_SIZE 1024
//-----------------------------------------------
void connected_proc(int clnt_sock,struct sockaddr_in *pclnt_addr)
{
   cout << "connected from " << inet_ntoa(pclnt_addr->sin_addr) << endl;

   // write your code you want to do just after connected from a client socket
   // ...
}
//----------------------------------------------
int read_proc(int clnt_sock)
{
   int recv_len;
   char recv_buf[BUF_SIZE];
   
   recv_len= recv(clnt_sock,recv_buf,BUF_SIZE,0);
   if(recv_len==0) return -1; // in case of EOF, that is, close from the client socket

   // write your code about the data received from a client socket
   // ...
   send(clnt_sock,recv_buf,recv_len,0);

   return 0;
}
//----------------------------------------------
int main(int argc,char *argv[])
{
#ifdef _WIN32
   SOCKET srv_sock;   
#else
   int srv_sock;
#endif
   struct sockaddr_in srv_addr;

#ifdef _WIN32
   WSADATA wsaData;

   // for linking ws2_32.lib on Windows
   if(WSAStartup(0x0202,&wsaData)!=0) {
      cout << "WSAStartup() error" << endl;
      exit(-1);
   }
#endif

#ifdef _WIN32
   SOCKET max_fd;
#else
   int max_fd;
#endif
   struct timeval timeout;
   fd_set read_set, read_tset;
   fd_set write_set, write_tset;
   fd_set urgent_set, urgent_tset;

   srv_sock = socket(AF_INET,SOCK_STREAM,0);
   if(srv_sock<0) {
      cout << "socket() error" << endl;
#ifdef _WIN32
      // for releasing ws2_32.lib on Windows
      WSACleanup();
#endif
      exit(-1);
   }

   memset(&srv_addr,0,sizeof(srv_addr));
   srv_addr.sin_family = AF_INET;
   srv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
   srv_addr.sin_port = htons(atoi(SRV_PORT));

   if(bind(srv_sock,(struct sockaddr *)&srv_addr,sizeof(srv_addr))<0) {
      cout << "bind() error" << endl;
#ifdef _WIN32
      closesocket(srv_sock);
      // for releasing ws2_32.lib on Windows
      WSACleanup();
#else
      close(srv_sock);
#endif
      exit(-1);
   }

   if(listen(srv_sock,WAIT_QUEUE)<0) {
      cout << "listen() error" << endl;
#ifdef _WIN32
      closesocket(srv_sock);
      // for releasing ws2_32.lib on Windows
      WSACleanup();
#else
      close(srv_sock);
#endif
      exit(-1);
   }

   // initialize fd_set datatype
   FD_ZERO(&read_set);
   FD_ZERO(&write_set);
   FD_ZERO(&urgent_set);
   
   // set the server socket inspected by means of using select()
   FD_SET(srv_sock,&read_set);
   max_fd = srv_sock;

   // finally, select() get started
   while(1) {
      read_tset = read_set;
      write_tset = write_set;
      urgent_tset = urgent_set;

      // set timeout when selecting
      timeout.tv_sec = 5;
      timeout.tv_usec = 0;


      if(select(max_fd+1,&read_tset,&write_tset,&urgent_tset,&timeout)<0) {
         cout << "select() error" << endl;
#ifdef _WIN32
         closesocket(srv_sock);
         // for releasing ws2_32.lib on Windows
         WSACleanup();
#else
         close(srv_sock);
#endif
         exit(-1);
      }

      // deal with sockets changed by select()
#ifdef _WIN32
      SOCKET fd;
#else
      int fd;
#endif
      
      for(fd=0;fd<max_fd+1;fd++) {
         if(FD_ISSET(fd,&read_tset)) {
            if(fd==srv_sock) { // happen when there is a client socekt accepted from the server socket
#ifdef _WIN32
               SOCKET clnt_sock;
#else
               int clnt_sock;
#endif
               struct sockaddr_in clnt_addr;
               int clnt_len = sizeof(clnt_addr);
               
               if((clnt_sock=accept(srv_sock,(struct sockaddr *)&clnt_addr,&clnt_len))<0) {
                  cout << "accept() error" << endl;
#ifdef _WIN32
                  closesocket(srv_sock);
                  // for releasing ws2_32.lib on Windows
                  WSACleanup();
#else
                  close(srv_sock);
#endif
                  exit(-1);
               }

               // ready to select the new client socket accepted by the server socket
               FD_SET(clnt_sock,&read_set);
               FD_SET(clnt_sock,&write_set);
               FD_SET(clnt_sock,&urgent_set);

               // set max_fd newly to specifiy the range of selecting
               if(max_fd<clnt_sock) max_fd = clnt_sock;

               // process something after connected to a client socket
               connected_proc(clnt_sock,&clnt_addr);
            }
            else { // happen when there are some data to read from a client socket
               cout << "read : " << fd << endl;
               if(read_proc(fd)<0) {
                  // the data received from a client socket were a request to close the socket
                  // therefore, delete the client socket from read_set
                  FD_CLR(fd,&read_set);
                  FD_CLR(fd,&write_set);
                  FD_CLR(fd,&urgent_set);
#ifdef _WIN32
                  closesocket(fd);
#else
                  close(fd);
#endif
               }
            }
         }
         else if(FD_ISSET(fd,&write_tset)) {
         
         }
         else if(FD_ISSET(fd,&urgent_tset)) {
         
         }         
      }
   }

#ifdef _WIN32
   closesocket(srv_sock);
   // for releasing ws2_32.lib on Windows
   WSACleanup();
#else
   close(srv_sock);
#endif
   return 0;
}

-----------------
김우현(woorung)

NHN





출처 : http://itbiz.tistory.com/407
recv(fd, 어쩌구... );
==>
nonblock( fd, 1); // 논블록모드로 세팅
recv(fd, 
어쩌구... ); // 반드시 리턴값을 검사해야 함!!!! 매우 중요함!!
// 논블록시의 리턴값과 블록시의 리턴값이 
상이함!
nonblock( fd, o); // 다시 블록모드로 세팅
의 형태로 하시면 될듯 합니다.
당연하게도 리턴값 검사는 
하셔야 합니다. 위에 써드린 코드는 대략의 pseudo code 일뿐.
* nonblock으로 검색해 보면 이미 많은 구현이 있습니다.
운영체제 버전별로 다소 차이가 있으나, 대충 정리하면 아래와 
같습니다.
골라서 테스트해보시고 쓰시길....
// 최신버전의 유닉스 : 대부분
int nonblock(int fd, int nblockFlag)
{
int 
flags;
flags = fcntl( fd, F_GETFL, 0);
if ( nblockFlag == 1 )
return fcntl( 
fd, F_SETFL, flags | O_NONBLOCK);
else
return fcntl( fd, F_SETFL, flags 
& (~O_NONBLOCK));
}
// 오래된 버전의 유닉스들
int nonblock(int fd, int nblockFlag)
{
int 
flags;
flags = nblockFlag;
return ioctl( fd, FIONBIO, &flags);
}
// 윈도우
int nonblock(int fd, int nblockFlag)
{
unsigned long 
flags;
flags = nblockFlag;
return ioctlsocket( fd, FIONBIO, &flags);
}
// Amiga
int nonblock(int fd, int nblockFlag)
{
return IoctlSocket( 
fd, FIONBIO, (long)nblockFlag);
}
// BeOS
int nonblock(int fd, int nblockFlag)
{
long b = nblockFlag ? 
1 : 0;
return setsockopt(sockfd, SOL_SOCKET,SO_NONBLOCK,&b,sizeof(b));
}