Java NIO之EPollSelectorImpl详解

本文简述JDK1.7的NIO在linux平台上的实现，对java NIO的一些核心概念如Selector,Channel,Buffer等，不会做过多解释，这些请参考JDK的文档。JDK 1.7 NIO Selector在linux平台上的实现类是sun.nio.ch.EPollSelectorImpl，这个类通过linux下的epoll系列系统调用实现NIO，因此在介绍这个类的实现之前，先介绍一下linux的epoll。epoll是poll/select系统调用的一个改进版本，能以更高的性能实现IO事件的检测和分发（主要归功于epoll的事件回调机制，下文详述），主要包含以下3个系统调用：

上述函数中，epoll_create函数负责创建一个检测IO事件的epoll实例，size参数用于“暗示”操作系统事件队列的长度，在linux-2.6.32内核中，此参数被忽略。epoll_ctl函数用于管理文件描述符的事件集，使用此函数可以注册、修改、删除一个或多个事件。epoll_wait负责检测事件，这三个函数的详细描，请参阅epoll的man文档。

Java类sun.nio.ch.EPollSelectorImpl主要的功能都委托给sun.nio.ch. EPollArrayWrapper实现 (下文所引java代码反编译自linux版jdk_1.7.0_17/lib/rt.jar)：

  package sun.nio.ch;  class EPollArrayWrapper{  private native int epollCreate();  private native void epollCtl(int paramInt1, int paramInt2, int paramInt3, int paramInt4);  private native int epollWait(long paramLong1, int paramInt1, long paramLong2, int paramInt2) throws IOException;  }  

EPollArrayWrapper的三个native方法的实现代码可参阅openjdk7/jdk/src/solaris/native/sun/nio/ch/ EPollArrayWrapper.c,可看到这三个native方法正是对上述epoll系列系统调用的包装。（其他jdk的实现代码会有所不同，但归根结底都是对epoll系列系统调用的包装）。

EPollSelectorImpl. implRegister方法(Selector.register方法的具体实现)，通过调用epoll_ctl向epoll实例中注册事件：

  protected void implRegister(SelectionKeyImpl paramSelectionKeyImpl) {       if (this.closed)         throw new ClosedSelectorException();       SelChImpl localSelChImpl = paramSelectionKeyImpl.channel;       this.fdToKey.put(Integer.valueOf(localSelChImpl.getFDVal()), paramSelectionKeyImpl);       this.pollWrapper.add(localSelChImpl);       this.keys.add(paramSelectionKeyImpl);  }  

上述方法中，除了向epoll实例注册事件外，还将注册的文件描述符(fd)与SelectionKey的对应关系添加到fdToKey中，这个map维护了文件描述符与SelectionKey的映射。每当向Selector中注册一个Channel时，向此map中添加一条记录，而当Channel.close、SelectionKey.cancel方法调用时，则从fdToKey中移除与Channel的fd相关联的SelectionKey，具体代码在EPollSelectorImpl.implDereg方法中：

  protected void implDereg(SelectionKeyImpl paramSelectionKeyImpl) throws IOException {       assert (paramSelectionKeyImpl.getIndex() >= 0);       SelChImpl localSelChImpl = paramSelectionKeyImpl.channel;       int i = localSelChImpl.getFDVal();       this.fdToKey.remove(Integer.valueOf(i));       this.pollWrapper.release(localSelChImpl);       paramSelectionKeyImpl.setIndex(-1);       this.keys.remove(paramSelectionKeyImpl);       this.selectedKeys.remove(paramSelectionKeyImpl);       deregister(paramSelectionKeyImpl);       SelectableChannel localSelectableChannel = paramSelectionKeyImpl.channel();       if ((!localSelectableChannel.isOpen()) && (!localSelectableChannel.isRegistered()))         ((SelChImpl)localSelectableChannel).kill();  }  

EPollSelectorImpl. doSelect(Selector.select方法的实现)，则通过调用epoll_wait实现事件检测：

  protected int doSelect(long paramLong)       throws IOException     {       if (this.closed)         throw new ClosedSelectorException();       processDeregisterQueue();       try {         begin();         this.pollWrapper.poll(paramLong);       } finally {         end();       }       processDeregisterQueue();       int i = updateSelectedKeys();       if (this.pollWrapper.interrupted())       {         this.pollWrapper.putEventOps(this.pollWrapper.interruptedIndex(), 0);         synchronized (this.interruptLock) {           this.pollWrapper.clearInterrupted();           IOUtil.drain(this.fd0);           this.interruptTriggered = false;         }       }       return i;     }  

此方法的主要流程概括如下：

1. 通过epoll_wait调用（this.pollWrapper.poll）获取已就绪的文件描述符集合

