1.概述

这篇文章分析一下linux中epoll的实现原理，主要为了增强自己对网络调用的理解。业界使用epoll的框架比较多，随便就能列出来很多，比如jdk的nio在linux下的实现，以及netty、redis等涉及到长链接网络请求的地方，我们都可以直接使用epoll。文末会从redis源码简单看看如何使用epoll做IO多路复用实现高并发。

2.具体实现

参考官方文档描述：

The central concept of the epoll API is the epoll instance, an inkernel data structure which, from a user-space perspective, can be considered as a container for two lists

所以 其实就是epoll就是内核的一个数据结构。从用户空间的角度，其实就是两个链表。所以基本上就是维护两个链表就可以了。 理解完这段话，我们也就能理解Epoll提供了三个方法了：

  • create是初始化这个内核的数据结构。返回一个fd。众所周知，unix万物皆文件。所以，这里创建返回的是一个文件fd。每次操作我们只需要传入fd，内核便能拿到epoll对应的数据结构。
  • epoll_ctl就是对其中一个链表的操作。这个链表存放用户感兴趣的io事件。当然在注册事件的时候，会有一些其他的操作。后面详细解释
  • epoll_wait，则是返回就绪事件（感兴趣的事件）。然后让应用层去处理。
int epoll_create(int size)；
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)；
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);struct epoll_event {__uint32_t events;epoll_data_t data;
};

epoll_create方法

SYSCALL_DEFINE1(epoll_create1, int, flags)
{int error, fd;struct eventpoll *ep = NULL;struct file *file;// 创建内部数据结构eventpoll error = ep_alloc(&ep);//查询未使用的fdfd = get_unused_fd_flags(O_RDWR | (flags & O_CLOEXEC));file = anon_inode_getfile("[eventpoll]", &eventpoll_fops, ep,O_RDWR | (flags & O_CLOEXEC));ep->file = file;fd_install(fd, file);  //建立fd和file的关联关系return fd;
out_free_fd:put_unused_fd(fd);
out_free_ep:ep_free(ep);return error;
}

简单的说一下这个方法。首先这个方法返回一个文件描述符。我们使用这个文件描述符就可以找到对应的结构体。也就是一块内存区域。epoll的所有数据都会保存在这里。 这块内存区域用eventpoll结构体表示。 所以这个方法逻辑如下：

1.创建eventpoll结构体。初始化结构体相应的数据

2.查询一个未使用的fd，然后为epoll创建一个文件。将file的->private_data指向ep。创建文件的过程就不多说

3.将ep->file指向file。其实就是绑定一下。

4.将fd和file关联。这样我们就能通过fd找到对应的file。并找到ep对应的结构体（内存区域） 这里再说一下 file的private_data其实在设备驱动程序中非常重要，它可以指向一块自定义的数据结构。这也就是保证了一个设备驱动程序可以适配多个设备的原因。因为不同的设备可能存在不同的属性。epoll这里用private_data指向自己的数据结构是完全没有问题的。

eventpoll结构体内容如下。后面遇到详细说。

struct eventpoll {spinlock_t lock;struct mutex mtx;wait_queue_head_t wq; //sys_epoll_wait（）使用的等待队列wait_queue_head_t poll_wait; //file->poll()使用的等待队列struct list_head rdllist; //所有准备就绪的文件描述符列表struct rb_root rbr; //用于储存已监控fd的红黑树根节点struct epitem *ovflist; //用于监听文件的结构。如果rdllist被锁定，临时事件会被连接到这里struct wakeup_source *ws; // 当ep_scan_ready_list运行时使用wakeup_sourcestruct user_struct *user; //创建eventpoll描述符的用户struct file *file;int visited;           //用于优化循环检测检查struct list_head visited_list_link;
};

epoll_ctl方法

该方法主要就是对监听事件进行增删改。

