I/O多路复用-EPOLL探索

Posted 2020-10-12 世界有你更精彩

什么是I/O多路复用

I/O多路复用就是通过一种机制，可以监视多个描述符，一旦某个IO能够读写，通知程序进行相应的读写操作。

I/O多路复用的场合

1、当客户处理多个描述字时（通常是交互式输入和网络套接字），必须使用I/O复用

2、如果一个TCP服务器既要处理监听套接字，又要处理已连接套接字，一般也要用到I/O复用

3、如果一个服务器即要处理TCP，又要处理UDP，一般要使用I/O复用

Linux 下I/O多路复用的方式

SELECT、POLL、EPOLL

对比SELECT、POLL、EPOLL

SELECT缺点

每次调用select，都需要把fd集合从用户态拷贝到内核态，开销和fd的数量成正比

每次调用select都需要在内核遍历传递进来的所有fd，开销和fd的数量成正比

select支持的文件描述符数量太少，32位1024个，64位2048个

POLL缺点

poll和select本质相同，使用链表存储文件描述符，没有最大连接数的限制

下图是select、poll、kqueue(FreeBSD平台)、epoll四种方式连接数和时间关系图

上图可以看出，随着fd数量的增大，select、poll的时间消耗非常大，kqueue和epoll基本上没有变化

EPOLL优点

1、它所支持的FD上限是最大可以打开文件的数目,在1GB内存的机器上大约是10万左 右，具体数目可以cat /proc/sys/fs/file-max察看,一般来说这个数目和系统内存关系很大。

2、IO 效率不随FD数目增加而线性下降

3、epoll不像select或poll一样每次都把current轮流加入fd对应的设备等待队列中，而只在epoll_ctl时把current挂一遍（这一遍必不可少）并为每个fd指定一个回调函数，当设备就绪，唤醒等待队列上的等待者时，就会调用这个回调函数，而这个回调函数会把就绪的fd加入一个就绪链表）。epoll_wait的工作实际上就是在这个就绪链表中查看有没有就绪的fd，而不是全部遍历。

 

 

下面是一个场景，一个服务端Server.py，多个客户端订阅，服务端不断的生成消息，客户端订阅后服务端会不断的把消息发送给客户端：

Server：

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8 -*-
 3 
 4 __author__ = \'Andy\'
 5 import socket, select, traceback, time
 6 import threading
 7 import Queue
 8 
 9 gen = Queue.Queue()
10 connections = {}
11 requests = {}
12 responses = {}
13 
14 
15 def run():
16     serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
17     serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
18     serversocket.bind((\'127.0.0.1\', 8080))
19     serversocket.listen(5)
20     serversocket.setblocking(0)
21     epoll = select.epoll()  # 创建一个epoll对象
22     epoll.register(serversocket.fileno(), select.EPOLLIN)  # 给新建的serversocket.fileno注册一个读event
23 
24     try:
25         count = 0
26         while True:
27             events = epoll.poll()  # 激活的fileno举手
28             count += 1
29             for fileno, event in events:
30                 if fileno == serversocket.fileno():  # 当激活的fileno是新建的,给该fileno注册一个读event
31                     connection, address = serversocket.accept()
32                     connection.setblocking(0)
33                     epoll.register(connection.fileno(), select.EPOLLIN)
34                     connections[connection.fileno()] = connection
35                     requests[connection.fileno()] = b\'\'
36                     responses[connection.fileno()] = b""
37                     print "new conn.fileno is %s" % connection.fileno()
38                 elif event & select.EPOLLIN:  # 如果fileno是读event,接收发送来消息,并修改该fileno为写event,下次循环时写数据
39                     print "read event is happing"
40                     requests[fileno] += connections[fileno].recv(1024)
41                     epoll.modify(fileno, select.EPOLLOUT)
42                     print(\'-\' * 40 + \'\\n\' + requests[fileno].decode()[:-2])
43                 elif event & select.EPOLLOUT:  # 如果fileno是写事件,写完后正常的为挂起
44                     if responses[fileno]:
45                         byteswritten = connections[fileno].send(responses[fileno])
46                         responses[fileno] = responses[fileno][byteswritten:]
47                         if len(responses[fileno]) == 0:
48                             epoll.modify(fileno, select.EPOLLOUT)  # 需要向订阅者一直发消息,这里发完后仍为写event
49                             print "change event to write"
50                 elif event & select.EPOLLHUP:
51                     epoll.unregister(fileno)
52                     connections[fileno].close()
53                     del connections[fileno]
54                     print "event is HUP ===%s" % fileno
55         pass
56     except Exception, err:
57         print traceback.print_exc()
58     finally:
59         epoll.unregister(serversocket.fileno())
60         epoll.close()
61         serversocket.close()
62         print "finally"
63 
64 
65 def create_data():
66     count = 1
67     while True:
68         count += 1
69         res = "Message-%s" % count
70         gen.put_nowait(res)  # 把消息放入队列
71         time.sleep(0.5)
72 
73 
74 def update_message():
75     while True:
76         message = gen.get()
77         for res in responses:  # 遍历所有的活跃用户,更新消息
78             responses[res] = message
79         time.sleep(0.05)
80 
81 
82 if __name__ == "__main__":
83     p = threading.Thread(target=create_data)  # 开个线程向队列里放数据
84     p1 = threading.Thread(target=update_message)  # 从队列中取出数据
85     p.start()
86     p1.start()
87     run()
88     p.join()
89     p1.join()

View Code

 

Client:

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8 -*-
 3 
 4 __author__ = \'Andy\'
 5 import socket,time
 6 def sim_client(name,i):
 7     connFd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
 8     connFd.connect(("127.0.0.1", 8080))
 9     connFd.send("Process-%s start subscibe\\n\\n" % name)
10     while True:
11         try:
12             readData = connFd.recv(1024)
13             if readData:
14                 print "*"*40 + "\\n" + readData.decode()
15         except:
16             time.sleep(0.2)
17 
18 
19 if __name__=="__main__":
20     sim_client("progressage",1)

View Code

 

来看一下输出结果

服务端接受订阅的消息：

 

客户端订阅服务端的消息

 

 

才疏学浅，目前只研究了EPOLL的水平触发，还有边缘触发需要去探索

 

 

作者：Andy
出处：http://www.cnblogs.com/onepiece-andy/
本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。