文章详情

登录

短信预约-IT技能 免费直播动态提醒

选择考试省份

  • 北京
  • 上海
  • 天津
  • 重庆
  • 河北
  • 山东
  • 辽宁
  • 黑龙江
  • 吉林
  • 甘肃
  • 青海
  • 河南
  • 江苏
  • 湖北
  • 湖南
  • 江西
  • 浙江
  • 广东
  • 云南
  • 福建
  • 海南
  • 山西
  • 四川
  • 陕西
  • 贵州
  • 安徽
  • 广西
  • 内蒙
  • 西藏
  • 新疆
  • 宁夏
  • 兵团

请输入下面的图形验证码

提交验证

短信预约提醒成功

从内核看Epoll的实现

2024-12-03 03:00

关注

epoll是现代服务器的基石，也是高效处理大量请求的利器，从设计上来看，epoll的设计思想也是非常优秀的，本文介绍epoll的实现，从中我们不仅看到epoll的实现原理和机制，同时也能领略到其中优秀的设计思想。

epoll的使用非常简单，主要是几个API，下面我们一个个分析。

1 epoll_create

epoll_create是创建epoll实例的API，对使用方来说，epoll是一个黑盒子，我们通过操作系统提供的API，拿到一个实例(黑盒子)之后，就可以往里面注册我们想要监听的fd和事件，条件满足的时候，epoll就会通知我们，下面我们看看epoll_create的实现。

  1. SYSCALL_DEFINE1(epoll_create1, int, flags){ 
  2.     return do_epoll_create(flags); 
  3. } 
  4.  
  5. SYSCALL_DEFINE1(epoll_create, int, size){ 
  6.     if (size <= 0) 
  7.         return -EINVAL; 
  8.     return do_epoll_create(0); 
  9. } 

我们看到epoll_create有两个版本，其中epoll_create1多支持了flags参数，比如设置非阻塞模式，两个API具体的区别不大。接下来我们看do_epoll_create。

  1. static int do_epoll_create(int flags){ 
  2.     int error, fd; 
  3.     struct eventpoll *ep = NULL; 
  4.     struct file *file; 
  5.  
  6.     // 只支持CLOEXEC 
  7.     if (flags & ~EPOLL_CLOEXEC) 
  8.         return -EINVAL; 
  9.  
  10.     // 分配一个eventpoll 
  11.     error = ep_alloc(&ep); 
  12.  
  13.     // 获取一个空闲文件描述符 
  14.     fd = get_unused_fd_flags(O_RDWR | (flags & O_CLOEXEC)); 
  15.  
  16.     // 获取一个file，并且关联eventpoll_fops和上下文ep 
  17.     file = anon_inode_getfile("[eventpoll]", &eventpoll_fops, ep, O_RDWR | (flags & O_CLOEXEC)); 
  18.  
  19.     // ep和file关联起来，上面是file和ep关联 
  20.     ep->file = file; 
  21.     // 关联fd和file 
  22.     fd_install(fd, file); 
  23.     return fd; 
  24. } 

我们看到do_epoll_create的实现非常简单，主要是创建了一个eventpoll结构体，eventpoll结构体比较复杂。下面列出核心的字段。

  1. struct eventpoll { 
  2.     struct mutex mtx; 
  3.     // 阻塞在该epoll的进程队列 
  4.     wait_queue_head_t wq; 
  5.     // 当epoll被另一个epoll监听时需要使用poll_wait记录阻塞在该epoll的队列 
  6.     wait_queue_head_t poll_wait; 
  7.     // 就绪队列 
  8.     struct list_head rdllist; 
  9.     rwlock_t lock; 
  10.     // 红黑树根节点 
  11.     struct rb_root_cached rbr; 
  12.     // 记录epitem的单链表 
  13.     struct epitem *ovflist; 
  14.     struct wakeup_source *ws; 
  15.     // 创建该epoll的用户信息 
  16.     struct user_struct *user; 
  17.     // epoll对应的file 
  18.     struct file *file; 
  19. }; 

创建了一个eventpoll结构体后，接着申请了一个file和fd，并且把file和eventpoll关联起来，主要的作用是调用方后续可以通过fd操作eventpoll，架构如下。

2 epoll_ctl

epoll_ctl是操作epoll的总入口，也是非常复杂的开始，但是简单来说就是增删改的接口。

  1. SYSCALL_DEFINE4(epoll_ctl, int, epfd, int, op, int, fd, 
  2.         struct epoll_event __user *, event){ 
  3.     struct epoll_event epds; 
  4.     // 判断是否需要复制数据，如果是删除则不需要，根据fd删除就行 
  5.     if (ep_op_has_event(op) && 
  6.         copy_from_user(&epds, event, sizeof(struct epoll_event))) 
  7.         return -EFAULT; 
  8.  
  9.     return do_epoll_ctl(epfd, op, fd, &epds, false); 
  10. } 

