epoll

Posted 2021-02-28 testzcy

Epoll的实现原理： 
Epoll采用事件触发的机制，通过用户创建Epoll对象并注册事件宏监听具体事件，以达到事件发生时触发任务的执行。 
为了更好得理解Epoll的机制，我简单得理解为Socket的交互本身就是两个读、写缓冲区，然后Epoll就是监听这两个缓冲区的数据非空、非满的状态，非空代表有数据读入，非满代表可以有数据写入，然后Epoll检测到用户注册的事件发生以后开始执行对应的IO操作。

对比Select： 
Select同样是采用了监听触发的机制去取代最原始的为每一个连接新建一个线程的处理方式。但是Select的原理是利用函数监听一个可读队列和一个可写队列。当select监听到有事件发生时，它并不知道是哪个客户端的事件，因此需要做一次轮询，查找发生的对象，然后再进行后续的处理，其次是select也有最大监听描述符数量的限制，而epoll是没有这个限制的，当然epoll也是有可优化的点在的，比如如何异步去执行IO操作而不是阻塞等待返回再去检测epoll 的函数返回。这样可以更高效的处理客户端的数据。

为了充分探究每一个事件操作如何影响epoll监听的描述符，我在下面的程序运行中做了一系列的打印，得出了以下的结论： 
1. 在服务端启动没有客户端连接的情况下，程序阻塞在 epoll_fd = select.epoll( ) 处，此时服务端等待客户端连接才能使程序往下跑； 
2.客户端发起连接请求，epoll监听到了事件，epoll_list打印为[ （3,1）]，执行操作服务端接收客户端的连入请求并注册客户端的epoll事件； 
3. 客户端请求发送数据到服务端，在while循环中。epoll_list第一次打印为[ （5,1）]，可读事件触发，此时服务端接受数据，并更改客户端的监听事件为可写，为了后续服务端回传相同的数据给客户端做准备，循环的第二次打印为[（5,4）]，可写事件触发，服务端将储存在变量中的数据回传给客户端，并重新更改客户端的epoll监听事件为可读事件，监听客户端下一次数据的发送。 
4. 整个过程中，需要有三个变量分别保存连接的客户端，客户端发送的数据这两个必要数据。 
注：epoll_list里面存放的是文件描述符编号以及事件返回值

补充解释： 
Epoll的触发方分为水平触发（LT）和边缘触发（ET） 
水平触发是两个缓冲区只要存在没有被读取的数据存在就一定会使epoll触发通知，它监听的是数据； 
边缘触发是两个缓冲区如果发生了数据的变化才会使epoll触发通知，并不在乎数据，它监听的是变化； 
举个栗子： 
如果读入缓冲区新增了一个100k的数据，两种触发方式都一定会工作使epoll触发通知，此时如果程序只是拿了其中50k的数据，还有残存50k在缓冲区内，这时候水平触发一定会继续通知，而边缘触发就不会。

贴上测试的代码分享： 
服务端代码

#!/usr/bin/python
#-*- coding:utf-8 -*-

import socket
import select, errno

