文章详情

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

请输入下面的图形验证码

提交验证

短信预约提醒成功

从Linux源码看Epoll

2024-12-03 02:11

关注

本文就是笔者在探究epoll源码过程中,对kernel将就绪描述符添加到epoll并唤醒对应进程的一次源码分析(基于linux-2.6.32内核版本)。由于篇幅所限,笔者聚焦于tcp协议下socket可读事件的源码分析。

简单的epoll例子

下面的例子,是从笔者本人用c语言写的dbproxy中的一段代码。由于细节过多,所以做了一些删减。

  1. int init_reactor(int listen_fd,int worker_count){ 
  2.     ...... 
  3.     // 创建多个epoll fd,以充分利用多核 
  4.     for(i=0;i
  5.         reactor->worker_fd = epoll_create(EPOLL_MAX_EVENTS); 
  6.     } 
  7.      
  8.     // 将accept后的事件加入到对应的epoll fd中 
  9.     int client_fd = accept(listen_fd,(struct sockaddr *)&client_addr,&client_len))); 
  10.     // 将连接描述符注册到对应的worker里面 
  11.     epoll_ctl(reactor->client_fd,EPOLL_CTL_ADD,epifd,&event); 
  12. // reactor的worker线程 
  13. static void* rw_thread_func(void* arg){ 
  14.     ...... 
  15.  
  16.     for(;;){ 
  17.           // epoll_wait等待事件触发 
  18.         int retval = epoll_wait(epfd,events,EPOLL_MAX_EVENTS,500); 
  19.         if(retval > 0){ 
  20.             for(j=0; j < retval; j++){ 
  21.                 // 处理读事件 
  22.                if(event & EPOLLIN){ 
  23.                  handle_ready_read_connection(conn); 
  24.                  continue
  25.              } 
  26.               
  27.             } 
  28.         } 
  29.     } 
  30.     ...... 

上述代码事实上就是实现了一个reactor模式中的accept与read/write处理线程,如下图所示:

epoll_create

Unix的万物皆文件的思想在epoll里面也有体现,epoll_create调用返回一个文件描述符,此描述符挂载在anon_inode_fs(匿名inode文件系统)的根目录下面。让我们看下具体的epoll_create系统调用源码:

  1. SYSCALL_DEFINE1(epoll_create, intsize
  2.     if (size <= 0) 
  3.         return -EINVAL; 
  4.  
  5.     return sys_epoll_create1(0); 

由上述源码可见,epoll_create的参数是基本没有意义的,kernel简单的判断是否为0,然后就直接就调用了sys_epoll_create1。由于linux的系统调用是通过(SYSCALL_DEFINE1,SYSCALL_DEFINE2……SYSCALL_DEFINE6)定义的,那么sys_epoll_create1对应的源码即是SYSCALL_DEFINE(epoll_create1)。

(注:受限于寄存器数量的限制,(80x86下的)kernel限制系统调用最多有6个参数。据ulk3所述,这是由于32位80x86寄存器的限制)

接下来,我们就看下epoll_create1的源码:

  1. SYSCALL_DEFINE1(epoll_create1, int, flags) 
  2.     // kzalloc(sizeof(*ep), GFP_KERNEL),用的是内核空间 
  3.     error = ep_alloc(&ep); 
  4.     // 获取尚未被使用的文件描述符,即描述符数组的槽位 
  5.     fd = get_unused_fd_flags(O_RDWR | (flags & O_CLOEXEC)); 
  6.     // 在匿名inode文件系统中分配一个inode,并得到其file结构体 
  7.     // 且file->f_op = &eventpoll_fops 
  8.     // 且file->private_data = ep; 
  9.     file = anon_inode_getfile("[eventpoll]", &eventpoll_fops, ep, 
  10.                  O_RDWR | (flags & O_CLOEXEC)); 
  11.     // 将file填入到对应的文件描述符数组的槽里面 
  12.     fd_install(fd,file); 
  13.     ep->file = file; 
  14.     return fd; 

最后epoll_create生成的文件描述符如下图所示:

struct eventpoll

所有的epoll系统调用都是围绕eventpoll结构体做操作,现简要描述下其中的成员:

  1.  
  2. struct eventpoll { 
  3.     // 自旋锁,在kernel内部用自旋锁加锁,就可以同时多线(进)程对此结构体进行操作 
  4.     // 主要是保护ready_list 
  5.     spinlock_t lock; 
  6.     // 这个互斥锁是为了保证在eventloop使用对应的文件描述符的时候,文件描述符不会被移除掉 
  7.     struct mutex mtx; 
  8.     // epoll_wait使用的等待队列,和进程唤醒有关 
  9.     wait_queue_head_t wq; 
  10.     // file->poll使用的等待队列,和进程唤醒有关 
  11.     wait_queue_head_t poll_wait; 
  12.     // 就绪的描述符队列 
  13.     struct list_head rdllist; 
  14.     // 通过红黑树来组织当前epoll关注的文件描述符 
  15.     struct rb_root rbr; 
  16.     // 在向用户空间传输就绪事件的时候,将同时发生事件的文件描述符链入到这个链表里面 
  17.     struct epitem *ovflist; 
  18.     // 对应的user 
  19.     struct user_struct *user
  20.     // 对应的文件描述符 
  21.     struct file *file; 
  22.     // 下面两个是用于环路检测的优化 
  23.     int visited; 
  24.     struct list_head visited_list_link; 
  25. }; 

本文讲述的是kernel是如何将就绪事件传递给epoll并唤醒对应进程上,因此在这里主要聚焦于(wait_queue_head_t wq)等成员。

epoll_ctl(add)

我们看下epoll_ctl(EPOLL_CTL_ADD)是如何将对应的文件描述符插入到eventpoll中的。

借助于spin_lock(自旋锁)和mutex(互斥锁),epoll_ctl调用可以在多个KSE(内核调度实体,即进程/线程)中并发执行。

  1. SYSCALL_DEFINE4(epoll_ctl, int, epfd, int, op, int, fd, 
  2.         struct epoll_event __user *, event) 
  3.      
  4.     // 此处的互斥锁是为了防止并发调用epoll_ctl,即保护内部数据结构 
  5.     // 不会被并发的添加修改删除破坏 
  6.     mutex_lock_nested(&ep->mtx, 0); 
  7.     switch (op) { 
  8.         case EPOLL_CTL_ADD: 
  9.             ... 
  10.             // 插入到红黑树中 
  11.             error = ep_insert(ep, &epds, tfile, fd); 
  12.             ... 
  13.             break; 
  14.         ...... 
  15.     } 
  16.     mutex_unlock(&ep->mtx); 

上述过程如下图所示:

ep_insert

在ep_insert中初始化了epitem,然后初始化了本文关注的焦点,即事件就绪时候的回调函数,代码如下所示:

  1. static int ep_insert(struct eventpoll *ep, struct epoll_event *event, 
  2.              struct file *tfile, int fd) 
  3.      
  4.     // &epq.pt->qproc = ep_ptable_queue_proc 
  5.     init_poll_funcptr(&epq.pt, ep_ptable_queue_proc); 
  6.     // 在这里将回调函数注入 
  7.     revents = tfile->f_op->poll(tfile, &epq.pt); 
  8.     // 如果当前有事件已经就绪,那么一开始就会被加入到ready list 
  9.     // 例如可写事件 
  10.     // 另外,在tcp内部ack之后调用tcp_check_space,最终调用sock_def_write_space来唤醒对应的epoll_wait下的进程 
  11.     if ((revents & event->events) && !ep_is_linked(&epi->rdllink)) { 
  12.         list_add_tail(&epi->rdllink, &ep->rdllist); 
  13.         // wake_up ep对应在epoll_wait下的进程 
  14.         if (waitqueue_active(&ep->wq)){ 
  15.             wake_up_locked(&ep->wq); 
  16.         } 
  17.         ...... 
  18.     } 
  19.     // 将epitem插入红黑树 
  20.     ep_rbtree_insert(ep, epi); 
  21.     ...... 

tfile->f_op->poll的实现

向kernel更底层注册回调函数的是tfile->f_op->poll(tfile, &epq.pt)这一句,我们来看一下对于对应的socket文件描述符,其fd=>file->f_op->poll的初始化过程:

  1. // 将accept后的事件加入到对应的epoll fd中 
  2. int client_fd = accept(listen_fd,(struct sockaddr *)&client_addr,&client_len))); 
  3. // 将连接描述符注册到对应的worker里面 
  4. epoll_ctl(reactor->client_fd,EPOLL_CTL_ADD,epifd,&event); 