2. 通过fdToKey查找文件描述符对应的SelectionKey，并更新之，更新SelectionKey的具体代码在EPollSelectorImpl .updateSelectedKeys中：

  private int updateSelectedKeys()     {       int i = this.pollWrapper.updated;       int j = 0;       for (int k = 0; k < i; k++) {        int m = this.pollWrapper.getDescriptor(k);        SelectionKeyImpl localSelectionKeyImpl = (SelectionKeyImpl)this.fdToKey.get(Integer.valueOf(m));          if (localSelectionKeyImpl != null) {          int n = this.pollWrapper.getEventOps(k);          if (this.selectedKeys.contains(localSelectionKeyImpl)) {            if (localSelectionKeyImpl.channel.translateAndSetReadyOps(n, localSelectionKeyImpl))              j++;          }          else {            localSelectionKeyImpl.channel.translateAndSetReadyOps(n, localSelectionKeyImpl);            if ((localSelectionKeyImpl.nioReadyOps() & localSelectionKeyImpl.nioInterestOps()) != 0) {              this.selectedKeys.add(localSelectionKeyImpl);              j++;            }          }        }      }      return j;    } 

下面聊一聊epoll系统调用的具体实现，它的实现代码在(linux-2.6.32.65)fs/eventpoll.c中（下文所引内核代码，由于较长，所以只贴出主流程，省略了错误处理及一些相对次要的细节如参数检查、并发控制等），先看epoll_create 系统调用的实现：

fs/eventpoll.c

  SYSCALL_DEFINE1(epoll_create, int, size)  {   if (size <= 0)   return -EINVAL;  return sys_epoll_create1(0); } 

SYSCALL_DEFINE1是一个宏，用于定义有一个参数的系统调用函数，上述宏展开后即成为：

int sys_epoll_create(int size)

这就是epoll_create系统调用的入口。至于为何要用宏而不是直接声明，主要是因为系统调用的参数个数、传参方式都有严格限制，最多六个参数, SYSCALL_DEFINE2 -SYSCALL_DEFINE6分别用来定义有2-6个参数的系统调用。由上述代码可知，epoll_create函数最终调用sys_epoll_create1实现具体功能，同时也可以看出size参数被忽略了。sys_epoll_create1的主要代码如下（省略了错误处理及一些次要的细节如参数检查等）：

fs/eventpoll.c

  SYSCALL_DEFINE1(epoll_create1, int, flags)  {   int error, fd;   struct eventpoll *ep = NULL;   struct file *file;   error = ep_alloc(&ep);   file = anon_inode_getfile("[eventpoll]", &eventpoll_fops, ep,       O_RDWR | (flags & O_CLOEXEC));   fd_install(fd, file);   ep->file = file;   return fd;  }  

上述代码主要是分配一个struct eventpoll实例，并分配与此实例相关联的文件描述符，后续的epoll_ctl,epoll_wait调用通过此文件描述符引用此实例。struct eventpoll的结构如下：

fs/eventpoll.c

  struct eventpoll {   spinlock_t lock;   struct mutex mtx;   wait_queue_head_t wq;   wait_queue_head_t poll_wait;   struct list_head rdllist;   struct rb_root rbr;   struct epitem *ovflist;   struct user_struct *user;   struct file *file;   int visited;   struct list_head visited_list_link;  }  

上述数据结构的关键部分是：

1. 一个等待队列wq，epoll正是通过此等待队列实现的事件回调

2. 一个就绪列表rdllist,此列表以双链表的形式保存了已就绪的文件描述符

3. 一个红黑树rbr，用于保存已注册过的文件描述符，若重复注册相同的文件描述符，则会返回错误

等待队列是epoll系统调用的核心机制(不只是epoll，linux下事件的通知、回调等机制大都依赖于等待队列)，在讲述epoll_ctl,epoll_wait的实现之前，先来看看等待队列。等待队列可以使一组进程/线程在等待某个事件时睡眠，当等待的事件发生时，内核会唤醒睡眠的进程/线程。注意，下文并不区分进程和线程，在linux下，进程和线程在调度这个意义下(调度就是指linux的进程调度，包括进程的切换、睡眠、唤醒等)并无差别。此机制可以类比java.lang.Object类的wait和notify/notifyAll方法，其中wait方法使线程睡眠，notify/notifyAll方法则唤醒睡眠的一个或全部线程。等待队列主要涉及两个数据结构：