SYSCALL_DEFINE4(epoll_ctl, int, epfd, int, op, int, fd,struct epoll_event __user *, event)
{int error;int full_check = 0;struct fd f, tf;struct eventpoll *ep;    struct epitem *epi;     struct epoll_event epds; struct eventpoll *tep = NULL;error = -EFAULT;//如果不是删除操作，将用户空间的epoll_event 拷贝到内核if (ep_op_has_event(op) &&copy_from_user(&epds, event, sizeof(struct epoll_event)))f = fdget(epfd); //epfd对应的文件tf = fdget(fd); //fd对应的文件....ep = f.file->private_data; // 取出epoll_create过程创建的ep...epi = ep_find(ep, tf.file, fd); //ep红黑树中查看该fdswitch (op) {case EPOLL_CTL_ADD:if (!epi) {epds.events |= POLLERR | POLLHUP;error = ep_insert(ep, &epds, tf.file, fd, full_check); }if (full_check)clear_tfile_check_list();break;case EPOLL_CTL_DEL:if (epi)error = ep_remove(ep, epi); break;case EPOLL_CTL_MOD:if (epi) {epds.events |= POLLERR | POLLHUP;error = ep_modify(ep, epi, &epds); }break;}mutex_unlock(&ep->mtx);fdput(tf);fdput(f);...return error;
}

分享更多关于C/C++ Linux后端开发网络底层原理知识学习提升 点击 学习资料 获取，完善技术栈，内容知识点包括Linux，Nginx，ZeroMQ，MySQL，Redis，线程池，MongoDB，ZK，Linux内核，CDN，P2P，epoll，Docker，TCP/IP，协程，DPDK等等。

上面省略了一些判断的代码。主要核心就是根据不同的事件类型执行不同的函数。

epoll调用ep_find从红黑树拿到对应的epi。如果已经存在，则不需要进行add。如果不存在，也不可进行remove和modify操作。整个过程会加锁。因为是红黑树，所以查找以及插入性能都是logn级别。所以对于高并发场景，也是能实现快速注册监听的。下面我们分别看一下这三个操作的逻辑。

ep_insert操作

顾名思义，就是加入监听事件。

static int ep_insert(struct eventpoll *ep, struct epoll_event *event,struct file *tfile, int fd, int full_check)
{int error, revents, pwake = 0;unsigned long flags;long user_watches;struct epitem *epi;struct ep_pqueue epq; //[小节2.4.5]user_watches = atomic_long_read(&ep->user->epoll_watches);if (unlikely(user_watches >= max_user_watches))return -ENOSPC;if (!(epi = kmem_cache_alloc(epi_cache, GFP_KERNEL)))return -ENOMEM;//构造并填充epi结构体INIT_LIST_HEAD(&epi->rdllink);INIT_LIST_HEAD(&epi->fllink);INIT_LIST_HEAD(&epi->pwqlist);epi->ep = ep;ep_set_ffd(&epi->ffd, tfile, fd); // 将tfile和fd都赋值给ffdepi->event = *event;epi->nwait = 0;epi->next = EP_UNACTIVE_PTR;if (epi->event.events & EPOLLWAKEUP) {error = ep_create_wakeup_source(epi);} else {RCU_INIT_POINTER(epi->ws, NULL);}epq.epi = epi;//设置轮询回调函数init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);//执行poll方法revents = ep_item_poll(epi, &epq.pt);spin_lock(&tfile->f_lock);list_add_tail_rcu(&epi->fllink, &tfile->f_ep_links);spin_unlock(&tfile->f_lock);ep_rbtree_insert(ep, epi); //将将当前epi添加到RB树spin_lock_irqsave(&ep->lock, flags);//事件就绪 并且 epi的就绪队列有数据if ((revents & event->events) && !ep_is_linked(&epi->rdllink)) {list_add_tail(&epi->rdllink, &ep->rdllist);ep_pm_stay_awake(epi);//唤醒正在等待文件就绪，即调用epoll_wait的进程if (waitqueue_active(&ep->wq))wake_up_locked(&ep->wq);if (waitqueue_active(&ep->poll_wait))pwake++;}spin_unlock_irqrestore(&ep->lock, flags);atomic_long_inc(&ep->user->epoll_watches);if (pwake)ep_poll_safewake(&ep->poll_wait); //唤醒等待eventpoll文件就绪的进程return 0;
...
}

