epoll的实现原理-阿里云开发者社区

概述

当发数据的时候，协议栈是知道的，但是应用程序是不知道的，epoll的作用就是在协议栈和应用层之间做到对应的连接有没有数据的检测，并且epoll会管理众多的fd。epoll其实就相当于协议栈和应用程序之间的一个组件。应用程序通过调用epoll的三个接口跟epoll交互，而协议栈所做的事情就是当协议栈触发收或者发数据的时候，就将红黑树的节点添加到就绪队列当中，这里面会做一个回调，由协议栈callback到epoll，使得epoll_wait能够拷贝到数据.

实现自定义epoll时，需要实现哪些东西呢？

epitem/epoll_event结构体
epoll_create
epoll_ctl
epoll_wait
epollin/epollout 对应的事件
callback 协议栈有触发收或者发数据的时候（总共有四处：三次握手完成、接收数据、发送数据、接收到fin包），就会触发回调，回调中红黑树的节点添加到就绪队列当中，并条件通知到epoll_wait

epoll数据结构

epoll数据结构的选择

我们发现，当我们添加、删除、修改一个fd对应的事件的时候，都需要去查找，所以它是一个查找频率很高的数据结构，对于查找频率很高的数据结构，一般有如下几个数据结构：

红黑树
哈希
b/b+树

我们分析一下，首先epoll工作的fd的数量是不确定的，有可能一个，也有可能n多个，hash从一开始占的内存就比较大，如果从一开始你就知道fd的数量很多的时候，你就可以选择hash，B树/B+树查找的性能没有红黑树高，综上，红黑树是最合适的

epoll数据结构定义

epoll 主要由两个结构体：eventpoll 与 epitem。epitem 是每一个 IO 所对应的的事件。比如 epoll_ctl EPOLL_CTL_ADD 操作的时候，就需要创建一个 epitem。eventpoll 是每一个 epoll 所对应的的。比如 epoll_create 就是创建一个 eventpoll。

epitem定义

struct epitem {
  RB_ENTRY(epitem) rbn;   // 红黑树节点
  LIST_ENTRY(epitem) rdlink;  // 用来做就绪的队列
  int rdy;          // 是否在就绪队列中，1存在，0不存在
  int sockfd;         // key
  struct epoll_event event;   // value
};

eventpoll 定义

struct eventpoll {
  int fd;
  ep_rb_tree rbr;       // 红黑树的根节点
  int rbcnt;          // 添加到红黑树中节点的数量
  LIST_HEAD( ,epitem) rdlist; // 队列的节点
  int rdnum;          // 就绪数量
  int waiting;
  pthread_mutex_t mtx;    // rbtree update
  pthread_spinlock_t lock;  // rdlist update
  pthread_cond_t cond;    // block for event  使用条件变量主要是因为epoll_wait这个函数里面是带阻塞的
  pthread_mutex_t cdmtx;    // mutex for cond
};

具体如下图所示：

List 用来存储准备就绪的 IO。对于数据结构主要讨论两方面：insert 与 remove。同样如此，对于 list 我们也讨论 insert 与 remove。何时将数据插入到 list 中呢？当内核 IO 准备就绪的时候，则会执行 epoll_event_callback 的回调函数，将 epitem 添加到 list 中。那何时删除 list 中的数据呢？当 epoll_wait 激活重新运行的时候，将 list 的 epitem 逐一 copy 到 events 参数中。 Rbtree 用来存储所有 io 的数据，方便快速通 io_fd 查找。也从 insert 与 remove 来讨论。对于rbtree 何时添加：当 App 执行 epoll_ctl EPOLL_CTL_ADD 操作，将 epitem 添加到 rbtree 中。何时删除呢？当 App 执行 epoll_ctl EPOLL_CTL_DEL 操作，将 epitem 从rbtree中移除。

epoll三大应用层接口实现

nepoll_create

