Libev源码分析09：select突破处理描述符个数的限制

        众所周知，Linux下的多路复用函数select采用描述符集表示处理的描述符。描述符集的大小就是它所能处理的最大描述符限制。通常情况下该值为1024，等同于每个进程所能打开的描述符个数。

        增大描述符集大小的唯一方法是先增大FD_SETSIZE的值，然后重新编译内核，不重新编译内核而改变其值时不够的。

         

        在阅读Libev源码时，发现它实现了一种突破这种限制的方法。该方法本质上而言，就是自定义fd_set结构，以及FD_SET，FD_CLR，FD_ISSET宏。

        首先看一下Linux中原fd_set结构的实现细节，我的虚拟机的系统版本是Ubuntu 14.10，内核为3.16.0-23-generic。在其中的/usr/include/i386-linux-gnu/目录下的<sys/select.h>、<bits/select.h>、<bits/typesizes.h>文件中找到了fd_set结构的实现细节，代码如下：

//<bits/typesizes.h>
#define __FD_SETSIZE        1024

//<sys/select.h>
typedef long int __fd_mask;
typedef __fd_mask fd_mask;

#define FD_SETSIZE      __FD_SETSIZE

#define __NFDBITS   (8 * (int) sizeof (__fd_mask))
#define NFDBITS     __NFDBITS

#define __FD_ELT(d) ((d) / __NFDBITS)
#define __FD_MASK(d)    ((__fd_mask) 1 << ((d) % __NFDBITS))

typedef struct
  {
    /* XPG4.2 requires this member name.  Otherwise avoid the name
       from the global namespace.  */
#ifdef __USE_XOPEN
    __fd_mask fds_bits[__FD_SETSIZE / __NFDBITS];
# define __FDS_BITS(set) ((set)->fds_bits)
#else
    __fd_mask __fds_bits[__FD_SETSIZE / __NFDBITS];
# define __FDS_BITS(set) ((set)->__fds_bits)
#endif
  } fd_set;


//<bits/select.h>
#define __FD_SET(d, set) 
  ((void) (__FDS_BITS (set)[__FD_ELT (d)] |= __FD_MASK (d)))
#define __FD_CLR(d, set) 
  ((void) (__FDS_BITS (set)[__FD_ELT (d)] &= ~__FD_MASK (d)))
#define __FD_ISSET(d, set) 
  ((__FDS_BITS (set)[__FD_ELT (d)] & __FD_MASK (d)) != 0)

#if defined __GNUC__ && __GNUC__ >= 2
# define __FD_ZERO(fdsp) 
  do {                                        
    int __d0, __d1;                               
    __asm__ __volatile__ ("cld; rep; " __FD_ZERO_STOS                 
              : "=c" (__d0), "=D" (__d1)                  
              : "a" (0), "0" (sizeof (fd_set)             
                      / sizeof (__fd_mask)),          
                "1" (&__FDS_BITS (fdsp)[0])               
              : "memory");                        
  } while (0)

#else   /* ! GNU CC */

# define __FD_ZERO(set)  
  do {                                        
    unsigned int __i;                                 
    fd_set *__arr = (set);                            
    for (__i = 0; __i < sizeof (fd_set) / sizeof (__fd_mask); ++__i)          
      __FDS_BITS (__arr)[__i] = 0;                        
  } while (0)

#endif  /* GNU CC */
  
//<sys/select.h>  
#define FD_SET(fd, fdsetp)  __FD_SET (fd, fdsetp)
#define FD_CLR(fd, fdsetp)  __FD_CLR (fd, fdsetp)
#define FD_ISSET(fd, fdsetp)    __FD_ISSET (fd, fdsetp)
#define FD_ZERO(fdsetp)     __FD_ZERO (fdsetp)

        代码比较简单，就不一一分析了，总结一下就是：fd_set是个结构体，它内部包含一个long  int类型的数组，数组长度为__FD_SETSIZE /__NFDBITS。也就是说该数组一共包含__FD_SETSIZE个位。每一位就代表一个描述符。

        FD_SET就是将该数组相应的位置1，FD_CLR就是将该数组相应的位置0，FD_ISSET就是判断某位是否为1.

        以上就是fd_set的具体实现。Libev采用的方法是动态申请fd_set的空间，并实现相应的宏。下面是模仿Libev的代码，实现的一个简单的echo服务器：

# define NFDBYTES (NFDBITS / 8)