epoll_ctl是对do_epoll_ctl的封装。

  1. // 操作epoll 
  2. int do_epoll_ctl(int epfd, int op, int fd, struct epoll_event *epds, 
  3.          bool nonblock){ 
  4.     int error; 
  5.     int full_check = 0; 
  6.     struct fd f, tf; 
  7.     struct eventpoll *ep; 
  8.     struct epitem *epi; 
  9.     struct eventpoll *tep = NULL; 
  10.  
  11.     error = -EBADF; 
  12.     // 根据fd找到对应的数据结构 
  13.     f = fdget(epfd); 
  14.  
  15.     // 获取被操作的文件描述符的数据结构 
  16.     tf = fdget(fd); 
  17.  
  18.     error = -EPERM; 
  19.     // 资源有没有实现poll接口，使用epoll监听的资源需要实现poll钩子 
  20.     if (!file_can_poll(tf.file)) 
  21.         goto error_tgt_fput; 
  22.  
  23.     error = -EINVAL; 
  24.     // 保证被操作的fd不是自己，并且自己是epoll 
  25.     if (f.file == tf.file || !is_file_epoll(f.file)) 
  26.         goto error_tgt_fput; 
  27.  
  28.     // 根据fd找到epoll数据结构 
  29.     ep = f.file->private_data; 
  30.  
  31.     // 加锁 
  32.     epoll_mutex_lock(&ep->mtx, 0, nonblock); 
  33.  
  34.     // 判断fd是否已经存在epoll的红黑树中 
  35.     epi = ep_find(ep, tf.file, fd); 
  36.  
  37.     error = -EINVAL; 
  38.     switch (op) { 
  39.     // 新增 
  40.     case EPOLL_CTL_ADD: 
  41.         // 之前没有则可以新增，否则报错 
  42.         if (!epi) { 
  43.             epds->events |= EPOLLERR | EPOLLHUP; 
  44.             // 插入epoll 
  45.             error = ep_insert(ep, epds, tf.file, fd, full_check); 
  46.         } else 
  47.             error = -EEXIST; 
  48.         break; 
  49.     // 删除 
  50.     case EPOLL_CTL_DEL: 
  51.         // 存在则删除，否则报错 
  52.         if (epi) 
  53.             error = ep_remove(ep, epi); 
  54.         else 
  55.             error = -ENOENT; 
  56.         break; 
  57.     // 修改 
  58.     case EPOLL_CTL_MOD: 
  59.         // 存在则修改，否则报错 
  60.         if (epi) { 
  61.             if (!(epi->event.events & EPOLLEXCLUSIVE)) { 
  62.                 epds->events |= EPOLLERR | EPOLLHUP; 
  63.                 error = ep_modify(ep, epi, epds); 
  64.             } 
  65.         } else 
  66.             error = -ENOENT; 
  67.         break; 
  68.     } 
  69.     return error; 
  70. } 

我们看到do_epoll_ctl主要首先通过两个fd拿到对应的epoll和资源，然后做了一些校验，接着根据操作类型做进一步处理，操作类型有增删改，我们只需要分析插入就行，这是epoll核心。

  1. static int ep_insert(struct eventpoll *ep, const struct epoll_event *event, 
  2.              struct file *tfile, int fd, int full_check){ 
  3.     int error, pwake = 0; 
  4.     __poll_t revents; 
  5.     long user_watches; 
  6.     struct epitem *epi; 
  7.     struct ep_pqueue epq; 
  8.     lockdep_assert_irqs_enabled(); 
  9.     // 监听的文件描述符个数 
  10.     user_watches = atomic_long_read(&ep->user->epoll_watches); 
  11.     // 超了 
  12.     if (unlikely(user_watches >= max_user_watches)) 
  13.         return -ENOSPC; 
  14.     // 分配一个epitem 
  15.     if (!(epi = kmem_cache_alloc(epi_cache, GFP_KERNEL))) 
  16.         return -ENOMEM; 
  17.  
  18.     // 初始化 
  19.     INIT_LIST_HEAD(&epi->rdllink); 
  20.     INIT_LIST_HEAD(&epi->fllink); 
  21.     INIT_LIST_HEAD(&epi->pwqlist); 
  22.     // 所属的epoll 
  23.     epi->ep = ep; 
  24.     // 保存fd和file 
  25.     ep_set_ffd(&epi->ffd, tfile, fd); 
  26.     // 记录订阅事件 
  27.     epi->event = *event; 
  28.     epi->nwait = 0; 
  29.     epi->next = EP_UNACTIVE_PTR; 
  30.     spin_lock(&tfile->f_lock); 
  31.     // 把epi插入所属file的队列 
  32.     list_add_tail_rcu(&epi->fllink, &tfile->f_ep_links); 
  33.     spin_unlock(&tfile->f_lock); 
  34.     // 插入红黑树 
  35.     ep_rbtree_insert(ep, epi); 
  36.     error = -EINVAL; 
  37.     // 关联对应的epitem 
  38.     epq.epi = epi; 
  39.     // 初始化ep_pqueue 
  40.     init_poll_funcptr(&epq.pt, ep_ptable_queue_proc); 
  41.     // 判断是否有事件触发了 
  42.     revents = ep_item_poll(epi, &epq.pt, 1); 
  43.     error = -ENOMEM; 
  44.     write_lock_irq(&ep->lock); 
  45.     // 事件触发了，并且还没有加入就绪队列则加入 
  46.     if (revents && !ep_is_linked(epi)) { 
  47.         list_add_tail(&epi->rdllink, &ep->rdllist); 
  48.         // 等待队列非空则唤醒阻塞在该epoll的队列 
  49.         if (waitqueue_active(&ep->wq)) 
  50.             wake_up(&ep->wq); 
  51.         // 一个epoll被另一个监听，唤醒主epoll 
  52.         if (waitqueue_active(&ep->poll_wait)) 
  53.             pwake++; 
  54.     } 
  55.     // 一个epoll被另一个监听，唤醒主epoll 
  56.     if (pwake) 
  57.         ep_poll_safewake(ep, NULL); 
  58.     write_unlock_irq(&ep->lock); 
  59.     // 监听数加一 
  60.     atomic_long_inc(&ep->user->epoll_watches); 
  61.     return 0; 
  62. } 

插入操作的逻辑分为以下几个部分

