个人主页网站设计代码,厦门外贸网站建设报价,网站商城app建设方案,wordpress用的编辑器外接从linux源码看epoll前言在linux的高性能网络编程中#xff0c;绕不开的就是epoll。和select、poll等系统调用相比,epoll在需要监视大量文件描述符并且其中只有少数活跃的时候#xff0c;表现出无可比拟的优势。epoll能让内核记住所关注的描述符#xff0c;并在对应的描述符事…从linux源码看epoll前言在linux的高性能网络编程中绕不开的就是epoll。和select、poll等系统调用相比,epoll在需要监视大量文件描述符并且其中只有少数活跃的时候表现出无可比拟的优势。epoll能让内核记住所关注的描述符并在对应的描述符事件就绪的时候,在epoll的就绪链表中添加这些就绪元素并唤醒对应的epoll等待进程。 本文就是笔者在探究epoll源码过程中对kernel将就绪描述符添加到epoll并唤醒对应进程的一次源码分析(基于linux-2.6.32内核版本)。由于篇幅所限笔者聚焦于tcp协议下socket可读事件的源码分析。简单的epoll例子给大家分享一个Linux下支撑亿级io的底层基石 epoll实战视频支撑亿级io的底层基石 epoll实战揭秘下面的例子是从笔者本人用c语言写的dbproxy中的一段代码。由于细节过多所以做了一些删减。int init_reactor(int listen_fd,int worker_count){......// 创建多个epoll fd以充分利用多核for(i0;iworker_count;i){reactor-worker_fd epoll_create(EPOLL_MAX_EVENTS);}/* epoll add listen_fd and accept */// 将accept后的事件加入到对应的epoll fd中int client_fd accept(listen_fd,(struct sockaddr *)client_addr,client_len)));// 将连接描述符注册到对应的worker里面epoll_ctl(reactor-client_fd,EPOLL_CTL_ADD,epifd,event);
}
// reactor的worker线程
static void* rw_thread_func(void* arg){......for(;;){// epoll_wait等待事件触发int retval epoll_wait(epfd,events,EPOLL_MAX_EVENTS,500);if(retval 0){for(j0; j retval; j){// 处理读事件if(event EPOLLIN){handle_ready_read_connection(conn);continue;}/* 处理其它事件 */}}}......
}
上述代码事实上就是实现了一个reactor模式中的accept与read/write处理线程如下图所示:需要c/cLinux后台服务器开发高级视频学习资料的朋友可以加qun720209036获取 我们在腾讯课堂上每晚都有一个免费的直播课程。内容包括C/CLinuxNginxZeroMQMySQLRedisMongoDBZK流媒体P2PK8SDockerTCP/IP协程DPDK多个高级知识点分享。这是我们【免费的】腾讯课堂链接 不需要注册点跳过C/CLinux服务器开发/后台架构师【零声学院】-学习视频教程-腾讯课堂附上一份大纲学习图epoll_createUnix的万物皆文件的思想在epoll里面也有体现epoll_create调用返回一个文件描述符此描述符挂载在anon_inode_fs(匿名inode文件系统)的根目录下面。让我们看下具体的epoll_create系统调用源码:SYSCALL_DEFINE1(epoll_create, int, size)
{if (size 0)return -EINVAL;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的源码:SYSCALL_DEFINE1(epoll_create1, int, flags)
{// kzalloc(sizeof(*ep), GFP_KERNEL),用的是内核空间error ep_alloc(ep);// 获取尚未被使用的文件描述符即描述符数组的槽位fd get_unused_fd_flags(O_RDWR | (flags O_CLOEXEC));// 在匿名inode文件系统中分配一个inode,并得到其file结构体// 且file-f_op eventpoll_fops// 且file-private_data ep;file anon_inode_getfile([eventpoll], eventpoll_fops, ep,O_RDWR | (flags O_CLOEXEC));// 将file填入到对应的文件描述符数组的槽里面fd_install(fd,file); ep-file file;return fd;
}
最后epoll_create生成的文件描述符如下图所示:struct eventpoll所有的epoll系统调用都是围绕eventpoll结构体做操作,现简要描述下其中的成员:/** 此结构体存储在file-private_data中*/
struct eventpoll {// 自旋锁在kernel内部用自旋锁加锁就可以同时多线(进)程对此结构体进行操作// 主要是保护ready_listspinlock_t lock;// 这个互斥锁是为了保证在eventloop使用对应的文件描述符的时候文件描述符不会被移除掉struct mutex mtx;// epoll_wait使用的等待队列和进程唤醒有关wait_queue_head_t wq;// file-poll使用的等待队列和进程唤醒有关wait_queue_head_t poll_wait;// 就绪的描述符队列struct list_head rdllist;// 通过红黑树来组织当前epoll关注的文件描述符struct rb_root rbr;// 在向用户空间传输就绪事件的时候将同时发生事件的文件描述符链入到这个链表里面struct epitem *ovflist;// 对应的userstruct user_struct *user;// 对应的文件描述符struct file *file;// 下面两个是用于环路检测的优化int visited;struct list_head visited_list_link;
};
本文讲述的是kernel是如何将就绪事件传递给epoll并唤醒对应进程上因此在这里主要聚焦于(wait_queue_head_t wq)等成员。epoll_ctl(add)我们看下epoll_ctl(EPOLL_CTL_ADD)是如何将对应的文件描述符插入到eventpoll中的。 借助于spin_lock(自旋锁)和mutex(互斥锁),epoll_ctl调用可以在多个KSE(内核调度实体,即进程/线程)中并发执行。SYSCALL_DEFINE4(epoll_ctl, int, epfd, int, op, int, fd,struct epoll_event __user *, event)
{/* 校验epfd是否是epoll的描述符 */// 此处的互斥锁是为了防止并发调用epoll_ctl,即保护内部数据结构// 不会被并发的添加修改删除破坏mutex_lock_nested(ep-mtx, 0);switch (op) {case EPOLL_CTL_ADD:...// 插入到红黑树中error ep_insert(ep, epds, tfile, fd);...break;......}mutex_unlock(ep-mtx);
}
上述过程如下图所示ep_insert在ep_insert中初始化了epitem然后初始化了本文关注的焦点,即事件就绪时候的回调函数代码如下所示:static int ep_insert(struct eventpoll *ep, struct epoll_event *event,struct file *tfile, int fd)
{/* 初始化epitem */// epq.pt-qproc ep_ptable_queue_procinit_poll_funcptr(epq.pt, ep_ptable_queue_proc);// 在这里将回调函数注入revents tfile-f_op-poll(tfile, epq.pt);// 如果当前有事件已经就绪那么一开始就会被加入到ready list// 例如可写事件// 另外在tcp内部ack之后调用tcp_check_space,最终调用sock_def_write_space来唤醒对应的epoll_wait下的进程if ((revents event-events) !ep_is_linked(epi-rdllink)) {list_add_tail(epi-rdllink, ep-rdllist);// wake_up ep对应在epoll_wait下的进程if (waitqueue_active(ep-wq)){wake_up_locked(ep-wq);}......} // 将epitem插入红黑树ep_rbtree_insert(ep, epi);......
}
tfile-f_op-poll的实现向kernel更底层注册回调函数的是tfile-f_op-poll(tfile, epq.pt)这一句我们来看一下对于对应的socket文件描述符其fdfile-f_op-poll的初始化过程// 将accept后的事件加入到对应的epoll fd中int client_fd accept(listen_fd,(struct sockaddr *)client_addr,client_len)));// 将连接描述符注册到对应的worker里面epoll_ctl(reactor-client_fd,EPOLL_CTL_ADD,epifd,event);
回顾一下上述user space代码fd即client_fd是由tcp的listen_fd通过accept调用而来那么我们看下accept调用链的关键路径:accept|-accept4|-sock_attach_fd(newsock, newfile, flags O_NONBLOCK);|-init_file(file,...,socket_file_ops);|-file-f_op fop;/* file-f_op socket_file_ops */|-fd_install(newfd, newfile); // 安装fd
那么由accept获得的client_fd的结构如下图所示:既然知道了tfile-f_op-poll的实现我们就可以看下此poll是如何将安装回调函数的。回调函数的安装kernel的调用路径如下:sock_poll /*tfile-f_op-poll(tfile, epq.pt)*/;|-sock-ops-poll|-tcp_poll/* 这边重要的是拿到了sk_sleep用于KSE(进程/线程)的唤醒 */|-sock_poll_wait(file, sk-sk_sleep, wait);|-poll_wait|-p-qproc(filp, wait_address, p);/* p为epq.pt,而且epq.pt-qproc ep_ptable_queue_proc*/|- ep_ptable_queue_proc(filp,wait_address,p);
绕了一大圈之后我们的回调函数的安装其实就是调用了eventpoll.c中的ep_ptable_queue_proc,而且向其中传递了sk-sk_sleep作为其waitqueue的head,其源码如下所示:static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead,poll_table *pt)
{// 取出当前client_fd对应的epitemstruct epitem *epi ep_item_from_epqueue(pt);// pwq-wait-funcep_poll_callback用于回调唤醒// 注意这边不是init_waitqueue_entry,即没有将当前KSE(current,当前进程/线程)写入到// wait_queue当中因为不一定是从当前安装的KSE唤醒而应该是唤醒epoll_wait的KSEinit_waitqueue_func_entry(pwq-wait, ep_poll_callback);// 这边的whead是sk-sk_sleep,将当前的waitqueue链入到socket对应的sleep列表add_wait_queue(whead, pwq-wait);
}
这样client_fd的结构进一步完善如下图所示:ep_poll_callback函数是唤醒对应epoll_wait的地方我们将在后面一起讲述。epoll_waitepoll_wait主要是调用了ep_poll:SYSCALL_DEFINE4(epoll_wait, int, epfd, struct epoll_event __user *, events,int, maxevents, int, timeout)
{/* 检查epfd是否是epoll_create创建的fd */// 调用ep_pollerror ep_poll(ep, events, maxevents, timeout);...
}
紧接着我们看下ep_poll函数:static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,int maxevents, long timeout)
{......
retry:// 获取spinlockspin_lock_irqsave(ep-lock, flags);// 将当前task_struct写入到waitqueue中以便唤醒// wq_entry-func default_wake_function;init_waitqueue_entry(wait, current);// WQ_FLAG_EXCLUSIVE排他性唤醒配合SO_REUSEPORT从而解决accept惊群问题wait.flags | WQ_FLAG_EXCLUSIVE;// 链入到ep的waitqueue中__add_wait_queue(ep-wq, wait);for (;;) {// 设置当前进程状态为可打断set_current_state(TASK_INTERRUPTIBLE);// 检查当前线程是否有信号要处理有则返回-EINTRif (signal_pending(current)) {res -EINTR;break;}spin_unlock_irqrestore(ep-lock, flags);// schedule调度让出CPUjtimeout schedule_timeout(jtimeout);spin_lock_irqsave(ep-lock, flags);}// 到这里表明超时或者有事件触发等动作导致进程重新调度__remove_wait_queue(ep-wq, wait);// 设置进程状态为runningset_current_state(TASK_RUNNING);......// 检查是否有可用事件eavail !list_empty(ep-rdllist) || ep-ovflist ! EP_UNACTIVE_PTR;......// 向用户空间拷贝就绪事件ep_send_events(ep, events, maxevents)
}
上述逻辑如下图所示:ep_send_eventsep_send_events函数主要就是调用了ep_scan_ready_list,顾名思义ep_scan_ready_list就是扫描就绪列表:static int ep_scan_ready_list(struct eventpoll *ep,int (*sproc)(struct eventpoll *,struct list_head *, void *),void *priv,int depth)
{...// 将epfd的rdllist链入到txlistlist_splice_init(ep-rdllist, txlist);.../* sproc ep_send_events_proc */error (*sproc)(ep, txlist, priv);...// 处理ovflist,即在上面sproc过程中又到来的事件...
}
其主要调用了ep_send_events_proc:static int ep_send_events_proc(struct eventpoll *ep, struct list_head *head,void *priv)
{for (eventcnt 0, uevent esed-events;!list_empty(head) eventcnt esed-maxevents;) {// 遍历ready list epi list_first_entry(head, struct epitem, rdllink);list_del_init(epi-rdllink);// readylist只是表明当前epi有事件具体的事件信息还是得调用对应file的poll// 这边的poll即是tcp_poll,根据tcp本身的信息设置掩码(mask)等信息 上兴趣事件掩码则可以得知当前事件是否是epoll_wait感兴趣的事件revents epi-ffd.file-f_op-poll(epi-ffd.file, NULL) epi-event.events;if(revents){/* 将event放入到用户空间 *//* 处理ONESHOT逻辑 */// 如果不是边缘触发则将当前的epi重新加回到可用列表中这样就可以下一次继续触发poll,如果下一次poll的revents不为0那么用户空间依旧能感知 */else if (!(epi-event.events EPOLLET)){list_add_tail(epi-rdllink, ep-rdllist);}/* 如果是边缘触发那么就不加回可用列表因此只能等到下一个可用事件触发的时候才会将对应的epi放到可用列表里面*/eventcnt}/* 如poll出来的revents事件epoll_wait不感兴趣(或者本来就没有事件)那么也不会加回到可用列表 */......}return eventcnt;
}
上述代码逻辑如下所示:事件到来添加到epoll就绪队列(rdllist)的过程经过上述章节的详述之后我们终于可以阐述tcp在数据到来时是怎么加入到epoll的就绪队列的了。可读事件到来首先我们看下tcp数据包从网卡驱动到kernel内部tcp协议处理调用链:step1:网络分组到来的内核路径网卡发起中断后调用netif_rx将事件挂入CPU的等待队列并唤起软中断(soft_irq)再通过linux的软中断机制调用net_rx_action如下图所示:注:上图来自PLKA(深入Linux内核架构)step2:紧接着跟踪next_rx_actionnext_rx_action|-process_backlog......|-packet_type-func 在这里我们考虑ip_rcv|-ipprot-handler 在这里ipprot重载为tcp_protocol(handler 即为tcp_v4_rcv)
我们再看下对应的tcp_v4_rcvtcp_v4_rcv|-tcp_v4_do_rcv|-tcp_rcv_state_process|-tcp_data_queue|- sk-sk_data_ready(sock_def_readable)|-wake_up_interruptible_sync_poll(sk-sleep,...)|-__wake_up|-__wake_up_common|-curr-func/* 这里已经被ep_insert添加为ep_poll_callback,而且设定了排它标识WQ_FLAG_EXCLUSIVE*/|-ep_poll_callback
这样我们就看下最终唤醒epoll_wait的ep_poll_callback函数:static int ep_poll_callback(wait_queue_t *wait, unsigned mode, int sync, void *key)
{// 获取wait对应的epitem struct epitem *epi ep_item_from_wait(wait);// epitem对应的eventpoll结构体struct eventpoll *ep epi-ep;// 获取自旋锁保护ready_list等结构spin_lock_irqsave(ep-lock, flags);// 如果当前epi没有被链入ep的ready list,则链入// 这样就把当前的可用事件加入到epoll的可用列表了if (!ep_is_linked(epi-rdllink))list_add_tail(epi-rdllink, ep-rdllist);// 如果有epoll_wait在等待的话则唤醒这个epoll_wait进程// 对应的ep-wq是在epoll_wait调用的时候通过init_waitqueue_entry(wait, current)而生成的// 其中的current即是对应调用epoll_wait的进程信息task_structif (waitqueue_active(ep-wq))wake_up_locked(ep-wq);
}
上述过程如下图所示:最后wake_up_locked调用__wake_up_common,然后调用了在init_waitqueue_entry注册的default_wake_function,调用路径为:wake_up_locked|-__wake_up_common|-default_wake_function|-try_wake_up (wake up a thread)|-activate_task|-enqueue_task running
将epoll_wait进程推入可运行队列等待内核重新调度进程,然后epoll_wait对应的这个进程重新运行后就从schedule恢复继续下面的ep_send_events(向用户空间拷贝事件并返回)。 wake_up过程如下图所示:可写事件到来可写事件的运行过程和可读事件大同小异: 首先在epoll_ctl_add的时候预先会调用一次对应文件描述符的poll如果返回事件里有可写掩码的时候直接调用wake_up_locked以唤醒对应的epoll_wait进程。 然后在tcp在底层驱动有数据到来的时候可能携带了ack从而可以释放部分已经被对端接收的数据于是触发可写事件这一部分的调用链为:tcp_input.c
tcp_v4_rcv|-tcp_v4_do_rcv|-tcp_rcv_state_process|-tcp_data_snd_check|-tcp_check_space|-tcp_new_space|-sk-sk_write_space/* tcp下即是sk_stream_write_space*/
最后在此函数里面sk_stream_write_space唤醒对应的epoll_wait进程void sk_stream_write_space(struct sock *sk)
{// 即有1/3可写空间的时候才触发可写事件if (sk_stream_wspace(sk) sk_stream_min_wspace(sk) sock) {clear_bit(SOCK_NOSPACE, sock-flags);if (sk-sk_sleep waitqueue_active(sk-sk_sleep))wake_up_interruptible_poll(sk-sk_sleep, POLLOUT |POLLWRNORM | POLLWRBAND)......}
}
关闭描述符(close fd)值得注意的是我们在close对应的文件描述符的时候会自动调用eventpoll_release将对应的file从其关联的epoll_fd中删除kernel关键路径如下:close fd|-filp_close|-fput|-__fput|-eventpoll_release|-ep_remove
所以我们在关闭对应的文件描述符后并不需要通过epoll_ctl_del来删掉对应epoll中相应的描述符。
