先说说内核的职责

我们已经知道了所有的io操作都是交给内核去处理了，在linux中，已经抽象出了一个文件系统，对任何io设备的读写都可以当做对文件系统的某一个文件进行读写。文件是一个抽象出来的概念（它包含了实际对应的驱动，当前文件指针，文件大小，数据读写缓冲区指针等信息），当用户程序需要读写一个文件时，需要先调用sys_open，这样内核会从文件系统读取该文件的节点信息，每个进程都有一个fd数组，内核会给fd数组增加一条记录，然后返回给用户程序这个新增的数组下标。所以我们平时用户程序拿到的fd其实仅仅是内核中为我们打开的fd数组中的下标，我们是对文件的细节一无所知的。剩下我们读写的时候，会传入fd，这样内核就能从数组中读取对应的节点对象，从中取出缓冲区等信息，当然了内核也是分为很多模块的，除了驱动层会直接跟真正的外设打交道，其他模块都是对缓冲区进行操作的。

内核本就是异步的

内核层对文件的处理本就是异步进行的，读的时候会告诉驱动程序读取的扇面号（比如是读写磁盘）和需要读取的缓冲区地址，驱动程序向文件控制模块发送指令后就干别的了，当硬件完成了工作后会向cpu发送中断信号从而被内核捕获，内核会从中断点开始继续执行。

用户程序为什么需要异步

看早期的linux代码，会发现还没加入非堵塞io，也就是说用户进程读写一个fd的时候，虽然内核是异步的，但内核却会把进程给睡眠掉，而当中断返回后才唤醒进程（其实只有读，写的话非严格模式时内核只是把数据从用户空间搬到了文件缓冲区中并把缓冲区设置为脏，内核会在未来某个时候才去调用驱动真正写入磁盘）。

但这样就对用户程序有很多限制了，有很多程序并不是计算型的，而是io型的，比如一个记事本程序，它最主要的工作就是坚挺鼠标和键盘的输入，如果只有堵塞模型，会有很大的困难。

select的出现

所以这时候出现select也就是一件很自然的事情了，如果我们自己假设是内核开发人员，是不是也能大概猜到一些细节了。是的，内核本来就是异步的，以前进程只能监听一个fd，那我们就增加一个委托人帮我们监听多个fd，进程把需要监听的fd交给我们的委托人，委托人去监听这些fd，当任何一个fd可以读写的时候委托人就跑过来唤醒我们的进程。没错，这个委托人就是今天的主角select。

源码

源码是1.0版本的，虽然很老，但思想都是差不多的，最新版本的模块化啥的看起来如果对内核不熟悉很难看（其实就是因为我菜哈哈）

首先看入口方法sys_select