if __name__ == "__main__":
    try:
        # 创建 TCP socket 作为监听 socket
        listen_fd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    except socket.error, msg:
        print "create socket failed"

    try:
        # 设置 SO_REUSEADDR 选项
        listen_fd.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    except socket.error, msg:
        print "setsocketopt SO_REUSEADDR failed"

    try:
        # 进行 bind -- 此处未指定 ip 地址，即 bind 了全部网卡 ip 上
        listen_fd.bind((‘‘, 7080))
    except socket.error, msg:
        print "bind failed"

    try:
        # 设置 listen 的 backlog 数
        listen_fd.listen(10)
    except socket.error, msg:
        print msg

    try:
        # 创建 epoll 句柄
        epoll_fd = select.epoll()
        # 向 epoll 句柄中注册 监听 socket 的 可读 事件
        epoll_fd.register(listen_fd.fileno(), select.EPOLLIN)
    except select.error, msg:
        print msg

    connections = {}
    addresses = {}
    datalist = {}
    print "Server Is On!"
    while True:
        # epoll 进行 fd 扫描的地方 -- 未指定超时时间则为阻塞等待
        epoll_list = epoll_fd.poll()
        print "epoll_list:", epoll_list
        for fd, events in epoll_list:
            # 若为监听 fd 被激活
            if fd == listen_fd.fileno():
                # 进行 accept -- 获得连接上来 client 的 ip 和 port，以及 socket 句柄
                conn, addr = listen_fd.accept()
                print "accept connection from %s, %d, fd = %d" % (addr[0], addr[1], conn.fileno())
                # 将连接 socket 设置为 非阻塞
                conn.setblocking(0)
                # 向 epoll 句柄中注册 连接 socket 的 可读 事件
                epoll_fd.register(conn.fileno(), select.EPOLLIN | select.EPOLLET)
                # 将 conn 和 addr 信息分别保存起来
                connections[conn.fileno()] = conn
                addresses[conn.fileno()] = addr
            elif select.EPOLLIN & events:
                # 有 可读 事件激活
                datas = ‘‘
                while True:
                    try:
                        # 从激活 fd 上 recv 10 字节数据
                        data = connections[fd].recv(10)
                        # 若当前没有接收到数据，并且之前的累计数据也没有
                        if not data and not datas:
                            # 从 epoll 句柄中移除该 连接 fd
                            epoll_fd.unregister(fd)
                            # server 侧主动关闭该 连接 fd
                            connections[fd].close()
                            print "%s, %d closed" % (addresses[fd][0], addresses[fd][1])
                            break
                        else:
                            # 将接收到的数据拼接保存在 datas 中
                            datas += data
                    except socket.error, msg:
                        # 在 非阻塞 socket 上进行 recv 需要处理 读穿 的情况
                        # 这里实际上是利用 读穿 出 异常 的方式跳到这里进行后续处理
                        if msg.errno == errno.EAGAIN:
                            # logger.debug("%s receive %s" % (fd, datas))
                            print "receive %s from %s" % (datas, fd)
                            # 将已接收数据保存起来
                            datalist[fd] = datas
                            # 更新 epoll 句柄中连接d 注册事件为 可写
                            epoll_fd.modify(fd, select.EPOLLET | select.EPOLLOUT)
                            break
                        else:
                            # 出错处理
                            epoll_fd.unregister(fd)
                            connections[fd].close()
                            print msg
                            break
            elif select.EPOLLHUP & events:
                # 有 HUP 事件激活
                epoll_fd.unregister(fd)
                connections[fd].close()
                logger.debug("%s, %d closed" % (addresses[fd][0], addresses[fd][1]))
            elif select.EPOLLOUT & events:
                # 有 可写 事件激活
                sendLen = 0
                # 通过 while 循环确保将 buf 中的数据全部发送出去     
                while True:
                    # 将之前收到的数据发回 client -- 通过 sendLen 来控制发送位置
                    sendLen += connections[fd].send(datalist[fd][sendLen:])
                    # 在全部发送完毕后退出 while 循环
                    if sendLen == len(datalist[fd]):
                        break
                # 更新 epoll 句柄中连接 fd 注册事件为 可读
                epoll_fd.modify(fd, select.EPOLLIN | select.EPOLLET)

客户端代码：

import socket
import time

if __name__ == "__main__":
    try:
        connFd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
    except socket.error, msg:
        print msg

    try:
        connFd.connect(("127.0.0.1", 7080))
        print "connect to network server success"
    except socket.error,msg:
        print msg

    for i in range(1, 3):
        data = str(raw_input("Input Something:"))
        if connFd.send(data) != len(data):
            print "send data to network server failed"
            break
        readData = connFd.recv(1024)
        print "Receive:",readData
        time.sleep(1)
    connFd.close()