// 创建eventpoll结构体，并对其进行初始化
int nepoll_create(int size) {
  if (size <= 0) return -1;
  int epfd = get_fd_frombitmap();
  struct eventpoll *ep = (struct eventpoll*)rte_calloc("eventpoll", 1, sizeof(struct eventpoll), 0);
  if (!ep) {
    set_fd_frombitmap(epfd);
    return -1;
  }
  ep->rbcnt = 0;
  RB_INIT(&ep->rbr);
  LIST_INIT(&ep->rdlist);
  if (pthread_mutex_init(&ep->mtx, NULL)) {
    rte_free(ep);
    set_fd_frombitmap(epfd);
    return -2;
  }
  if (pthread_mutex_init(&ep->cdmtx, NULL)) {
    pthread_mutex_destroy(&ep->mtx);
    rte_free(ep);       // dpdk中提供了一系列内存操作，不允许私自分配和释放
    set_fd_frombitmap(epfd);
    return -2;
  }
  if (pthread_cond_init(&ep->cond, NULL)) {
    pthread_mutex_destroy(&ep->cdmtx);
    pthread_mutex_destroy(&ep->mtx);
    rte_free(ep);
    set_fd_frombitmap(epfd);
    return -2;
  }
  if (pthread_spin_init(&ep->lock, PTHREAD_PROCESS_SHARED)) {
    pthread_cond_destroy(&ep->cond);
    pthread_mutex_destroy(&ep->cdmtx);
    pthread_mutex_destroy(&ep->mtx);
    rte_free(ep);
    set_fd_frombitmap(epfd);
    return -2;
  }
  return epfd;
}

nepoll_ctl

// 这个函数主要有三个操作：ADD/DEL/MOD
int nepoll_ctl(int epfd, int op, int fd, struct epoll_event *event) {
  struct eventpoll *ep = (struct eventpoll *)get_hostinfo_fromfd(epfd);
  if (!ep || (!event && op != EPOLL_CTL_DEL)) {
    errno = -EINVAL;
    return -1;
  }
  if (op == EPOLL_CTL_ADD) {
    pthread_mutex_lock(&ep->mtx);
    struct epitem tmp;
    tmp.sockfd = fd;
    struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);
    if (epi) {
      // 走进这里说明红黑树中已经存在这个节点了
      pthread_mutex_unlock(&ep->mtx);
      return -1;
    }
    // 不存在就创建一个节点
    epi = (struct epitem*)rte_calloc("epitem", 1, sizeof(struct epitem), 0);
    if (!epi) {
      pthread_mutex_unlock(&ep->mtx);
      errno = -ENOMEM;
      return -1;
    }
    epi->sockfd = fd;
    memcpy(&epi->event, event, sizeof(struct epoll_event));
    epi = RB_INSERT(_epoll_rb_socket, &ep->rbr, epi);
    //assert(epi == NULL);
    ep->rbcnt ++;
    pthread_mutex_unlock(&ep->mtx);
  } else if (op == EPOLL_CTL_DEL) {
    pthread_mutex_lock(&ep->mtx);
    struct epitem tmp;
    tmp.sockfd = fd;
    struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);
    if (!epi) {
      pthread_mutex_unlock(&ep->mtx);
      return -1;
    }
    epi = RB_REMOVE(_epoll_rb_socket, &ep->rbr, epi);
    if (!epi) {
      pthread_mutex_unlock(&ep->mtx);
      return -1;
    }
    ep->rbcnt --;
    rte_free(epi);
    pthread_mutex_unlock(&ep->mtx);
  } else if (op == EPOLL_CTL_MOD) {
    struct epitem tmp;
    tmp.sockfd = fd;
    struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);
    if (epi) {
      epi->event.events = event->events;
      epi->event.events |= EPOLLERR | EPOLLHUP;
    } else {
      errno = -ENOENT;
      return -1;
    }
  } else {
    //assert(0);
  }
  return 0;
}

nepoll_wait

// 作用就是从就绪队列里面把节点拷贝出来，跟红黑树没啥关系
// epfd指具体对应哪个epfd
// events:就绪队列里面的数据会最终拷贝到events中
// maxevents:指events中最多装多少个
// timeout:指如果就绪队列为空，我们等待时间有多长
  // -1，表示如果为空，就一直等
  // 0，立马返回，不管就绪队列中有没有数据
  // >0的数，比如10，那么就等待10ms返回