asmlinkage int sys_select( unsigned long *buffer )
{
// 还记得c方法中select的参数吗
// 第一个是最大fd值，第二个是监听的读fd_set，第三个是监听的写fd_set，第四个是错误fd_set，最后一个是超时 
// 这儿参数只有一个，其实是一种变参，C中局部变量是放在栈中的，按从右到左的顺序入栈
// 所以拿到第一个参数，就能依次获取剩下的参数了
// get_fs_long方法，linux中进入内核态使用的段是内核段，但内核会把fs寄存器存入用户段选择符，所以
// get_fs_long方法就是从用户内存空间读数据到内核空间
    int i;
    fd_set res_in, in, *inp;
    fd_set res_out, out, *outp;
    fd_set res_ex, ex, *exp;
    int n;
    struct timeval *tvp;
    unsigned long timeout;

// 内存检测
    i = verify_area(VERIFY_READ, buffer, 20);
    if (i)
        return i;
// 这就是取第一个参数最大fd值n，并且把buffer指针加一，让它指向第二个参数
    n = get_fs_long(buffer++);
// 对n进行校验，可以看到这儿判断了n的最大值不能超过NR_OPEN，这也就是网上很多说select有数量限制的原因，这个后面细说
    if (n < 0)
        return -EINVAL;
    if (n > NR_OPEN)
        n = NR_OPEN;
// 依次读出用户程序的参数
    inp = (fd_set *) get_fs_long(buffer++);
    outp = (fd_set *) get_fs_long(buffer++);
    exp = (fd_set *) get_fs_long(buffer++);
    tvp = (struct timeval *) get_fs_long(buffer);

// 上面的inp,outp,exp,tvp看出变量名都有个p，它们都是指针，我们只是把4个指针读到内核空间了
// get_fd_set作用就是从用户空间把指针对应的值读入内核空间的in,out,ex变量，超时是一个时间对象，这个具体可以自己查询一下
    if ((i = get_fd_set(n, inp, &in)) ||
        (i = get_fd_set(n, outp, &out)) ||
        (i = get_fd_set(n, exp, &ex))) return i;
    timeout = ~0UL;
    if (tvp) {
        i = verify_area(VERIFY_WRITE, tvp, sizeof(*tvp));
        if (i)
            return i;
        timeout = ROUND_UP(get_fs_long((unsigned long *)&tvp->tv_usec),(1000000/HZ));
        timeout += get_fs_long((unsigned long *)&tvp->tv_sec) * HZ;
        if (timeout)
            timeout += jiffies + 1;
    }
// 设置进程的超时属性，current代表当前进程的描述符对象
    current->timeout = timeout;
// 好了上面其实都是从用户空间搬运数据到内核，接下来才是真正的主角do_select
    i = do_select(n, &in, &out, &ex, &res_in, &res_out, &res_ex);
    /* 记录实际等待时间 */
    if (current->timeout > jiffies)
        timeout = current->timeout - jiffies;
    else
        timeout = 0;
    current->timeout = 0;
    if (tvp) {
        put_fs_long(timeout/HZ, (unsigned long *) &tvp->tv_sec);
        timeout %= HZ;
        timeout *= (1000000/HZ);
        put_fs_long(timeout, (unsigned long *) &tvp->tv_usec);
    }
    if (i < 0)
        return i;
    if (!i && (current->signal & ~current->blocked))
        return -ERESTARTNOHAND;
    /* 将监测到的结果回写到参数对应的内存当中 */
    set_fd_set(n, inp, &res_in);
    set_fd_set(n, outp, &res_out);
    set_fd_set(n, exp, &res_ex);
    return i;
}

上面注释已经写得蛮清晰了，总结一下sys_select干了什么，其实就是先从用户空间把需要监听的fd_set和超时时间读到内核然后调用do_select方法。那接下来就看看do_select

int do_select(int n, fd_set *in, fd_set *out, fd_set *ex,
    fd_set *res_in, fd_set *res_out, fd_set *res_ex)
{
    int count;
    select_table wait_table, *wait;
    struct select_table_entry *entry;
    unsigned long set;
    int i,j;
    int max = -1;

    /* 循环检查每个文件集合的位 */
    for (j = 0 ; j < __FDSET_LONGS ; j++) {
        /* 一个unsigned long能够表示32个文件描述符，32=2的5次方*/
        i = j << 5;
        /* 超过最大文件描述符，则停止 */
        if (i >= n)
            break;
        set = in->fds_bits[j] | out->fds_bits[j] | ex->fds_bits[j];
        /* set移动8次就等于0了 */
        for ( ; set ; i++,set >>= 1) {
            if (i >= n)
                goto end_check;
            /* 测试集合中的最后一位 */
            if (!(set & 1))
                continue;
            // 如果进程并未打开该fd返回错误
            if (!current->filp[i])
                return -EBADF;
            // 如果进程打开的fd没有节点属性返回错误
            if (!current->filp[i]->f_inode)
                return -EBADF;
            /* 记录最大的文件描述符 */
            max = i;
        }
    }
end_check:
    /* 记录实际监视的文件描述符的最大值+1 */
    n = max + 1;

    /* 获取占用一页大小的select_table_entry内存 */
    if(!(entry = (struct select_table_entry*) __get_free_page(GFP_KERNEL)))
        return -ENOMEM;
    FD_ZERO(res_in);
    FD_ZERO(res_out);
    FD_ZERO(res_ex);
    count = 0;
    /* 初始化等待列表 */
    wait_table.nr = 0;
    wait_table.entry = entry;
    wait = &wait_table;
repeat:
    current->state = TASK_INTERRUPTIBLE;
    /* 循环扫描所有监视的文件描述符，注意这里的wait第一次检测到有文件变动之前传入check函数不为NULL,
     * 之后就都为NULL,因为每一个网络文件描述符都对应一个唯一的struct sock，当前进程只需要在struct sock
     * 的sleep中等待一次就够了。
     */
    for (i = 0 ; i < n ; i++) {
        if (FD_ISSET(i,in) && check(SEL_IN,wait,current->filp[i])) {
            FD_SET(i, res_in);
            count++;
            wait = NULL; /* 此处设置为NULL,就代表已经检测到了一个文件变动，所以不需要在等待在其他文件上
                              * 在if(!count-----)判断时就可以快速返回 
                              */
        }
        if (FD_ISSET(i,out) && check(SEL_OUT,wait,current->filp[i])) {
            FD_SET(i, res_out);
            count++;
            wait = NULL;
        }
        if (FD_ISSET(i,ex) && check(SEL_EX,wait,current->filp[i])) {
            FD_SET(i, res_ex);
            count++;
            wait = NULL;
        }
    }
    /* 所有的文件被扫描一次过后，就已经添加到struct sock中的sleep当中，
     * 如果wait不为NULL，则会继续添加一次 
     */
    wait = NULL;
    /* 注意这里的阻塞道理，经过一轮扫描过后，一旦监视到文件有变动，就立即返回
     * 在第三个判断条件当中，如果当前进程接收到了信号，并且没有
     * 被阻塞，则select函数不会继续阻塞了，因为该函数是在内核态当中，
     * 因为信号的处理函数是在内核态返回到用户态的时候执行的，所以为了
     * 尽快响应信号，就让进程从该函数当中快速退出 
     */
    if (!count && current->timeout && !(current->signal & ~current->blocked)) {
        /* 此时当前进程已经添加到所有strut sock中的sleep链表当中，主动放弃cpu */
        schedule();
        goto repeat;
    }
    /* 将当前进程从所有已经添加到struct sock的sleep链表中删除 */
    free_wait(&wait_table);
    free_page((unsigned long) entry);
    current->state = TASK_RUNNING;
    return count;
}