回顾一下上述user space代码,fd即client_fd是由tcp的listen_fd通过accept调用而来,那么我们看下accept调用链的关键路径:

  1. accept 
  2.       |->accept4 
  3.             |->sock_attach_fd(newsock, newfile, flags & O_NONBLOCK); 
  4.                   |->init_file(file,...,&socket_file_ops); 
  5.                         |->file->f_op = fop; 
  6.                                
  7.             |->fd_install(newfd, newfile); // 安装fd 

那么,由accept获得的client_fd的结构如下图所示:

(注:由于是tcp socket,所以这边sock->ops=inet_stream_ops,这个初始化的过程在我的另一篇博客<<从linux源码看socket的阻塞和非阻塞>>中,博客地址如下:

https://my.oschina.net/alchemystar/blog/1791017)

既然知道了tfile->f_op->poll的实现,我们就可以看下此poll是如何将安装回调函数的。

回调函数的安装

kernel的调用路径如下:

  1. sock_poll ; 
  2.     |->sock->ops->poll 
  3.         |->tcp_poll 
  4.              
  5.             |->sock_poll_wait(file, sk->sk_sleep, wait); 
  6.                 |->poll_wait 
  7.                     |->p->qproc(filp, wait_address, p); 
  8.                      
  9.                         |-> ep_ptable_queue_proc(filp,wait_address,p); 

绕了一大圈之后,我们的回调函数的安装其实就是调用了eventpoll.c中的ep_ptable_queue_proc,而且向其中传递了sk->sk_sleep作为其waitqueue的head,其源码如下所示:

  1. static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead, 
  2.                  poll_table *pt) 
  3.     // 取出当前client_fd对应的epitem 
  4.     struct epitem *epi = ep_item_from_epqueue(pt); 
  5.     // &pwq->wait->func=ep_poll_callback,用于回调唤醒 
  6.     // 注意,这边不是init_waitqueue_entry,即没有将当前KSE(current,当前进程/线程)写入到 
  7.     // wait_queue当中,因为不一定是从当前安装的KSE唤醒,而应该是唤醒epoll\_wait的KSE 
  8.     init_waitqueue_func_entry(&pwq->wait, ep_poll_callback); 
  9.     // 这边的whead是sk->sk_sleep,将当前的waitqueue链入到socket对应的sleep列表 
  10.     add_wait_queue(whead, &pwq->wait); 