1.首先会初始化一个epi，对于目标文件的监听都是需要通过epi去维护。一个文件的监听对应一个epi。并且保存在ep的红黑树中。

struct epitem {union {struct rb_node rbn; //RB树节点将此结构链接到eventpoll RB树struct rcu_head rcu; //用于释放结构体epitem};struct list_head rdllink; //时间的就绪队列，主要就是链接到eventpoll的rdlliststruct epitem *next; //配合eventpoll中的ovflist一起使用来保持单向链的条目struct epoll_filefd ffd; //该结构 监听的文件描述符信息，每一个socket fd都会对应一个epitem 。就是通过这个结构关联int nwait; //附加到poll轮询中的活跃等待队列数struct list_head pwqlist; //用于保存被监听文件的等待队列struct eventpoll *ep;  //epi所属的epstruct list_head fllink; //主要是为了实现一个文件被多个epoll监听。将该结构链接到文件的f_ep_link。struct wakeup_source __rcu *ws; //设置EPOLLWAKEUP时使用的wakeup_sourcestruct epoll_event event; //监控的事件和文件描述符
};

2.初始化结束后，会将文件的fd以及对应文件指针绑定到epi的ffd中。主要作用就是将fd和改epi绑定起来。

struct epoll_filefd {struct file *file;int fd;
} __packed;

3.为epq的pt（其实就是一个poll_table）注册对应的函数ep_ptable_queue_proc。

struct ep_pqueue {poll_table pt;struct epitem *epi;
};typedef struct poll_table_struct {poll_queue_proc _qproc;unsigned long _key;
} poll_table;

这里epq是一个结构体，里面绑定了epi以及poll_table。poll_table主要注册了ep_ptable_queue_proc函数。_key用于记录事件。 所以epq就保存了epi，以及对应的ep_ptable_queue_proc。后续执行回调函数的时候，我们可以通过poll_table的地址拿到对应的epq，最终拿到对应的epi，这也就是定义这个结构的目的。

4.调用ep_item_poll方法。这个方法我简单说一下。他会调用文件系统的poll方法。

static inline unsigned int ep_item_poll(struct epitem *epi, poll_table *pt)
{pt->_key = epi->event.events;、return epi->ffd.file->f_op->poll(epi->ffd.file, pt) & epi->event.events;
}

不同的驱动程序，都会有自己的poll方法，如果是TCP套接字，这个poll方法就是tcp_poll。在TCP中，会周期性的调用这个方法，调用频率取决于协议栈中断频率的设置。一旦有事件到达后，对应的tcp_poll方法被调用，tcp_poll方法会回调用sock_poll_wait()，该方法会调用这里注册的ep_ptable_queue_proc方法。epoll其实就是通过此机制实现将自己的回调函数加入到文件的waitqueue中的。这也是ep_ptable_queue_proc的目的。

5.调用ep_item_poll后会返回revents。也就是该fd触发的事件。如果有我们感兴趣的事件，会将其插入到ep的rdllist中。如果有进程正在等待文件的就绪状态，也就是调用epoll_wait睡眠的进程。那么会唤醒等待进程。

