Kanon(六): Epoller(epoll wrapper)

谈谈Epoller的实现细节。

UpdateChannel

根据man epoll可以知道对内核的事件表的操作是通过epoll_ctl()和其参数指定的：

  • EPOLL_ADD
  • EPOLL_MOD
  • EPOLL_DEL

因此封装成：

1
2
3
void UpdateChannel(Channel* ch);
// Forward to UpdateEpollEvent()
void UpdateEpollEvent(int op, Channel* ch);

对于epoll_ctl()，它只需要fd，因此Channel本身是不兼容的，它需要额外存储，为了避免多余的空间开销，epoll它并不是返回fd和触发事件，而是返回epoll_data_t数据和触发事件，具体来说，是这样的：

1
2
3
4
5
6
7
8
9
10
11
12
13
typedef union epoll_data {
  void *ptr;
  int fd;
  uint32_t u32;
  uint64_t u64;
} epoll_data_t;

struct epoll_event {
  uint32_t events; // Epoll events
  epoll_data_t data; // user data variable
  // int fd;
};

可以看出，struct epoll_data有个void*成员，所以可以存储任何类型的数据，兼容用户自定义数据结构，到时候只需要转换回来就行了。

相比，如果struct epoll_event的第二个数据成员是fd，那就十分不灵活，不利于扩展。

这样对于UpdateEpollEvent也就没有疑问了：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
void Epoller::UpdateEpollEvent(int op, Channel* ch) noexcept {
  int fd = ch->GetFd();
  struct epoll_event ev;

  ev.events = ch->GetEvents(); // interested
  if (!ch->IsNoneEvent() && is_et_mode_) {
    ev.evnets |= EPOLLET;
  }

  ev.data.ptr = static_cast<void*>(ch);

  if (::epoll_ctl(epoll_fd_, op, fd, &ev)) {
    // error handling
  }
}

接下来实现UpdateChannel()又遇到一个问题：我怎么知道Channel是否添加过？故而Channel有一个成员记录它的状态，在Epoller中分为三种状态：

1
2
3
4
enum EventStatus {
  kNew = -1,
  kAdded = 1,
};

所以具体逻辑为：

  • 如果状态为kNew，调用UpdateEpollEvent(EPOLL_CTL_ADD, ch)，状态置为KAdded
  • 否则表示当前有新事件注册或旧事件注销，调用UpdateEpollEvent(EPOLL_MOD, ch)

PS: 当前所有事件都不关心了应不应该调用UpdateEpollEvent(EPOLL_CTL_DEL, ch)？
这么做的好处是：

  • 在之后没有新事件注册时，可以节省一次系统调用
  • 带来了心理负担，得考虑先EnableXXX()再DisableXXX()，不然会引发下面提到的坏处。

坏处是：

  • 可能之后还会注册新事件，先注销再注册，红黑树需要进行旋转和重染色。
  • 不符合UpdateChannel()的语义，与RemoveChannel()语义重复

由于弊大于利，故即使当前没有感兴趣的事件，也不注销内核事件表中的事件

1
2
3
4
5
6
7
8
9
10
11
12
void Epoller::UpdateChannel(Channel* ch) {
  AssertInThread();

  int index = ch->GetIndex();

  if (index == kNew) {
    UpdateEpollEvent(EPOLL_CTL_ADD, ch);
    ch->SetIndex(kAdded);
  } else { // ch->GetIndex() = kAdded
    UpdateEpollEvent(EPOLL_CTL_MOD, ch);
  }
}

RemoveChannel

删除API比较简单，因为只需要针对KAdded的Channel进行删除罢了。

1
2
3
4
5
6
7
8
9
10
11
12
13
void Epoller::RemoveChannel(Channel* ch) {
  AssertInThread();
  
  int fd = ch->GetFd();

  auto index = ch->GetIndex();
  
  if (index == kAdded) {
    UpdateEpollEvent(EPOLL_CTL_DEL, ch);
    // set status to kNew, then we can add it again
    ch->SetIndex(kNew);
  }
}

Poll

Poll说白了就是epoll_wait()的wrapper，只不过需要考虑转换为Channel，与库兼容。
前面也说到了需要struct epoll_data[]去接受就绪事件，用std::vector表示可以通过resize()来进行动态扩展，如果返回的就绪事件数与容器大小相等，那么进行扩容，因为下次可能会有更多的事件就绪，并且该容器不考虑主动缩容，避免导致多余的扩容->缩容->扩容->缩容->…。
顺便一提，初始大小是16。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
TimeStamp Epoller::Poll(int ms, ChannelVec& active_channels) {
  AssertInThread();

  int ev_nums = ::epoll_wait(epoll_fd_, 
                             events_.data(),
                             static_cast<int>(events_.size()), // maxsize
                             ms);

  int saved_errno = errno; 
  TimeStamp now{ TimeStamp::Now() };

  if (ev_nums > 0) {
    LOG_TRACE_KANON << ev_nums << " events are ready";
    FillActiveChannels(ev_nums, active_channels);
    
    // since epoll_wait does not expand space and 
    // not use any modifier member function of std::vector
    // so size is not modified automacally
    // ==> should use resize() instead of reserve()
    if (static_cast<int>(events_.size()) == ev_nums) {
      events_.resize(ev_nums << 1);
    }
  } else if (ev_nums == 0) {
    LOG_TRACE_KANON << "none events ready";
  } else {
    // use saved_errno to avoid misunderstand error
    if (saved_errno != EINTR) {
      errno = saved_errno;
      LOG_SYSERROR << "epoll_wait() error occurred";
    }
  }

  return now;
}

void Epoller::FillActiveChannels(int ev_nums, 
                                 ChannelVec& active_channels) noexcept {  
  assert(ev_nums <= static_cast<int>(events_.size()));
  
  for (int i = 0; i < ev_nums; ++i) {
    // we use the data.ptr so that no need to look up in channels_map_
    auto channel = reinterpret_cast<Channel*>(events_[i].data.ptr);
    assert(!channel->IsNoneEvent());

    int fd = channel->GetFd();KANON_UNUSED(fd);

    channel->SetRevents(events_[i].events);
    active_channels.emplace_back(channel);
  }
}

ET模式的支持

kanon是比较粗糙地在UpdateChannel()中给关注事件加上EPOLL_ET从而启动ET模式。

1
2
3
4
5
6
7
8
9
10
11
12
13
/**
  * \brief Set Epoller working in edge trigger mode
  * 
  * In fact, epolle() cannot be set to edge trigger mode
  * Set this will add EPOLL_ET to interested events of 
  * fd
  */
void Epoller::SetEdgeTriggertMode() noexcept
{ is_et_mode_ = true; }

//! Check if working in edge trigger mode
bool Epoller::IsEdgeTriggerMode() const noexcept
{ return is_et_mode_; }

EventLoop的相关API是转发给Epoller的