1 分配一个epitem表示一个被epoll监听的项，插入红黑树。

2 判断当前被监听的fd订阅的事件是否触发了，即注册的时候，事件就触发了，是则插入就绪队列。

3 初始化并注册节点到资源对应的队列中。

1，2的逻辑是很自然的，执行完后的架构如下

我们重点来分析3，3也是epoll最核心的设计，也就是资源满足条件的时候是如何通知epoll的，核心代码如下。

  1. struct ep_pqueue epq; 
  2. // 关联对应的epitem 
  3. epq.epi = epi; 
  4. // 把函数ep_ptable_queue_proc保存到epq.pt 
  5. init_poll_funcptr(&epq.pt, ep_ptable_queue_proc); 
  6. // 判断是否有事件触发了 
  7. revents = ep_item_poll(epi, &epq.pt, 1); 

我们看到上面代码初始化了一个ep_pqueue结构体，重点是把epitem关联到了ep_pqueue结构体中，后面会看到它的作用。我们看看ep_pqueue结构体的定义。

  1. typedef struct poll_table_struct { 
  2.     // 函数指针 
  3.     poll_queue_proc _qproc; 
  4.     // unsigned 
  5.     __poll_t _key; 
  6. } poll_table; 
  7.  
  8. struct ep_pqueue { 
  9.     poll_table pt; 
  10.     struct epitem *epi; 
  11. }; 

上面代码执行完之后架构如下。

初始化完后接着看ep_item_poll函数。

  1. static __poll_t ep_item_poll(const struct epitem *epi, poll_table *pt, 
  2.                  int depth){ 
  3.     struct eventpoll *ep; 
  4.     bool locked; 
  5.  
  6.     pt->_key = epi->event.events; 
  7.     // 不是epoll，则执行钩子函数poll 
  8.     if (!is_file_epoll(epi->ffd.file)) 
  9.         return vfs_poll(epi->ffd.file, pt) & epi->event.events; 
  10. } 
  11.  
  12. static inline __poll_t vfs_poll(struct file *file, struct poll_table_struct *pt){ 
  13.     if (unlikely(!file->f_op->poll)) 
  14.         return DEFAULT_POLLMASK; 
  15.     return file->f_op->poll(file, pt); 
  16. } 

ep_item_poll的逻辑是主要是执行poll钩子函数。epoll是一种机制，支持epoll的其他模块，需要实现poll钩子函数。下面以eventfd为例。

  1. static __poll_t eventfd_poll(struct file *file, poll_table *wait){ 
  2.     struct eventfd_ctx *ctx = file->private_data; 
  3.     __poll_t events = 0; 
  4.     u64 count; 
  5.      
  6.     poll_wait(file, &ctx->wqh, wait); 
  7.     // 判断当前触发的事件 
  8.     count = READ_ONCE(ctx->count); 
  9.  
  10.     if (count > 0) 
  11.         events |= EPOLLIN; 
  12.     if (count == ULLONG_MAX) 
  13.         events |= EPOLLERR; 
  14.     if (ULLONG_MAX - 1 > count) 
  15.         events |= EPOLLOUT; 
  16.  
  17.     return events; 
  18. } 

eventfd的poll函数为eventfd_poll。eventfd_poll会判断当前触发的事件，如果恰好是调用方订阅的事件，则直接插入就绪队列。我们主要看poll_wait的逻辑，这是非常核心的逻辑。

  1.  
  2. static inline void poll_wait(struct file * filp, wait_queue_head_t * wait_address, poll_table *p){ 
  3.     if (p && p->_qproc && wait_address) 
  4.         p->_qproc(filp, wait_address, p); 
  5. } 

poll_wait简单地调用_qproc函数。如果我们还有印象的话，可能会记得这个函数是ep_ptable_queue_proc。

  1. // 具体的资源方调用 
  2. static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead, 
  3.                  poll_table *pt){    
  4.     // 获取pt关联的epitem 
  5.     struct epitem *epi = ep_item_from_epqueue(pt); 
  6.     // 分配一个eppoll_entry 
  7.     struct eppoll_entry *pwq; 
  8.  
  9.     if (epi->nwait >= 0 && (pwq = kmem_cache_alloc(pwq_cache, GFP_KERNEL))) { 
  10.         // 初始化pwq，记录ep_poll_callback函数 
  11.         init_waitqueue_func_entry(&pwq->wait, ep_poll_callback); 
  12.         pwq->whead = whead; 
  13.         // 关联的epitem 
  14.         pwq->base = epi; 
  15.         // pwq插入whead队列，whead由具体资源提供，比如文件，管道，资源满足条件时会pwd对应的回调 
  16.         // 插入EPOLLEXCLUSIVE解决惊群 
  17.         if (epi->event.events & EPOLLEXCLUSIVE) 
  18.             add_wait_queue_exclusive(whead, &pwq->wait); 
  19.         else 
  20.             add_wait_queue(whead, &pwq->wait); 
  21.         // 插入关联的epi队列 
  22.         list_add_tail(&pwq->llink, &epi->pwqlist); 
  23.         epi->nwait++; 
  24.     } else { 
  25.          
  26.         epi->nwait = -1; 
  27.     } 
  28. } 

ep_ptable_queue_proc申请了一个eppoll_entry结构体，定义如下。

  1. struct wait_queue_entry { 
  2.     unsigned int        flags; 
  3.     void            *private; 
  4.     wait_queue_func_t   func; 
  5.     struct list_head    entry; 
  6. }; 
  7.  
  8. struct eppoll_entry { 
  9.     // 插入所属epitem节点的队列 
  10.     struct list_head llink; 
  11.     // 关联的epitem 
  12.     struct epitem *base; 
  13.     // 插入资源等待队列的节点 
  14.     wait_queue_entry_t wait; 
  15.     // 指向资源等待队列的头指针所在结构体 
  16.     wait_queue_head_t *whead; 
  17. }; 