 if ((revents & event->events) && !ep_is_linked(&epi->rdllink)) {list_add_tail(&epi->rdllink, &ep->rdllist);ep_pm_stay_awake(epi);//唤醒正在等待文件就绪，即调用epoll_wait的进程if (waitqueue_active(&ep->wq))wake_up_locked(&ep->wq);if (waitqueue_active(&ep->poll_wait))pwake++;}

ep_ptable_queue_proc方法

整个过程其实就是通过文件的poll方法绑定ep_ptable_queue_proc函数 。当该文件描述符对应的文件有事件到达后，回调用这个函数

Ps：其中file就是对应文件的结构体。当然 多个fd可以指向通过file结构。多个file可以同时指向同一个innode节点。在linux中，一个文件的内容是通过innode去定义描述的。file只是我们对文件操作的时候创建出来的。这个概念大家需要清楚。

static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead,poll_table *pt)
{struct epitem *epi = ep_item_from_epqueue(pt);struct eppoll_entry *pwq;  if (epi->nwait >= 0 && (pwq = kmem_cache_alloc(pwq_cache, GFP_KERNEL))) {//初始化回调方法init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);pwq->whead = whead;pwq->base = epi;//将ep_poll_callback放入等待队列wheadadd_wait_queue(whead, &pwq->wait);//将llink 放入epi->pwqlist的尾部list_add_tail(&pwq->llink, &epi->pwqlist);epi->nwait++;} else {epi->nwait = -1; //标记错误发生}
}static inline void
init_waitqueue_func_entry(wait_queue_t *q, wait_queue_func_t func)
{q->flags    = 0;q->private    = NULL;q->func        = func;
}

ep_ptable_queue_proc有三个参数。 file就是监听文件的指针，whead为该fd对应的设备等待队列。pt就是我们当时调用文件的poll传入的东西。

在ep_ptable_queue_proc，引入了eppoll_entry。也就是pwq。pwq主要完成epi和epi事件发生时callback函数的关联。

从上面代码可以看出。首先根据pt拿到对应的epi。然后通过pwq将三者关联。

最后通过add_wait_queue方法，将eppoll_entry挂在到fd的设备等待队列上。也就是注册epoll的回调函数。

所以这个方法的主要目标就是将eppoll_entry挂在到fd的设备等待队列上。当设备有硬件数据到达时，硬件中断处理函数会唤醒该队列上等待的进程时，会调用唤醒函数ep_poll_callback。

分享更多关于C/C++ Linux后端开发网络底层原理知识学习提升 点击 学习资料 获取，完善技术栈，内容知识点包括Linux，Nginx，ZeroMQ，MySQL，Redis，线程池，MongoDB，ZK，Linux内核，CDN，P2P，epoll，Docker，TCP/IP，协程，DPDK等等。

ep_poll_callback方法

这个函数主要功能就是，当被监听文件的事件就绪，将文件对应的epi加入到就绪队列。当应用层调用epoll_wait()的时候，内核会将就绪队列的事件拷贝到用户空间。报告给应用。

static int ep_poll_callback(wait_queue_t *wait, unsigned mode, int sync, void *key)
{int pwake = 0;unsigned long flags;struct epitem *epi = ep_item_from_wait(wait);struct eventpoll *ep = epi->ep;spin_lock_irqsave(&ep->lock, flags);if (unlikely(ep->ovflist != EP_UNACTIVE_PTR)) {if (epi->next == EP_UNACTIVE_PTR) {epi->next = ep->ovflist;ep->ovflist = epi;if (epi->ws) {__pm_stay_awake(ep->ws);}}goto out_unlock;}//如果此文件已在就绪列表中，很快就会退出if (!ep_is_linked(&epi->rdllink)) {//将epi就绪事件 插入到ep就绪队列list_add_tail(&epi->rdllink, &ep->rdllist);ep_pm_stay_awake_rcu(epi);}// 如果活跃，唤醒eventpoll等待队列和 ->poll()等待队列if (waitqueue_active(&ep->wq))wake_up_locked(&ep->wq);  //当队列不为空，则唤醒进程if (waitqueue_active(&ep->poll_wait))pwake++;
out_unlock:spin_unlock_irqrestore(&ep->lock, flags);if (pwake)ep_poll_safewake(&ep->poll_wait);if ((unsigned long)key & POLLFREE) {list_del_init(&wait->task_list); //删除相应的waitsmp_store_release(&ep_pwq_from_wait(wait)->whead, NULL);}return 1;
}//判断等待队列是否为空
static inline int waitqueue_active(wait_queue_head_t *q)
{return !list_empty(&q->task_list);
}

3.epoll实现总结

透过现象看本质，其实epoll的灵魂就是ep_item_poll和ep_poll_callback。

epoll依赖虚拟文件系统的ep_item_poll。将ep_poll_callback注册到对应文件的waitqueue中。当对应文件有数据到来。注册的函数就会被调用。epoll的回调会将对应文件的epi加入到就绪队列。

当用户调用epoll_wait()，epoll会加锁，将队列数据传递到用户空间，这个时间到的事件会被挂到ovflist中。

4.Redis使用Epoll

具体实现在ae_epoll.c中

typedef struct aeApiState {// epoll_event 实例描述符int epfd;// 事件槽struct epoll_event *events;} aeApiState;

aeApiCreate方法

redis在初始化server的时候会调用aeCreateEventLoop方法。aeCreateEventLoop回调用aeApiCreate去创建epoll实例。

static int aeApiCreate(aeEventLoop *eventLoop) {aeApiState *state = zmalloc(sizeof(aeApiState));if (!state) return -1;state->events = zmalloc(sizeof(struct kevent)*eventLoop->setsize);if (!state->events) {zfree(state);return -1;}state->kqfd = kqueue();if (state->kqfd == -1) {zfree(state->events);zfree(state);return -1;}eventLoop->apidata = state;return 0;
}

aeApiAddEvent方法

这个方法是关联事件到epoll，所以会调用epoll的ctl方法

static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {aeApiState *state = eventLoop->apidata;struct epoll_event ee;/* If the fd was already monitored for some event, we need a MOD* operation. Otherwise we need an ADD operation. ** 如果 fd 没有关联任何事件，那么这是一个 ADD 操作。* 如果已经关联了某个/某些事件，那么这是一个 MOD 操作。*/int op = eventLoop->events[fd].mask == AE_NONE ?EPOLL_CTL_ADD : EPOLL_CTL_MOD;// 注册事件到 epollee.events = 0;mask |= eventLoop->events[fd].mask; /* Merge old events */if (mask & AE_READABLE) ee.events |= EPOLLIN;if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;ee.data.u64 = 0; /* avoid valgrind warning */ee.data.fd = fd;if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;return 0;
}

这个方法在redis服务创建一个新的客户端的时候会调用。会注册这个客户端的读事件。

当redis需要给客户端写数据的时候会调用prepareClientToWrite方法。这个方法主要是注册对应fd的写事件。

如果注册失败，redis就不会将数据写入缓冲。

如果对应套件字可写，那么redis的事件循环就会将缓冲区新数据写入socket。

aeMain方法

Redis事件处理器的主循环。

void aeMain(aeEventLoop *eventLoop) {eventLoop->stop = 0;while (!eventLoop->stop) {// 如果有需要在事件处理前执行的函数，那么运行它if (eventLoop->beforesleep != NULL)eventLoop->beforesleep(eventLoop);// 开始处理事件aeProcessEvents(eventLoop, AE_ALL_EVENTS);}
}

这个方法最终会调用epoll_wait()获取对应事件并执行。

epoll源码分析以及在Redis中的实现相关推荐

  1. Hbase源码分析：Hbase UI中Requests Per Second的具体含义

    Hbase源码分析:Hbase UI中Requests Per Second的具体含义 让运维加监控,被问到Requests Per Second(见下图)的具体含义是什么?我一时竟回答不上来,虽然大 ...

  2. djangorestframework源码分析1：generics中的view执行流程

    djangorestframework源码分析 本文环境python3.5.2,djangorestframework (3.5.1)系列 djangorestframework源码分析-generi ...

  3. 【转】ABP源码分析二：ABP中配置的注册和初始化

    一般来说,ASP.NET Web应用程序的第一个执行的方法是Global.asax下定义的Start方法.执行这个方法前HttpApplication 实例必须存在,也就是说其构造函数必然已完成了执行 ...

  4. 【ReentrantLock源码分析】1.xdb中的使用 2.获取和阻塞(阻塞前的一些死心不改)的源码

    1.xdb中的使用例子 在xdb中,我们大概执行业务时的流程简化如下: package org.example.testReentrantLock;import java.util.concurren ...

  5. 云客Drupal源码分析之前端js中的翻译

    从本主题开始<云客Drupal源码分析>系列将连续发布和前端js相关的内容,如果您对JavaScript还不熟悉或者需要来一次系统性的整理回顾,在此云客为您准备了以下资料: <PHP ...

  6. Seata 源码分析 - tm、rm 中 xid 传递过程

    一.Seata 前面文章讲解了对 Seata 的 AT 和 TCC 模式的使用,本篇文章为大家讲解下 Seata 中 TM.RM 中 xid 传递过程,如果不了解 Seata 中的 xid,可以理解为 ...

  7. 硬核HashMap源码分析，HashMap文章中的圣经

    一 前言 本篇是继硬核ArrayList源码分析,答应我每天看一遍好么之后进行的第二篇源码分析:学完本篇你将对hashMap的结构和方法有个全面的了解:面试自己有多强,超人都不知道:比如HashMap ...

  8. redis源码分析 ppt_【Redis】redis各类型数据结构和底层实现源码分析

    一.简介和应用 Redis是一个由ANSI C语言编写,性能优秀.支持网络.可持久化的K-K内存数据库,并提供多种语言的API.它常用的类型主要是 String.List.Hash.Set.ZSet ...

  9. redis源码分析(beta版本)-redis实现的概述逻辑

    redis概述 redis作为一个高性能的kv数据库,基于c语言实现,数据主要基于内存存放,所以查询效率较高,并且支持较多的数据类型有字符串.字典.列表与集合等数据结构.另外redis叶支持持久化的日 ...

最新文章

  1. Android 第三方图表类 MPChart 的使用
  2. R语言CRAN不包含的安装包下载并手动使用RStudio进行安装实战：以下载VRPM包及绘制彩色列线图为例
  3. AndroidAnnotations框架简单使用方法
  4. 媒体洞察 | 让企业自由发展的云时代
  5. 获取版本号_Windows 10 2004迎来KB4577063更新 版本号升至19041.546
  6. linux下 最常用基本命令
  7. mysql 日志_MySQL日志系统
  8. Go语言11-日志系统客户端相关组件
  9. linux openssl 证书,Linux上的openSSL证书验证
  10. socket的原理和实验
  11. 大三寒假--算法复习
  12. SQL Sverver 2000 管理安全性
  13. 【转】Android 平台下使用 i2c-tools
  14. java arraydeque poll,Java ArrayDeque pollLast()方法
  15. Quartz-scheduler 定时器概述、核心 API 与 快速入门
  16. 我的NVIDIA开发者之旅——Caffe教程（2）[Jetson TK1]Caffe工具环境（Linux）搭建实例
  17. dcpb连接mysql_MySQL Cluster 7.3.7+CentOS7集群配置入门 MySQL双管理节点配置入门
  18. OPNsense用户手册-反病毒引擎
  19. matlab非参数检验（符号检验，秩和）
  20. vue 解决跨域问题404问题

热门文章

  1. 北京交通广播对话雷建平：揭秘中概股上市背后故事
  2. 外汇经典交易策略，最基本的趋势跟踪策略
  3. 安卓与服务器采用josn传递数据
  4. Vue-Cli4.x项目通过electron打包桌面应用程序
  5. MA5626 MA5620改交换机 二层端口互通命令
  6. 【P15】全分立 FET/MOS 甲类耳机放大器
  7. LCD1602学习笔记
  8. VB做论坛自动发贴软件（1）
  9. 【MATLAB项目实战】基于Morlet小波变换的滚动轴承故障特征提取研究
  10. node.js 是什么