// 这个timeout的10ms我们是怎么实现的？
  // 主要用到pthread_cond_wait和pthread_cond_timedwait两个函数去解决
int nepoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout) {
  struct eventpoll *ep = (struct eventpoll *)get_hostinfo_fromfd(epfd);
  if (!ep || !events || maxevents <= 0) {
    errno = -EINVAL;
    return -1;
  }
  if (pthread_mutex_lock(&ep->cdmtx)) {
    if (errno == EDEADLK) {
    }
    //assert(0);
  }
  while (ep->rdnum == 0 && timeout != 0) {
    ep->waiting = 1;
    if (timeout > 0) {
      struct timespec deadline;
      clock_gettime(CLOCK_REALTIME, &deadline);
      if (timeout >= 1000) {
        int sec;
        sec = timeout / 1000;
        deadline.tv_sec += sec;
        timeout -= sec * 1000;
      }
      deadline.tv_nsec += timeout * 1000000;
      if (deadline.tv_nsec >= 1000000000) {
        deadline.tv_sec++;
        deadline.tv_nsec -= 1000000000;
      }
      // timeout>0，就用pthread_cond_timedwait处理
      int ret = pthread_cond_timedwait(&ep->cond, &ep->cdmtx, &deadline);
      if (ret && ret != ETIMEDOUT) {
        pthread_mutex_unlock(&ep->cdmtx);
        return -1;
      }
      timeout = 0;
    } else if (timeout < 0) {
      // timeout < 0就使用pthread_cond_wait函数进行条件等待
      int ret = pthread_cond_wait(&ep->cond, &ep->cdmtx);
      if (ret) {
        pthread_mutex_unlock(&ep->cdmtx);
        return -1;
      }
    }
    ep->waiting = 0; 
  }
  pthread_mutex_unlock(&ep->cdmtx);
  pthread_spin_lock(&ep->lock);
  int cnt = 0;
  int num = (ep->rdnum > maxevents ? maxevents : ep->rdnum);
  int i = 0;
  while (num != 0 && !LIST_EMPTY(&ep->rdlist)) { //EPOLLET
    struct epitem *epi = LIST_FIRST(&ep->rdlist);
    LIST_REMOVE(epi, rdlink);
    epi->rdy = 0;
    memcpy(&events[i++], &epi->event, sizeof(struct epoll_event));
    num --;
    cnt ++;
    ep->rdnum --;
  }
  pthread_spin_unlock(&ep->lock);
  return cnt;
}

epoll与协议栈

epoll 的回调函数何时执行，此部分需要与 tcp 的协议栈一起来阐述。tcp 协议栈的时序图如下图所示，epoll 从协议栈回调的部分从下图的编号 1,2,3,4。具体 tcp 协议栈的实现，后续从另外的文章中表述出来。下面分别对四个步骤详细描述 :

编号 1：是 tcp 三次握手，对端反馈 ack 后，socket 进入 rcvd 状态。需要将监听 socket 的 event 置为 EPOLLIN，此时标识可以进入到 accept 读取 socket 数据。
编号 2：在 established 状态，收到数据以后，需要将 socket 的 event 置为 EPOLLIN 状态。
编号 3：在 established 状态，收到 fin 时，此时 socket 进入到 close_wait。需要 socket 的 event 置为 EPOLLIN。读取断开信息。
编号 4：检测 socket 的 send 状态，如果对端 cwnd>0 是可以，发送的数据。故需要将 socket 置为 EPOLLOUT。所以在此四处添加 EPOLL 的回调函数，即可使得 epoll 正常接收到 io 事件。

nepoll_event_callback