#define IFERR(res, msg)             
   if(res < 0)                      
   {                                
        perror(#msg " error");      
        return;                     
   }                        

#define MYFDSET(fd)         
    word = fd / NFDBITS;            
    mask = 1UL << (fd % NFDBITS);   
    if(word+1 > fdsize)                 
    {                                   
        roset = realloc(roset, (word+1) * NFDBYTES);    
        rset = realloc(rset, (word+1) * NFDBYTES);      
        for(; fdsize < word+1; fdsize++)                
        {                                               
            ((fd_mask *)rset)[fdsize] = ((fd_mask *)roset)[fdsize] = 0;   
        }                                               
    }                                                   
    ((fd_mask *)roset)[word] |= mask;       


#define MYFDCLR(fd)                 
    word = fd / NFDBITS;            
    mask = 1UL << (fd % NFDBITS);   
    ((fd_mask *)roset)[word] &= ~mask;
    

void echoserver()
{
    int listenfd = -1, confd = -1;
    int res = -1;
    int val = 1;
    int i;

    char rbuf[1024];
    char sbuf[1024];
    
    void *roset = NULL;
    void *rset = NULL;
    int fdsize = 0;

    int word;
    fd_mask mask;
    
    struct sockaddr_in serveraddr;
    int addrlen = sizeof(serveraddr);

    listenfd = socket(AF_INET, SOCK_STREAM, 0);
    IFERR(listenfd, socket);

    serveraddr.sin_family = AF_INET;
    serveraddr.sin_port = htons(8898);
    serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);

    res = setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, (void *)&val, sizeof(val));
    IFERR(res, setsockopt);

    res = bind(listenfd, (struct sockaddr *)&serveraddr, addrlen);
    IFERR(res, bind);

    MYFDSET(listenfd);

    res = listen(listenfd, 5);
    IFERR(res, "listen error")

    for(;;)
    {
        memcpy(rset, roset, fdsize*NFDBYTES);
        
        res = select(fdsize * NFDBITS, (fd_set *)rset, NULL, NULL, NULL);
        IFERR(res, select)

        for(i = fdsize-1; i >= 0; i--)
        {
            if(((fd_mask *)rset)[i] == 0)   continue;

            int bit = 0;
            fd_mask bitmask = 0;
            int fd = 0;
            for(bit = NFDBITS-1; bit >= 0; bit--)
            {
                bitmask = 1UL << bit;
                if(((fd_mask *)rset)[i] & bitmask)
                {
                    fd = i*NFDBITS + bit;
                    if(fd == listenfd)
                    {
                        confd = accept(listenfd, NULL, NULL);
                        IFERR(confd, accept)
                        MYFDSET(confd)
                    }
                    else
                    {
                        bzero(rbuf, 1024);
                        res = read(fd, rbuf, 1024);
                        if(res <= 0)
                        {
                            if(res < 0)perror("read error");
                            else    printf("client over
");
                            
                            MYFDCLR(fd);
                            close(fd);
                        }
                        else
                        {
                            snprintf(sbuf, 1024, "server echo: %s", rbuf);
                            write(fd ,sbuf, strlen(sbuf));
                        }
                    }
                }
            }       
        }
    }
}

        该代码中，rset和roset是表示描述符集的动态数组，fdsize * NFDBITS就是该描述符集所能表示的最大位数。当需要添加新的描述符时，如果空间不够，则调用realloc动态增加。并置相应的位为1。除了动态增长之外，MYFDSET和MYFDCLR实现思路与原来的FD_SET和FD_CLR一样的。

 

        下面是简单的测试代码，用python实现的TCP客户端：

from socket import *
import sys
from time import sleep


serverAddr = ('localhost', 8898)

clientlist = []
errlist = []
clientcnts = int(sys.argv[1])

for i in xrange(clientcnts):
    clientlist.insert(i, socket(AF_INET, SOCK_STREAM))

for index, client in enumerate(clientlist):
    try:
        client.connect(serverAddr) 
    except Exception as e: 
        print 'exception catched : ' ,e
        errlist.append(client)

for i in errlist:
    clientlist.remove(i)
    
for index, client in enumerate(clientlist):
    data = "this is %d client"%index
    client.send(data)
    data = client.recv(100)
    print 'recieve : ', data  

sleep(10)

for index, client in enumerate(clientlist):
    client.close()

        测试时，运行如下命令即可，3000表示并发量。

# python tcpclient.py 3000

 

        当然，在运行服务端和客户端之前，需要增加每个进程所能打开的描述符的限制，否则的话，客户端会出现错误：socket.error: [Errno 24] Too many open files，服务端会出现错误：accept error: Too many open files。   增加限制的命令如下：

# ulimit -n 65535

 

        总结： 虽然select的最大描述符限制是可以突破的，但是与poll或者epoll相比，select在处理大量描述符时依然性能有限。select的优势在于它的跨平台特性；以及在处理少量描述符时，它可能是最快的；而且如果多数的描述符都处于活跃状态的话，它的性能也会非常好。

 

参考：

http://pod.tst.eu/http://cvs.schmorp.de/libev/ev.pod

http://codemacro.com/2014/06/01/select-limit/