ep_ptable_queue_proc申请了eppoll_entry结构体并初始化后，插入资具体功能模块定义的队列中，架构如下。

我们看到调用方往epoll注册了fd和事件，epoll并没有自己去实现检测的逻辑，而是同样地注册一个节点到对应的底层资源，等待它的通知。

3 epoll_wait

注册完fd和事件后，我们就会执行epoll_wait等待事件的触发，虽然有时候我们epoll_wait的时候，事件已经触发了，但是很多情况下，事件往往是异步触发的，比如我们发送一个网络请求，等待响应的时候，下面我们来分析实现。

  1. SYSCALL_DEFINE4(epoll_wait, int, epfd, struct epoll_event __user *, events, 
  2.         int, maxevents, int, timeout){ 
  3.     return do_epoll_wait(epfd, events, maxevents, timeout); 
  4. } 

epoll_wait是对do_epoll_wait的封装。

  1. static int do_epoll_wait(int epfd, struct epoll_event __user *events, 
  2.              int maxevents, int timeout){ 
  3.     int error; 
  4.     struct fd f; 
  5.     struct eventpoll *ep; 
  6.  
  7.      
  8.     if (maxevents <= 0 || maxevents > EP_MAX_EVENTS) 
  9.         return -EINVAL; 
  10.  
  11.     //  通过fd拿到底层的数据结构 
  12.     f = fdget(epfd); 
  13.     error = -EINVAL; 
  14.     // 判断是不是epoll实例 
  15.     if (!is_file_epoll(f.file)) 
  16.         goto error_fput; 
  17.  
  18.     // 取得epoll的核心结构体 
  19.     ep = f.file->private_data; 
  20.  
  21.     error = ep_poll(ep, events, maxevents, timeout); 
  22. } 

do_epoll_wait逻辑也不多，主要是拿到epoll实例，继续看ep_poll。

  1. static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events, 
  2.            int maxevents, long timeout){ 
  3.     int res = 0, eavail, timed_out = 0; 
  4.     u64 slack = 0; 
  5.     wait_queue_entry_t wait; 
  6.     ktime_t expires, *to = NULL; 
  7.  
  8.     // 设置了阻塞时间 
  9.     if (timeout > 0) { 
  10.         struct timespec64 end_time = ep_set_mstimeout(timeout); 
  11.         slack = select_estimate_accuracy(&end_time); 
  12.         to = &expires; 
  13.         *to = timespec64_to_ktime(end_time); 
  14.     } else if (timeout == 0) {// 0说明不阻塞 
  15.         timed_out = 1; 
  16.         write_lock_irq(&ep->lock); 
  17.         // 是否有就绪事件 
  18.         eavail = ep_events_available(ep); 
  19.         write_unlock_irq(&ep->lock); 
  20.         // 直接返回 
  21.         goto send_events; 
  22.     } 
  23.  
  24. fetch_events: 
  25.     // 是否有就绪事件 
  26.     eavail = ep_events_available(ep); 
  27.     // 有则通知用户 
  28.     if (eavail) 
  29.         goto send_events; 
  30.     do { 
  31.          
  32.         init_wait(&wait); 
  33.         write_lock_irq(&ep->lock); 
  34.         __set_current_state(TASK_INTERRUPTIBLE); 
  35.         // 是否有就绪队列 
  36.         eavail = ep_events_available(ep); 
  37.         // 没有但是当前有信号需要处理则返回EINTR，否则把当前进程加入队列 
  38.         if (!eavail) { 
  39.             if (signal_pending(current)) 
  40.                 res = -EINTR; 
  41.             else 
  42.                 __add_wait_queue_exclusive(&ep->wq, &wait); 
  43.         } 
  44.         write_unlock_irq(&ep->lock); 
  45.         // 报错或者有就绪事件则break 
  46.         if (eavail || res) 
  47.             break; 
  48.         // 阻塞当前进程 
  49.         if (!schedule_hrtimeout_range(to, slack, HRTIMER_MODE_ABS)) { 
  50.             timed_out = 1; 
  51.             break; 
  52.         } 
  53.         eavail = 1; 
  54.     } while (0); 
  55.     // 进程继续执行 
  56.     __set_current_state(TASK_RUNNING); 
  57.     // 当前进程还在队列（阻塞队列）则移除，因为进程被唤醒了 
  58.     if (!list_empty_careful(&wait.entry)) { 
  59.         write_lock_irq(&ep->lock); 
  60.         __remove_wait_queue(&ep->wq, &wait); 
  61.         write_unlock_irq(&ep->lock); 
  62.     } 
  63.  
  64. send_events: 
  65.     // 没有报错并且有就绪事件，通知用户 
  66.     if (!res && eavail && 
  67.         !(res = ep_send_events(ep, events, maxevents)) && !timed_out) 
  68.         goto fetch_events; 
  69.  
  70.     return res; 
  71. } 

接下来我们看看ep_send_events的逻辑。

  1. static int ep_send_events(struct eventpoll *ep, 
  2.               struct epoll_event __user *events, int maxevents){ 
  3.     struct ep_send_events_data esed; 
  4.     // 定义保存触发的事件的结构体 
  5.     esed.maxevents = maxevents; 
  6.     esed.events = events; 
  7.  
  8.     ep_scan_ready_list(ep, ep_send_events_proc, &esed, 0, false); 
  9.     return esed.res; 
  10. } 