这样client_fd的结构进一步完善,如下图所示:

ep_poll_callback函数是唤醒对应epoll_wait的地方,我们将在后面一起讲述。

epoll_wait

epoll_wait主要是调用了ep_poll:

  1. SYSCALL_DEFINE4(epoll_wait, int, epfd, struct epoll_event __user *, events, 
  2.         int, maxevents, int, timeout) 
  3.      
  4.     // 调用ep_poll 
  5.     error = ep_poll(ep, events, maxevents, timeout); 
  6.     ... 

紧接着,我们看下ep_poll函数:

  1. static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events, 
  2.            int maxevents, long timeout) 
  3.     ...... 
  4. retry: 
  5.     // 获取spinlock 
  6.     spin_lock_irqsave(&ep->lock, flags); 
  7.     // 将当前task_struct写入到waitqueue中以便唤醒 
  8.     // wq_entry->func = default_wake_function; 
  9.     init_waitqueue_entry(&wait, current); 
  10.     // WQ_FLAG_EXCLUSIVE,排他性唤醒,配合SO_REUSEPORT从而解决accept惊群问题 
  11.     wait.flags |= WQ_FLAG_EXCLUSIVE; 
  12.     // 链入到ep的waitqueue中 
  13.     __add_wait_queue(&ep->wq, &wait); 
  14.     for (;;) { 
  15.         // 设置当前进程状态为可打断 
  16.         set_current_state(TASK_INTERRUPTIBLE); 
  17.         // 检查当前线程是否有信号要处理,有则返回-EINTR 
  18.         if (signal_pending(current)) { 
  19.             res = -EINTR; 
  20.             break; 
  21.         } 
  22.         spin_unlock_irqrestore(&ep->lock, flags); 
  23.         // schedule调度,让出CPU 
  24.         jtimeout = schedule_timeout(jtimeout); 
  25.         spin_lock_irqsave(&ep->lock, flags); 
  26.     } 
  27.     // 到这里,表明超时或者有事件触发等动作导致进程重新调度 
  28.     __remove_wait_queue(&ep->wq, &wait); 
  29.     // 设置进程状态为running 
  30.     set_current_state(TASK_RUNNING); 
  31.     ...... 
  32.     // 检查是否有可用事件 
  33.     eavail = !list_empty(&ep->rdllist) || ep->ovflist != EP_UNACTIVE_PTR; 
  34.     ...... 
  35.     // 向用户空间拷贝就绪事件 
  36.     ep_send_events(ep, events, maxevents) 