这一段的注释是摘自github，感谢

主要的判断fd是否准话好了是check方法，看看它干了啥

static int check(int flag, select_table * wait, struct file * file)
{
    struct inode * inode;
    struct file_operations *fops;
    int (*select) (struct inode *, struct file *, int, select_table *);

// 取fd对应文件的节点对象
    inode = file->f_inode;
// f_op是一个操作对象，因为文件有很多种，有块文件字符文件，网络文件，对应不同的f_op对象
// 取实际对象的select方法，然后执行
    if ((fops = file->f_op) && (select = fops->select))
        return select(inode, file, flag, wait)
            || (wait && select(inode, file, flag, NULL));
    /* 普通文件肯定是可以被读写的 */
    if (S_ISREG(inode->i_mode))
        return 1;
    return 0;
}

struct select_table_entry {
    struct wait_queue wait;         /* 实际添加到等待队列中的变量 */
    struct wait_queue ** wait_address;  /* 指向实际等待链表的首部，方便从wait_address中删除或添加wait */
};

typedef struct select_table_struct {
    int nr;
    struct select_table_entry * entry;
} select_table;

struct file {
    mode_t f_mode;          /* 文件不存在时，创建文件的权限 */
    dev_t f_rdev;           /* needed for /dev/tty */
    off_t f_pos;            /* 文件读写偏移量 */
    unsigned short f_flags; /* 以什么样的方式打开文件，如只读，只写等等 */
    unsigned short f_count;  /*文件的引用计数*/
    unsigned short f_reada;
    struct file *f_next, *f_prev;
    struct inode * f_inode;     /* 文件对应的inode */
    struct file_operations * f_op;    // 对应inode的文件操作
};

/* 对inode对应文件的操作
 */
struct file_operations {
    int (*lseek) (struct inode *, struct file *, off_t, int);
    int (*read) (struct inode *, struct file *, char *, int);
    int (*write) (struct inode *, struct file *, char *, int);
    int (*readdir) (struct inode *, struct file *, struct dirent *, int);
    int (*select) (struct inode *, struct file *, int, select_table *);
    int (*ioctl) (struct inode *, struct file *, unsigned int, unsigned long);
    int (*mmap) (struct inode *, struct file *, unsigned long, size_t, int, unsigned long);
    int (*open) (struct inode *, struct file *);
    void (*release) (struct inode *, struct file *);
    int (*fsync) (struct inode *, struct file *);
};