继续看ep_scan_ready_list。

  1. static __poll_t ep_scan_ready_list(struct eventpoll *ep, 
  2.                   __poll_t (*sproc)(struct eventpoll *, 
  3.                        struct list_head *, void *), 
  4.                   void *priv, int depth, bool ep_locked){ 
  5.     __poll_t res; 
  6.     struct epitem *epi, *nepi; 
  7.     LIST_HEAD(txlist); 
  8.     // 把就绪队列移到txlist 
  9.     list_splice_init(&ep->rdllist, &txlist); 
  10.     // 执行传进来的函数ep_read_events_proc 
  11.     res = (*sproc)(ep, &txlist, priv); 
  12.      
  13.     list_splice(&txlist, &ep->rdllist); 
  14.     return res;}static __poll_t ep_send_events_proc(struct eventpoll *ep, struct list_head *head, 
  15.                    void *priv){ 
  16.     struct ep_send_events_data *esed = priv; 
  17.     __poll_t revents; 
  18.     struct epitem *epi, *tmp; 
  19.     struct epoll_event __user *uevent = esed->events; 
  20.     struct wakeup_source *ws; 
  21.     poll_table pt; 
  22.  
  23.     init_poll_funcptr(&pt, NULL); 
  24.     esed->res = 0; 
  25.     // 遍历就绪队列 
  26.     list_for_each_entry_safe(epi, tmp, head, rdllink) { 
  27.         if (esed->res >= esed->maxevents) 
  28.             break; 
  29.         // 移出就绪队列 
  30.         list_del_init(&epi->rdllink); 
  31.         // 触发的事件 
  32.         revents = ep_item_poll(epi, &pt, 1); 
  33.         // 写入调用方传入的结构体，返回0说明成功 
  34.         if (__put_user(revents, &uevent->events) || 
  35.             __put_user(epi->event.data, &uevent->data)) { 
  36.             // 失败则插入队列中 
  37.             list_add(&epi->rdllink, head); 
  38.             return 0; 
  39.         } 
  40.         // 处理个数加一 
  41.         esed->res++; 
  42.         uevent++; 
  43.         // 设置了EPOLLONESHOT则清除订阅的事件 
  44.         if (epi->event.events & EPOLLONESHOT) 
  45.             epi->event.events &= EP_PRIVATE_BITS; 
  46.         // // 没有设置水平触发则重新插入，下次epoll_wait继续触发，边缘触发模式则只会触发一次 
  47.         else if (!(epi->event.events & EPOLLET)) { 
  48.             list_add_tail(&epi->rdllink, &ep->rdllist); 
  49.         } 
  50.     } 
  51.  
  52.     return 0; 
  53. } 

ep_send_events_proc主要是把触发的事件复制给调用方，并且根据工作模式和设置的属性对该次事件做进一步处理。至此，epoll的核心逻辑貌似分析完了，但是我们似乎遗留了一个重要的地方，那就是就绪队列的节点是谁又是什么时候插入的呢?

4 事件就绪

我们接着看资源有事件触发的时候是如何通知epoll的。这里以eventfd的eventfd_write为例，即写入的时候。

  1. static ssize_t eventfd_write(struct file *file, const char __user *buf, size_t count, 
  2.                  loff_t *ppos){ 
  3.     struct eventfd_ctx *ctx = file->private_data; 
  4.     ssize_t res; 
  5.     __u64 ucnt; 
  6.     spin_lock_irq(&ctx->wqh.lock); 
  7.     res = -EAGAIN; 
  8.     // 可写 
  9.     if (ULLONG_MAX - ctx->count > ucnt) 
  10.         res = sizeof(ucnt); 
  11.     // 写成功 
  12.     if (likely(res > 0)) { 
  13.         ctx->count += ucnt; 
  14.         // 队列非空，即操作epoll时，epoll注册的节点 
  15.         if (waitqueue_active(&ctx->wqh)) 
  16.             // ”唤醒“它，有数据可写 
  17.             wake_up_locked_poll(&ctx->wqh, EPOLLIN); 
  18.     } 
  19.     spin_unlock_irq(&ctx->wqh.lock); 
  20.  
  21.     return res; 
  22. } 