上述逻辑如下图所示:

ep_send_events

  1. static int ep_scan_ready_list(struct eventpoll *ep, 
  2.                   int (*sproc)(struct eventpoll *, 
  3.                        struct list_head *, void *), 
  4.                   void *priv, 
  5.                   int depth) 
  6.     ... 
  7.     // 将epfd的rdllist链入到txlist 
  8.     list_splice_init(&ep->rdllist, &txlist); 
  9.     ... 
  10.      
  11.     error = (*sproc)(ep, &txlist, priv); 
  12.     ... 
  13.     // 处理ovflist,即在上面sproc过程中又到来的事件 
  14.     ... 

其主要调用了ep_send_events_proc:

  1. static int ep_send_events_proc(struct eventpoll *ep, struct list_head *head, 
  2.                    void *priv) 
  3.     for (eventcnt = 0, uevent = esed->events; 
  4.          !list_empty(head) && eventcnt < esed->maxevents;) { 
  5.        // 遍历ready list 
  6.         epi = list_first_entry(head, struct epitem, rdllink); 
  7.         list_del_init(&epi->rdllink); 
  8.         // readylist只是表明当前epi有事件,具体的事件信息还是得调用对应file的poll 
  9.         // 这边的poll即是tcp_poll,根据tcp本身的信息设置掩码(mask)等信息 & 上兴趣事件掩码,则可以得知当前事件是否是epoll_wait感兴趣的事件 
  10.         revents = epi->ffd.file->f_op->poll(epi->ffd.file, NULL) & 
  11.             epi->event.events; 
  12.         if(revents){ 
  13.              
  14.              
  15.             // 如果不是边缘触发,则将当前的epi重新加回到可用列表中,这样就可以下一次继续触发poll,如果下一次poll的revents不为0,那么用户空间依旧能感知 */ 
  16.             else if (!(epi->event.events & EPOLLET)){ 
  17.                 list_add_tail(&epi->rdllink, &ep->rdllist); 
  18.             } 
  19.              
  20.             eventcnt++ 
  21.         } 
  22.          
  23.         ...... 
  24.     } 
  25.     return eventcnt; 

上述代码逻辑如下所示:

事件到来添加到epoll就绪队列(rdllist)的过程

经过上述章节的详述之后,我们终于可以阐述,tcp在数据到来时是怎么加入到epoll的就绪队列的了。

可读事件到来

首先我们看下tcp数据包从网卡驱动到kernel内部tcp协议处理调用链:

step1:

网络分组到来的内核路径,网卡发起中断后调用netif_rx将事件挂入CPU的等待队列,并唤起软中断(soft_irq),再通过linux的软中断机制调用net_rx_action,如下图所示:

注:上图来自PLKA(<<深入Linux内核架构>>)

step2:

紧接着跟踪next_rx_action

  1. next_rx_action 
  2.     |-process_backlog 
  3.         ...... 
  4.             |->packet_type->func 在这里我们考虑ip_rcv 
  5.                     |->ipprot->handler 在这里ipprot重载为tcp_protocol 
  6.                         (handler 即为tcp_v4_rcv) 

我们再看下对应的tcp_v4_rcv

  1. tcp_v4_rcv 
  2.       |->tcp_v4_do_rcv 
  3.             |->tcp_rcv_state_process 
  4.                   |->tcp_data_queue 
  5.                         |-> sk->sk_data_ready(sock_def_readable) 
  6.                               |->wake_up_interruptible_sync_poll(sk->sleep,...) 
  7.                                     |->__wake_up 
  8.                                           |->__wake_up_common 
  9.                                                 |->curr->func 
  10.                                                  
  11.                                                       |->ep_poll_callback 