现在知道了，内核对每个fd_set遍历，然后对每个node调用check方法，check方法是根据每个文件对象绑定的，这类似面向对象的接口，每一种文件类型都要实现它的file_operations属性，屏蔽底层实现的不同而对上层暴露统一的接口。
比如搜了下内核就会发现有

• pipe_select 通道
• mouse_select 鼠标
• tty_select 字符设备
• sock_select 网络文件
• tcp_select tcp

还有很多就不一一列举了，那我们找个简单的看一下吧- -！ pipe_select

static int pipe_select(struct inode * inode, struct file * filp, int sel_type, select_table * wait)
{
    switch (sel_type) {
        case SEL_IN:
            if (!PIPE_EMPTY(*inode) || !PIPE_WRITERS(*inode))
                return 1;
            select_wait(&PIPE_WAIT(*inode), wait);
            return 0;
        case SEL_OUT:
            if (!PIPE_FULL(*inode) || !PIPE_READERS(*inode))
                return 1;
            select_wait(&PIPE_WAIT(*inode), wait);
            return 0;
        case SEL_EX:
            if (!PIPE_READERS(*inode) || !PIPE_WRITERS(*inode))
                return 1;
            select_wait(&inode->i_wait,wait);
            return 0;
    }
    return 0;
}

extern inline void select_wait(struct wait_queue ** wait_address, select_table * p)
{
    struct select_table_entry * entry;

    /* 如果有任意一个指针为NULL,则返回，不做处理 */
    if (!p || !wait_address)
        return;
    /* 数量不能超过 */
    if (p->nr >= __MAX_SELECT_TABLE_ENTRIES)
        return;
    /* 获取当前操作的entry的地址 */
    entry = p->entry + p->nr;
    /* 注意这个地方很重要，也就是要记录wait变量是等待在哪个链表当中 */
    entry->wait_address = wait_address;
    entry->wait.task = current;
    entry->wait.next = NULL;
    add_wait_queue(wait_address,&entry->wait);
    /* 增加nr的数量 */
    p->nr++;
}

嗯，快接近真相了，内核最终调用add_wait_queue把资源加入等待队列中

extern inline void add_wait_queue(struct wait_queue ** p, struct wait_queue * wait)
{
    unsigned long flags;

// 调试相关
#ifdef DEBUG
    if (wait->next) {
        unsigned long pc;
        __asm__ __volatile__("call 1f\n"
            "1:\tpopl %0":"=r" (pc));
        printk("add_wait_queue (%08x): wait->next = %08x\n",pc,(unsigned long) wait->next);
    }
#endif
    save_flags(flags);
// 关中断
    cli();
    /* 如果队列为空，则让wait指向队首，next指向自己
     * 当继续向队列中添加节点时，整个队列就是一个闭合的圆环，
     * 否则将wait添加到以*p为队首的下一个地方
     */
    if (!*p) {
        wait->next = wait;
        *p = wait;
    } else {
        wait->next = (*p)->next;
        (*p)->next = wait;
    }
    restore_flags(flags);
}

// 恢复flags
#define restore_flags(x) \
__asm__ __volatile__("pushl %0 ; popfl": /* no output */ :"r" (x):"memory")

好了，就是当前进程和等待位置加入到对应的节点的等待队列中，这样当内核收到中断返回文件可用时候就会挨个通知等待队列上的任务了。

总结一下

select其实只是扩展了内核异步的结构，内核的唤醒机制没太大变化，只是给进程委托了一个select的代理对象，但是性能来看，首先每次select都要从用户空间把数据搬运到内核，而且还有一个fd_set这个数量限制，这里重点说一下这个吧，fd_set是一个long性数组，linux用它的二进制做位图，1.0版本数组大小是8，也就是可以存放最大848=256个描述符，现在linux默认是1024，我们可以修改该值来完成对select数量的限制。poll的原理其实和select差不多，只是不在使用这种参数传递的方式，而是传入一个数组对象，这样可以避开fd_set的大小的限制

#define __FDSET_LONGS 8
typedef struct fd_set {
    unsigned long fds_bits [__FDSET_LONGS];
} fd_set;