// 这个函数的作用就是找到红黑树中的fd对应的节点，并且把它假如到就绪队列中
// 这个函数是从协议栈回调到epoll的模块中间
// nepoll_event_callback和nepoll_wait其实是一个生产者和消费者的过程
int nepoll_event_callback(struct eventpoll *ep, int sockid, uint32_t event) {
  struct epitem tmp;
  tmp.sockfd = sockid;
  struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);
  if (!epi) {
    return -1;
  }
  if (epi->rdy) { // 如果就绪队列中已经存在了
    epi->event.events |= event;
    return 1;
  } 
  pthread_spin_lock(&ep->lock);
  epi->rdy = 1;
  LIST_INSERT_HEAD(&ep->rdlist, epi, rdlink);
  ep->rdnum ++;
  pthread_spin_unlock(&ep->lock);
  pthread_mutex_lock(&ep->cdmtx);
  pthread_cond_signal(&ep->cond);   // 条件通知的动作
  pthread_mutex_unlock(&ep->cdmtx);
  return 0;
}

注意：

nepoll_event_callback和nepoll_wait其实是一个生产者和消费者的过程。

epoll锁机制

epoll 从以下几个方面是需要加锁保护的。list 的操作，rbtree 的操作，epoll_wait 的等待。 list 使用最小粒度的锁 spinlock，便于在 SMP 下添加操作的时候，能够快速操作 list。

list添加

346 行：获取 spinlock。

347 行：epitem 的 rdy 置为 1，代表 epitem 已经在就绪队列中，后续再触发相同事件就只需更改 event。

348 行：添加到 list 中。

349 行：将 eventpoll 的 rdnum 域加 1。

350 行：释放 spinlock

list删除

301 行：获取 spinlock

304 行：判读 rdnum 与 maxevents 的大小，避免 event 溢出。

307 行：循环遍历 list，判断添加 list 不能为空

309 行：获取 list 首个结点 310 行：移除 list 首个结点。

311 行：将 epitem 的 rdy 域置为 0，标识 epitem 不再就绪队列中。

313 行：copy epitem 的 event 到用户空间的 events。

316 行：copy 数量加 1

317 行：eventpoll 中 rdnum 减一。

避免 SMP 体系下，多核竞争。此处采用自旋锁，不适合采用睡眠锁。

rbtree添加

149 行：获取互斥锁。

153 行：查找 sockid 的 epitem 是否存在。存在则不能添加，不存在则可以添加。

160 行：分配 epitem。

167 行：sockid 赋值

168 行：将设置的 event 添加到 epitem 的 event 域。

170 行：将 epitem 添加到 rbrtree 中。 173 行：释放互斥锁。

rbtree删除

177 行：获取互斥锁。

181 行：删除 sockid 的结点，如果不存在，则 rbtree 返回-1。

188 行：释放 epitem

190 行：释放互斥锁。

epoll_wait的等待

具体参见nepoll_event_callback（生产者）与nepoll_wait（消费者）

LT与ET

1、et 与lt的区别

et 边沿触发：当调用recv函数如果没有把协议栈中的recvbuffr中的数据全部读完，就不会继续触发，当客户端再发数据的时候，才会再次被触发。

lt水平触发：当调用recv函数如果没有把协议栈中的recvbuffr中的数据全部读完，就回一直触发，直到全部读完为止。

2、为什么会有水平触发和边沿触发？

水平触发和边沿触发不是一开始就故意设计出来的，其理念来自于嵌入式的电平的高低变化

3、如何实现水平触发和边沿触发？

et从协议栈中检测到recvbuffer中接收数据就调用回调，水平触发检测recvbuffer有数据就调回调，水平触发和边沿触发代码实现核心是内核通知epoll时执行回调函数的次数的区别。

epoll的实现原理

概述

epoll数据结构

epoll数据结构的选择

epoll数据结构定义

epoll三大应用层接口实现

nepoll_create

nepoll_ctl

nepoll_wait

epoll与协议栈

nepoll_event_callback

epoll锁机制

LT与ET

热门文章

最新文章

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

epoll的实现原理

概述

epoll数据结构

epoll数据结构的选择

epoll数据结构定义

epoll三大应用层接口实现

nepoll_create

nepoll_ctl

nepoll_wait

epoll与协议栈

nepoll_event_callback

epoll锁机制

LT与ET

热门文章

最新文章

相关电子书