include/linux/wait.h

  struct __wait_queue_head {   spinlock_t lock;   list_head task_list;  };  struct __wait_queue {   unsigned int flags;  #define WQ_FLAG_EXCLUSIVE 0x01   void *private;   wait_queue_func_t func;   struct list_head task_list;  };  

struct __wait_queue_head是队头结构，task_list 保存了添加到此队列上的元素,struct list_head是标准的linux双链表, 定义如下：

include/linux/list.h

  struct list_head {   struct list_head *next, *prev;  };  

注意，此结构既可以表示双链表的表头，也可以表示一个链表元素，且next,prev这两个指针可以指向任意数据结构。

struct __wait_queue是等待队列的元素结构，成员func是等待的进程被唤醒时执行的回调函数，其定义如下：

include/linux/wait.h

  typedef int (*wait_queue_func_t)(wait_queue_t *wait, unsigned mode, int flags, void *key);  

struct __wait_queue的成员task_list是一个链表元素用于将此结构放置到struct __wait_queue_head中（这和此结构的task_list成员含义是不同的，此成员的含义为双链表的头），private成员一般指向等待进程的task_struct实例(该结构成员很多，在此就不贴出了，只需要知道linux下每个进程都对应一个task_struct 实例)。

在使用上，等待队列主要涉及以下函数(或者宏)：

include/linux/wait.h

__add_wait_queue(wait_queue_head_t *q, wait_queue_t *wait);

#define wait_event(wq, condition)

#define wake_up_xxx(x,…)

__add_wait_queue用于将一个进程添加到等待队列，wait_event是一个宏，它用于等待一个事件，当事件未发生时使等待的进程休眠，wake_up_xxx是一系列的宏，包括wake_up，wake_up_all,wake_up_locked,wake_up_interruptible等，负责唤醒休眠在某个事件上的一个或一组进程。关于等待队列的具体实现细节，由于牵涉较广(涉及到进程调度、中断处理等)，这里不再详述，可以将add_wait_queue，wait_event类比java.lang.Object的wait方法，而wake_up则可以类比java.lang.Object的notify/notifyAll方法。

介绍完等待队列后，就可以进一步研究epoll_ctl的实现了，其代码实现中核心的部分是：

fs/eventpoll.c

  SYSCALL_DEFINE4(epoll_ctl, int, epfd, int, op, int, fd,    struct epoll_event __user *, event)  {   if (!tfile->f_op || !tfile->f_op->poll)    goto error_tgt_fput;   switch (op) {   case EPOLL_CTL_ADD:  error=ep_insert(ep, &epds, tfile, fd);    break;   case EPOLL_CTL_DEL:  error=ep_remove(ep, epi);    break;   case EPOLL_CTL_MOD:    error = ep_modify(ep, epi, &epds);    break;   }   return error;  }  

什么样的文件描述符可以注册?从那个if判断可以看出，只有文件描述符对应的文件实现了poll方法的才可以，一般而言，字符设备的文件都实现了此方法，网络相关的套接字也实现了此方法，而块设备文件例如ext2/ext3/ext4文件系统文件，都没有实现此方法。实现了poll方法的文件，对应于java NIO的java.nio.channels.SelectableChannel，这也是为何只有 SelectableChannel 才能注册到Selector上的原因。ep_insert，ep_remove,ep_modify分别对应事件的注册、删除、修改，我们以ep_insert为例，看一下事件注册的过程，其关键代码如下：

fs/eventpoll.c

  static int ep_insert(struct eventpoll *ep, struct epoll_event *event,         struct file *tfile, int fd)  {  init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);  revents = tfile->f_op->poll(tfile, &epq.pt);  ep_rbtree_insert(ep, epi);  if ((revents & event->events) && !ep_is_linked(&epi->rdllink)) {  list_add_tail(&epi->rdllink, &ep->rdllist);;  wake_up_locked(&ep->wq);  }  }  

上述代码的主要做的事是：

1. 绑定等待队列的回调函数ep_ptable_queue_proc

2. 调用对应文件的实例的poll方法，此方法的具体实现差别非常大，但绝大多数都会调用wait_event相关方法，在没有事件发生时，使进程睡眠，例如socket对应的实现(代码在net/ipv4/af_inet.c的tcp_poll方法，在此不再详述)；

3. 若注册的事件已经发生，则将已就绪的文件描述符插入到eventpoll实例的就绪列表（list_add_tail(&epi->rdllink, &ep->rdllist);），并唤醒睡眠的进程（wake_up_locked(&ep->wq)）

第1步绑定的回调函数ep_ptable_queue_proc,会在等待的事件发生时执行，其主要功能是将就绪的文件描述符插入到eventpoll实例的就绪列表（具体是通过ep_ptable_queue_proc绑定的另一个回调函数ep_poll_callback实现的）：

fs/eventpoll.c

  static int ep_poll_callback(wait_queue_t *wait, unsigned mode, int sync, void *key){   if (!ep_is_linked(&epi->rdllink))    list_add_tail(&epi->rdllink, &ep->rdllist);  }  

最后看epoll_wait的实现，有了就绪队列，epoll_wait的实现就比较简单了，只需检查就绪队列是否为空，若为空，则在必要时睡眠或等待：

fs/eventpoll.c

  SYSCALL_DEFINE4(epoll_wait, int, epfd, struct epoll_event __user *, events,    int, maxevents, int, timeout)  {   int error;   struct file *file;   struct eventpoll *ep;   file = fget(epfd);   ep = file->private_data;   error = ep_poll(ep, events, maxevents, timeout);   return error;  }  

此函数最终调用ep_poll完成其主要功能：

  static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,       int maxevents, long timeout)  {   retry:   if (list_empty(&ep->rdllist)) {    init_waitqueue_entry(&wait, current);    wait.flags |= WQ_FLAG_EXCLUSIVE;    __add_wait_queue(&ep->wq, &wait);