eventfd_write写入数据后，会通知等待该资源的节点，我们看看wake_up_locked_poll。

  1. #define wake_up_locked_poll(x, m)                       \ 
  2.     __wake_up_locked_key((x), TASK_NORMAL, poll_to_key(m)) 
  3.  
  4. void __wake_up_locked_key(struct wait_queue_head *wq_head, unsigned int mode, void *key){ 
  5.     __wake_up_common(wq_head, mode, 1, 0, key, NULL); 
  6. } 
  7.  
  8. static int __wake_up_common(struct wait_queue_head *wq_head, unsigned int mode, int nr_exclusive, int wake_flags, void *key, wait_queue_entry_t *bookmark){ 
  9.     wait_queue_entry_t *curr, *next; 
  10.     int cnt = 0; 
  11.     // 遍历队列 
  12.     list_for_each_entry_safe_from(curr, next, &wq_head->head, entry) { 
  13.         unsigned flags = curr->flags; 
  14.         int ret; 
  15.         // 执行回调 
  16.         ret = curr->func(curr, mode, wake_flags, key); 
  17.         if (ret < 0) 
  18.             break; 
  19.         // 设置了WQ_FLAG_EXCLUSIVE则只会回调一个，nr_exclusive是1 
  20.         if (ret && (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive) 
  21.             break; 
  22.     } 
  23.  
  24.     return nr_exclusive; 
  25. } 

如果我们有印象，这里执行的回调函数是ep_poll_callback

  1. // 条件满足时，具体资源回调 
  2. static int ep_poll_callback(wait_queue_entry_t *wait, unsigned mode, int sync, void *key){ 
  3.     // 拿到wait关联的epitem 
  4.     struct epitem *epi = ep_item_from_wait(wait); 
  5.     // 再拿到epitem关联的epoll 
  6.     struct eventpoll *ep = epi->ep; 
  7.     // 触发的事件 
  8.     __poll_t pollflags = key_to_poll(key); 
  9.     unsigned long flags; 
  10.     int ewake = 0; 
  11.     // 还没插入队列 
  12.     if (!ep_is_linked(epi)) { 
  13.         // 插入就绪队列 
  14.         if (list_add_tail_lockless(&epi->rdllink, &ep->rdllist)) 
  15.             ep_pm_stay_awake_rcu(epi); 
  16.     } 
  17.  
  18.     // 唤醒阻塞到epoll的进程队列 
  19.     if (waitqueue_active(&ep->wq)) { 
  20.         // “唤醒“阻塞在epoll的进程 
  21.         wake_up(&ep->wq); 
  22.     } 
  23. } 

ep_poll_callback会调用wake_up唤醒阻塞到epoll的进程，我们看看wake_up。

  1. #define wake_up(x)  __wake_up(x, TASK_NORMAL, 1, NULL) 
  2.  
  3. void __wake_up(struct wait_queue_head *wq_head, unsigned int mode, int nr_exclusive, void *key){ 
  4.     __wake_up_common_lock(wq_head, mode, nr_exclusive, 0, key); 
  5. } 
  6.  
  7. static void __wake_up_common_lock(struct wait_queue_head *wq_head, unsigned int mode, int nr_exclusive, int wake_flags, void *key){ 
  8.     unsigned long flags; 
  9.     wait_queue_entry_t bookmark; 
  10.  
  11.     bookmark.flags = 0; 
  12.     bookmark.private = NULL; 
  13.     bookmark.func = NULL; 
  14.     INIT_LIST_HEAD(&bookmark.entry); 
  15.     // 这里只会执行一次 
  16.     do { 
  17.         spin_lock_irqsave(&wq_head->lock, flags); 
  18.         nr_exclusive = __wake_up_common(wq_head, mode, nr_exclusive, 
  19.                         wake_flags, key, &bookmark); 
  20.         spin_unlock_irqrestore(&wq_head->lock, flags); 
  21.     } while (bookmark.flags & WQ_FLAG_BOOKMARK); 
  22. } 

核心逻辑是wake_up_common。wake_up_common函数的代码刚才已经贴过，但是因为这个函数的逻辑很重要，这里再简单贴一下。

  1. static int __wake_up_common(struct wait_queue_head *wq_head, unsigned int mode, 
  2.             int nr_exclusive, int wake_flags, void *key, 
  3.             wait_queue_entry_t *bookmark){ 
  4.     wait_queue_entry_t *curr, *next; 
  5.     int cnt = 0; 
  6.     // 头指针所在结构体 
  7.     curr = list_first_entry(&wq_head->head, wait_queue_entry_t, entry); 
  8.     // 遍历队列 
  9.     list_for_each_entry_safe_from(curr, next, &wq_head->head, entry) { 
  10.         unsigned flags = curr->flags; 
  11.         int ret; 
  12.         // 执行回调 
  13.         ret = curr->func(curr, mode, wake_flags, key); 
  14.         // 设置了WQ_FLAG_EXCLUSIVE则只执行一次，即只唤醒一个进程，解决惊群问题 
  15.         if (ret && (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive) 
  16.             break; 
  17.     } 
  18.  
  19.     return nr_exclusive; 
  20. } 

那么这里的回调又是什么呢?如果我还记得init_wait函数的话，就会知道，init_wait函数只在epoll_wait的时候执行的，用于记录阻塞于epoll的进程队列。

  1.  
  2. init_wait(&wait); 

我们看到函数是autoremove_wake_function。

  1. int autoremove_wake_function(struct wait_queue_entry *wq_entry, unsigned mode, int sync, void *key){ 
  2.     int ret = default_wake_function(wq_entry, mode, sync, key); 
  3.  
  4.     if (ret) 
  5.         list_del_init_careful(&wq_entry->entry); 
  6.  
  7.     return ret; 
  8. } 
  9.  
  10. int default_wake_function(wait_queue_entry_t *curr, unsigned mode, int wake_flags, 
  11.               void *key){ 
  12.     // curr->private为进程pcb即task_struct 
  13.     return try_to_wake_up(curr->private, mode, wake_flags); 
  14. } 
  15.  
  16. static inttry_to_wake_up(struct task_struct *p, unsigned int state, int wake_flags){ 
  17.     ttwu_runnable(p, wake_flags)； 
  18. } 
  19.  
  20. static int ttwu_runnable(struct task_struct *p, int wake_flags){ 
  21.     ttwu_do_wakeup(rq, p, wake_flags, &rf); 
  22. } 
  23.  
  24. static void ttwu_do_wakeup(struct rq *rq, struct task_struct *p, int wake_flags, 
  25.                struct rq_flags *rf){ 
  26.     p->state = TASK_RUNNING; 
  27. } 

autoremove_wake_function流程很长，最终设置进程为就绪状态。

5 监听epoll

监听epoll，这不仅是个非常有意思的功能，同时是一个很有意思的思想。把epoll本身抽象为一种资源。但是这种场景貌似还没有见过。下面我们看一下实现。epoll可以被epoll监听，也可以被poll(早期的io复用机制)监听。不过这种场景貌似很少，有epoll，为什么还会用poll呢?我能想到的场景就是业务代码早期用poll实现的，后期有了epoll，又不能改旧代码，所以就用poll来监听epoll，anyway，我们大概先了解一下实现，被epoll和poll监听是两种不同的情况，虽然代码是一样的，我们分来看。

5.1 poll监听epoll

要被poll监听，就需要实现poll钩子，我们从epoll实现的poll钩子ep_eventpoll_poll开始分析。

  1. static __poll_t ep_eventpoll_poll(struct file *file, poll_table *wait){ 
  2.     // 被监听的epoll 
  3.     struct eventpoll *ep = file->private_data; 
  4.     int depth = 0; 
  5.     // 熟悉的操作 
  6.     poll_wait(file, &ep->poll_wait, wait); 
  7.     // 判断当前有没有事件触发 
  8.     return ep_scan_ready_list(ep, ep_read_events_proc, 
  9.                   &depth, depth, false); 
  10. } 

poll_wait我们已经分析过了就不再分析，ep_scan_ready_list我们也分析过了，主要逻辑是在里面执行函数，ep_read_events_proc，我们看一下ep_read_events_proc是如何判断被监听的epoll中是否有事件触发的。

  1. static __poll_t ep_read_events_proc(struct eventpoll *ep, struct list_head *head, 
  2.                    void *priv){ 
  3.     struct epitem *epi, *tmp; 
  4.     poll_table pt; 
  5.     int depth = *(int *)priv; 
  6.  
  7.     init_poll_funcptr(&pt, NULL); 
  8.     depth++; 
  9.     // 遍历就绪队列 
  10.     list_for_each_entry_safe(epi, tmp, head, rdllink) { 
  11.         // ep_item_poll判断epitem中实现有事件触发 
  12.         if (ep_item_poll(epi, &pt, depth)) { 
  13.             return EPOLLIN | EPOLLRDNORM; 
  14.         }  
  15.     } 
  16.     return 0; 
  17. } 

5.2 epoll监听epoll

epoll监听epoll和epoll监听一般的fd是一样的，区别在于插入的时候，poll逻辑的实现。具体逻辑在ep_item_poll。

  1. static __poll_t ep_item_poll(const struct epitem *epi, poll_table *pt, 
  2.                  int depth){ 
  3.     struct eventpoll *ep; 
  4.     bool locked; 
  5.  
  6.     pt->_key = epi->event.events; 
  7.     // 不是epoll，则执行钩子函数poll 
  8.     if (!is_file_epoll(epi->ffd.file)) 
  9.         return vfs_poll(epi->ffd.file, pt) & epi->event.events; 
  10.     // 在epoll里监听另一个epoll，即epitem的fd是另一个epoll对应的fd 
  11.     // 是epoll则首先取得原始epoll的核心数据结构eventpoll 
  12.     ep = epi->ffd.file->private_data; 
  13.     // 执行pt中的函数 
  14.     poll_wait(epi->ffd.file, &ep->poll_wait, pt); 
  15.     locked = pt && (pt->_qproc == ep_ptable_queue_proc); 
  16.     // 判断是否有事件触发 
  17.     return ep_scan_ready_list(epi->ffd.file->private_data, 
  18.                   ep_read_events_proc, &depth, depth, 
  19.                   locked) & epi->event.events; 
  20. } 

我们看到ep_item_poll中做了一个判断，被poll的是epoll还是非epoll，是epoll的时候则进入另一种逻辑，不过操作和一般fd的情况是一样的，区别只是操作的具体数据结构。这里的逻辑看起来似乎可以放到ep_eventpoll_poll里，但是内核开发者没有这样做。这部分就先不深入分析，因为我们主要是要理解epoll的基础原理。

6 实现支持epoll机制的模块

最后我们实现一个简单的支持epoll机制的模块。该模块实现了一种通知机制，逻辑非常简单，如果值为0则可写，非0则可写，并通过这个条件约束进程的状态。实现进程的简单通信，具体可参考eventfd机制。

  1. struct demo_context { 
  2.     wait_queue_head_t head; 
  3.     unsigned int count; 
  4. }; 
  5.  
  6. static ssize_t demo_read(struct kiocb *iocb, struct iov_iter *to){ 
  7.     struct file *file = iocb->ki_filp; 
  8.     struct demo_context *ctx = file->private_data; 
  9.     spin_lock_irq(&ctx->head.lock); 
  10.     for (;;) { 
  11.         if (ctx->count == 0) { 
  12.             spin_unlock_irq(&ctx->head.lock); 
  13.             // 阻塞 
  14.         } else { 
  15.             break; 
  16.         } 
  17.         spin_lock_irq(&ctx->head.lock); 
  18.     } 
  19.     unsigned int ucnt = ctx->count; 
  20.     ctx->count = 0; 
  21.     if (waitqueue_active(&ctx->wqh)) 
  22.         wake_up_locked_poll(&ctx->wqh, EPOLLOUT); 
  23.     spin_unlock_irq(&ctx->head.lock); 
  24.     if (unlikely(copy_to_iter(&ucnt, sizeof(ucnt), to) != sizeof(ucnt))) 
  25.         return -EFAULT; 
  26.  
  27.     return sizeof(ucnt); 
  28. } 
  29.  
  30. static __poll_t demo_poll(struct file *file, poll_table *wait){ 
  31.     struct demo_context *ctx = file->private_data; 
  32.     __poll_t events = 0; 
  33.     unsigned int count; 
  34.  
  35.     poll_wait(file, &ctx->head, wait); 
  36.  
  37.     count = READ_ONCE(ctx->count); 
  38.  
  39.     if (count == 0) 
  40.         events |= EPOLLOUT; 
  41.     else 
  42.         events |= EPOLLIN; 
  43.  
  44.     return events; 
  45. } 
  46.  
  47. static ssize_t demo_write(struct file *file, const char __user *buf, size_t count, 
  48.                  loff_t *ppos){ 
  49.     struct demo_context *ctx = file->private_data; 
  50.     ssize_t res; 
  51.     unsigned int ucnt; 
  52.  
  53.     if (copy_from_user(&ucnt, buf, sizeof(ucnt))) 
  54.         return -EFAULT; 
  55.  
  56.     spin_lock_irq(&ctx->head.lock); 
  57.     for (;;) { 
  58.         if (ctx->count != 0) { 
  59.             spin_unlock_irq(&ctx->head.lock); 
  60.             // 阻塞 
  61.         } else { 
  62.             break; 
  63.         } 
  64.         spin_lock_irq(&ctx->head.lock); 
  65.     } 
  66.     ctx->count = ucnt; 
  67.     if (waitqueue_active(&ctx->wqh)) 
  68.         wake_up_locked_poll(&ctx->wqh, EPOLLIN); 
  69.     spin_unlock_irq(&ctx->head.lock); 
  70.  
  71.     return res; 
  72. } 
  73.  
  74. static const struct file_operations demo_fops = { 
  75.     .poll       = demo_poll, 
  76.     .read_iter  = demo_read, 
  77.     .write      = demo_write, 
  78. }; 
  79.  
  80. static int do_demo(unsigned int count){ 
  81.     struct demo_context *ctx = kmalloc(sizeof(*ctx), GFP_KERNEL); 
  82.     int fd = get_unused_fd_flags(flags); 
  83.     struct file *file = anon_inode_getfile("[demo]", &demo_fops, ctx, flags);; 
  84.     init_waitqueue_head(&ctx->head); 
  85.     ctx->count = count; 
  86.     fd_install(fd, file); 
  87.     return fd;}SYSCALL_DEFINE1(demo, unsigned int, count){ 
  88.     return do_demo(count); 
  89. } 

至此，真的分析完了，epoll的代码一共2522行，但是还涉及了操作系统中的很多代码，是非常复杂的模块，epoll不做具体的处理逻辑，他只是提供一种机制，遵循这种机制的资源(实现poll钩子)，都可以被监听。我们看到epoll的代码不仅复杂，而且关系非常绕，在epoll中，有几个概念我们需要了解。

1 进程

2 资源(比如网络、管道、eventfd)

3 epoll

4 epitem(管理一个被监听的项)

我们看看他们的关系。

后记：epoll作为一种机制，作用远远超过了它的代码量，存量和以后新增的模块都可以使用这种机制。比如管道、TCP、新增的eventfd等等。从中我们也看到了epoll本身的一些知识，比如他为什么高效、水平触发和边缘触发、epoll本身如何解决惊群现象。也看到了在简单的API使用下是如此复杂的实现。另外更有意思的是epoll也支持监听另外一个epoll，因为epoll也可以被当作一种资源。最后，本文不是epoll的全部，因为涉及的细节实在太多，感兴趣的同学可以自己研究一下，网上也有很多优秀的文章。

更多文章参考：https://github.com/theanarkh/read-linux5.9.9-code

来源：编程杂技内容投诉

免责声明：

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题

  • 2024年上半年信息系统项目管理师第二批次真题及答案解析（完整版）

    难度     813人已做
    查看

  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看

  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看

  • 2024年上半年软考高项第一、二批次真题考点汇总（完整版）

    难度     435人已做
    查看

  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机

从内核看Epoll的实现

后端开发Epoll服务器2024-12-03

从内核看IO_Uring的实现（一）

后端开发内核IO2024-12-03

从内核看文件描述符传递的实现

后端开发内核文件2024-12-03

图解 | Epoll怎么实现的

后端开发Epoll2024-12-03

深入了解Linux内核中的Epoll是如何工作的

后端开发EpollLinux2024-11-30

一张图看懂Linux内核中Percpu变量的实现

后端开发Linux内核2024-12-03

python 查看cpu的核数实现

后端开发2024-04-02

操作系统内核的进化之路：从微内核到宏内核的变迁

操作系统内核的进化之路：从微内核到宏内核的变迁
后端开发2024-04-02

如何实现4G内存服务器epoll并发量最大

后端开发2023-06-13

从C++ Addon看Napi的实现

后端开发C++Napi2024-12-03

Linux内核中的IPSEC实现(3)

后端开发2023-01-31

Linux内核如何实现多核模式

后端开发2023-06-17

ubuntu查看已安装的内核

后端开发2024-04-02

Linux 内核网络之 Shutdown 的实现

后端开发Linuxshutdown2024-11-30

Linux 内核网络之 connect 的实现

后端开发Linuxconnect2024-11-30

Linux 内核网络之 Listen 的实现

后端开发Linuxlisten2024-11-30

[Android]从零开始的内核编译

后端开发2023-08-17

聊聊Netty那些事儿之从内核角度看IO模型

后端开发阻塞与非阻塞同步与异步2024-12-02

如何实现LINUX系统2.4内核升级到2.6内核

后端开发2023-06-10

wireguardNT：Windows 内核下的高性能 wireguard 实现

后端开发WireGuardWindows内核2024-12-03
位置：首页-资讯-后端开发
咦！没有更多了？去看看其它编程学习网 内容吧
首页课程
资料下载问答资讯