这样,我们就看下最终唤醒epoll_wait的ep_poll_callback函数:

  1. static int ep_poll_callback(wait_queue_t *wait, unsigned mode, int sync, void *key
  2.     // 获取wait对应的epitem 
  3.     struct epitem *epi = ep_item_from_wait(wait); 
  4.     // epitem对应的eventpoll结构体 
  5.     struct eventpoll *ep = epi->ep; 
  6.     // 获取自旋锁,保护ready_list等结构 
  7.     spin_lock_irqsave(&ep->lock, flags); 
  8.     // 如果当前epi没有被链入ep的ready list,则链入 
  9.     // 这样,就把当前的可用事件加入到epoll的可用列表了 
  10.     if (!ep_is_linked(&epi->rdllink)) 
  11.         list_add_tail(&epi->rdllink, &ep->rdllist); 
  12.     // 如果有epoll_wait在等待的话,则唤醒这个epoll_wait进程 
  13.     // 对应的&ep->wq是在epoll_wait调用的时候通过init_waitqueue_entry(&wait, current)而生成的 
  14.     // 其中的current即是对应调用epoll_wait的进程信息task_struct 
  15.     if (waitqueue_active(&ep->wq)) 
  16.         wake_up_locked(&ep->wq); 

上述过程如下图所示:

最后wake_up_locked调用__wake_up_common,然后调用了在init_waitqueue_entry注册的default_wake_function,调用路径为:

  1. wake_up_locked 
  2.     |->__wake_up_common 
  3.         |->default_wake_function 
  4.             |->try_wake_up (wake up a thread) 
  5.                 |->activate_task 
  6.                     |->enqueue_task    running 

将epoll_wait进程推入可运行队列,等待内核重新调度进程,然后epoll_wait对应的这个进程重新运行后,就从schedule恢复,继续下面的ep_send_events(向用户空间拷贝事件并返回)。

wake_up过程如下图所示:

可写事件到来

可写事件的运行过程和可读事件大同小异:

首先,在epoll_ctl_add的时候预先会调用一次对应文件描述符的poll,如果返回事件里有可写掩码的时候直接调用wake_up_locked以唤醒对应的epoll_wait进程。

然后,在tcp在底层驱动有数据到来的时候可能携带了ack从而可以释放部分已经被对端接收的数据,于是触发可写事件,这一部分的调用链为:

  1. tcp_input.c 
  2. tcp_v4_rcv 
  3.     |-tcp_v4_do_rcv 
  4.         |-tcp_rcv_state_process 
  5.             |-tcp_data_snd_check 
  6.                 |->tcp_check_space 
  7.                     |->tcp_new_space 
  8.                         |->sk->sk_write_space 
  9.                          

最后在此函数里面sk_stream_write_space唤醒对应的epoll_wait进程

  1. void sk_stream_write_space(struct sock *sk) 
  2.     // 即有1/3可写空间的时候才触发可写事件 
  3.     if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk) && sock) { 
  4.         clear_bit(SOCK_NOSPACE, &sock->flags); 
  5.  
  6.         if (sk->sk_sleep && waitqueue_active(sk->sk_sleep)) 
  7.             wake_up_interruptible_poll(sk->sk_sleep, POLLOUT | 
  8.                         POLLWRNORM | POLLWRBAND) 
  9.         ...... 
  10.     } 

关闭描述符(close fd)

值得注意的是,我们在close对应的文件描述符的时候,会自动调用eventpoll_release将对应的file从其关联的epoll_fd中删除,kernel关键路径如下:

  1. close fd 
  2.       |->filp_close 
  3.             |->fput 
  4.                   |->__fput 
  5.                         |->eventpoll_release 
  6.                               |->ep_remove 

所以我们在关闭对应的文件描述符后,并不需要通过epoll_ctl_del来删掉对应epoll中相应的描述符。

总结

epoll作为linux下非常优秀的事件触发机制得到了广泛的运用。其源码还是比较复杂的,本文只是阐述了epoll读写事件的触发机制,探究linux kernel源码的过程非常快乐^_^

本文转载自微信公众号「解Bug之路」,可以通过以下二维码关注。转载本文请联系解Bug之路公众号。

 

来源: 解Bug之路内容投诉

免责声明:

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

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

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