如何在Python中使用Linux epoll
内容
- 介绍
- 阻塞套接字编程示例
- 异步套接字和Linux epoll的好处
- epoll的异步套接字编程示例
- 性能考量
- 源代码
介绍
从2.6版开始，Python包含用于访问Linux epoll库的API。本文使用Python3示例简要演示API。
阻塞套接字编程示例
示例1是一个简单的Python服务器，它在8080端口上侦听HTTP请求，将其打印到控制台，然后将HTTP响应发送回客户端。
- 第9行：创建服务套接字
- 第10行：即使最近另一个程序正在同一端口上侦听，也允许在第11行中使用bind（）。否则，直到使用该端口的上一个程序完成一两分钟后，该程序才能运行。
- 第11行：将服务器套接字绑定到此计算机上所有可用IPv4地址的端口8080。
- 第12行：告诉服务器套接字开始接受来自客户端的传入连接。
- 第14行：程序将在此处停止，直到接收到连接为止。发生这种情况时，服务器套接字将在此计算机上创建一个用于与客户端通信的新套接字。这个新的套接字由accept（）调用返回的clientconnection对象表示。地址对象指示连接另一端的IP地址和端口号。
- 第15-17行：组装客户端正在传输的数据，直到传输了完整的HTTP请求为止。 HTTP简易描述中介绍了HTTP协议。
- 第18行：将请求打印到控制台，以验证操作是否正确。
- 第19行：将响应发送给客户端。
- 第20-22行：关闭与客户端以及侦听服务器套接字的连接。
官方的HOWTO对使用Python的套接字编程有更详细的描述。
// Example 1 (All examples use Python 3)
import socket
EOL1 = b\'\n\n\'
EOL2 = b\'\n\r\n\'
response = b\'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n\'
response += b\'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n\'
response += b\'Hello, world!\'
serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serversocket.bind((\'0.0.0.0\', 8080))
serversocket.listen(1)
connectiontoclient, address = serversocket.accept()
request = b\'\'
while EOL1 not in request and EOL2 not in request:
request += connectiontoclient.recv(1024)
print(request.decode())
connectiontoclient.send(response)
connectiontoclient.close()
serversocket.close()
示例2在第15行中添加了一个循环，以重复处理客户端连接，直到被用户中断（例如，键盘中断）。 这更清楚地说明了服务器套接字从未用于与客户端交换数据。 而是，它接受来自客户端的连接，然后在服务器计算机上创建用于与客户端通信的新套接字。
第23-24行的finally语句块可确保侦听服务器套接字始终关闭，即使发生异常也是如此。
// Example 2
import socket
EOL1 = b\'\n\n\'
EOL2 = b\'\n\r\n\'
response = b\'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n\'
response += b\'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n\'
response += b\'Hello, world!\'
serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serversocket.bind((\'0.0.0.0\', 8080))
serversocket.listen(1)
try:
while True:
connectiontoclient, address = serversocket.accept()
request = b\'\'
while EOL1 not in request and EOL2 not in request:
request += connectiontoclient.recv(1024)
print(\'-\'*40 + \'\n\' + request.decode()[:-2])
connectiontoclient.send(response)
connectiontoclient.close()
finally:
serversocket.close()
异步套接字和Linux epoll的好处
示例2中显示的套接字称为阻塞套接字，因为Python程序会停止运行直到事件发生。第16行中的accept（）调用将阻塞，直到从客户端接收到连接为止。第19行中的recv（）调用将阻塞，直到从客户端接收到数据为止（或直到没有其他数据要接收为止）。第21行中的send（）调用将阻塞，直到Linux将所有返回给客户端的数据排队等待准备传输。
当程序使用阻塞套接字时，它通常使用一个线程（甚至是专用进程）在每个套接字上进行通信。主程序线程将包含侦听服务器套接字，该套接字接受来自客户端的传入连接。它将一次接受这些连接，将新创建的套接字传递给一个单独的线程，然后该线程将与客户端进行交互。因为这些线程中的每一个仅与一个客户端通信，所以任何阻塞都不会阻止其他线程执行其各自的任务。
将阻塞套接字与多个线程一起使用会导致代码简单明了，但存在许多缺点。 共享资源时，可能难以确保线程适当协作。 在只有一个CPU的计算机上，这种编程风格的效率可能较低。
C10K问题讨论了用于处理多个并发套接字的一些替代方法，例如异步套接字的使用。 这些套接字在某些事件发生之前不会阻塞。 而是，程序在异步套接字上执行一个操作，并立即通知该操作成功还是失败。 该信息使程序可以决定如何进行。 由于异步套接字是非阻塞的，因此不需要多个执行线程。 所有工作都可以在单个线程中完成。 这种单线程方法有其自身的挑战，但对于许多程序来说可能是一个不错的选择。 它也可以与多线程方法结合使用：使用单线程的异步套接字可以用于服务器的网络组件，而线程可以用于访问其他阻塞资源，例如 数据库。
Linux有许多用于管理异步套接字的机制，其中三种由Python select，poll和epoll API公开。 epoll和poll比select更好，因为Python程序不必检查每个套接字中是否有感兴趣的事件。 相反，它可以依靠操作系统来告诉它哪些套接字可能发生这些事件。 epoll比poll更好，因为它不需要操作系统每次在Python程序查询时都检查所有套接字中是否有感兴趣的事件。 相反，Linux会跟踪这些事件的发生情况，并在由Python查询时返回一个列表。 这些图显示了使用数千个并行套接字连接时epoll的优势。
epoll的异步套接字编程示例
使用epoll的程序通常按以下顺序执行操作：
- 创建一个epoll对象
- 告诉epoll对象监视特定套接字上的特定事件
- 询问epoll对象，自上次查询以来，哪些套接字可能已经发生了指定的事件
- 在这些套接字上执行一些操作
- 告诉epoll对象修改要监视的套接字和/或事件的列表
- 重复步骤3至5，直到完成
- 销毁epoll对象
示例3复制了示例2的功能然而使用了异步套接字。 该程序更加复杂，因为单个线程正在与多个客户端进行通信交互。
- 第1行：select模块包含epoll功能。
- 第13行：由于默认情况下套接字是阻塞的，因此使用非阻塞（异步）模式是必需的。
- 第15行：创建一个epoll对象。
- 第16行：对服务器套接字上的读取事件感兴趣。只要服务器套接字接受套接字连接，就会发生读取事件。
- 第19行：连接字典将文件描述符（整数）映射到它们相应的网络连接对象。
- 第21行：查询epoll对象以查明是否可能发生了感兴趣的事件。参数“ 1”表示我们愿意等待一秒钟以等待此类事件的发生。如果在此查询之前 发生了任何感兴趣的事件，该查询将立即返回并列出这些事件。
- 第22行：事件以（文件号，事件代码）元组的序列返回。 fileno是文件描述符的同义词，并且始终是整数。
- 第23行：如果套接字服务器上发生读取事件，则可能已经创建了新的套接字连接。
- 第25行：将新套接字设置为非阻塞模式。
- 第26行：对新套接字的读取（EPOLLIN）事件感兴趣。
- 第31行：如果发生读取事件，则读取从客户端发送的新数据。
- 第33行：收到完整的请求后，然后取消注册对读取事件的兴趣并注册对写入（EPOLLOUT）事件的兴趣。当可以将响应数据发送回客户端时，将发生写事件。
- 第34行：打印完整的请求，表明尽管与客户的通信是交错的，但这些数据可以作为整体消息进行组合和处理。
- 第35行：如果客户端套接字上发生了写入事件，则它可以接受新数据以发送到客户端。
- 第36-38行：一次发送一次响应数据，直到将完整的响应传递到操作系统进行传输为止。
- 第39行：发送完完整的响应后，请停止对进一步的读取或写入事件感兴趣。
- 第40行：如果显式关闭了连接，则套接字关闭是可选的。此示例程序使用它来使客户端首先关闭。 shutdown调用通知客户端套接字不应再发送或接收任何数据，并且将使行为良好的客户端从其末端关闭套接字连接。
- 第41行：HUP（挂断）事件表示客户端套接字已断开连接（即已关闭），因此该端也已关闭。无需注册对HUP事件的兴趣。它们始终显示在向epoll对象注册的套接字上。
- 第42行：取消对此套接字连接的兴趣。
- 43行：关闭套接字连接。
- 第18-45行：包含了try-catch块，因为示例程序很可能会被KeyboardInterrupt异常中断
- 第46-48行：不需要关闭打开的套接字连接，因为Python会在程序终止时关闭它们。出于良好的形式将它们包括在内。
// Example 3
import socket, select
EOL1 = b\'\n\n\'
EOL2 = b\'\n\r\n\'
response = b\'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n\'
response += b\'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n\'
response += b\'Hello, world!\'
serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serversocket.bind((\'0.0.0.0\', 8080))
serversocket.listen(1)
serversocket.setblocking(0)
epoll = select.epoll()
epoll.register(serversocket.fileno(), select.EPOLLIN)
try:
connections = {}; requests = {}; responses = {}
while True:
events = epoll.poll(1)
for fileno, event in events:
if fileno == serversocket.fileno():
connection, address = serversocket.accept()
connection.setblocking(0)
epoll.register(connection.fileno(), select.EPOLLIN)
connections[connection.fileno()] = connection
requests[connection.fileno()] = b\'\'
responses[connection.fileno()] = response
elif event & select.EPOLLIN:
requests[fileno] += connections[fileno].recv(1024)
if EOL1 in requests[fileno] or EOL2 in requests[fileno]:
epoll.modify(fileno, select.EPOLLOUT)
print(\'-\'*40 + \'\n\' + requests[fileno].decode()[:-2])
elif event & select.EPOLLOUT:
byteswritten = connections[fileno].send(responses[fileno])
responses[fileno] = responses[fileno][byteswritten:]
if len(responses[fileno]) == 0:
epoll.modify(fileno, 0)
connections[fileno].shutdown(socket.SHUT_RDWR)
elif event & select.EPOLLHUP:
epoll.unregister(fileno)
connections[fileno].close()
del connections[fileno]
finally:
epoll.unregister(serversocket.fileno())
epoll.close()
serversocket.close()
epoll具有两种操作模式，称为边沿触发和水平触发。在边缘触发的操作模式下，对epoll.poll（）的调用仅在套接字上发生读取或写入事件之后，才在该套接字上返回一个事件。调用程序必须处理与该事件相关的所有数据，而在后续对epoll.poll（）的调用中没有进一步的通知。当来自特定事件的数据耗尽时，在套接字上进行其他操作的尝试将导致异常。相反，在级别触发的操作模式下，重复调用epoll.poll（）将导致重复关注感兴趣的事件，直到处理完与该事件相关的所有数据为止。电平触发模式下通常不会发生异常。
例如，假设服务器套接字已向epoll对象注册以进行读取事件。在边缘触发模式下，程序将需要接受（）新的套接字连接，直到出现socket.error异常。而在级别触发的操作模式下，可以进行单个accept（）调用，然后可以再次查询epoll对象以获取服务器套接字上的新事件，该事件表示应进行附加的accept（）调用。
示例3使用了电平触发模式，这是默认的操作模式。示例4演示了如何使用边沿触发模式。在示例4中，第25、36和45行引入了循环，直到发生异常为止（否则，其他所有数据将被处理）。第32、38和48行捕获了预期的套接字异常。最后，第16、28、41和51行添加了EPOLLET掩码，该掩码用于设置边沿触发模式。
// Example 4
import socket, select
EOL1 = b\'\n\n\'
EOL2 = b\'\n\r\n\'
response = b\'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n\'
response += b\'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n\'
response += b\'Hello, world!\'
serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serversocket.bind((\'0.0.0.0\', 8080))
serversocket.listen(1)
serversocket.setblocking(0)
epoll = select.epoll()
epoll.register(serversocket.fileno(), select.EPOLLIN | select.EPOLLET)
try:
connections = {}; requests = {}; responses = {}
while True:
events = epoll.poll(1)
for fileno, event in events:
if fileno == serversocket.fileno():
try:
while True:
connection, address = serversocket.accept()
connection.setblocking(0)
epoll.register(connection.fileno(), select.EPOLLIN | select.EPOLLET)
connections[connection.fileno()] = connection
requests[connection.fileno()] = b\'\'
responses[connection.fileno()] = response
except socket.error:
pass
elif event & select.EPOLLIN:
try:
while True:
requests[fileno] += connections[fileno].recv(1024)
except socket.error:
pass
if EOL1 in requests[fileno] or EOL2 in requests[fileno]:
epoll.modify(fileno, select.EPOLLOUT | select.EPOLLET)
print(\'-\'*40 + \'\n\' + requests[fileno].decode()[:-2])
elif event & select.EPOLLOUT:
try:
while len(responses[fileno]) > 0:
byteswritten = connections[fileno].send(responses[fileno])
responses[fileno] = responses[fileno][byteswritten:]
except socket.error:
pass
if len(responses[fileno]) == 0:
epoll.modify(fileno, select.EPOLLET)
connections[fileno].shutdown(socket.SHUT_RDWR)
elif event & select.EPOLLHUP:
epoll.unregister(fileno)
connections[fileno].close()
del connections[fileno]
finally:
epoll.unregister(serversocket.fileno())
epoll.close()
serversocket.close()
由于它们相似，因此在移植使用选择或轮询机制的应用程序时通常使用级别触发模式，而当程序员不需要或希望操作系统提供尽可能多的帮助时，可以使用边缘触发模式。 在管理事件状态。
除了这两种操作模式外，还可以使用EPOLLONESHOT事件掩码向epoll对象注册套接字。 使用此选项时，已注册事件仅对epoll.poll（）的一次调用有效，此后将其自动从要监视的已注册套接字列表中删除。
性能考量
监听积压队列大小
在示例1-4中，第12行显示了对serversocket.listen（）方法的调用。 此方法的参数是侦听积压队列大小。 它告诉操作系统在Python程序接受之前有多少TCP / IP连接要接受并放置在积压队列中。 每次Python程序在服务器套接字上调用accept（）时，都会从队列中删除其中一个连接，并且该插槽可用于另一个传入连接。 如果队列已满，则新的传入连接将被静默忽略，从而导致网络连接客户端不必要的延迟。 生产服务器通常会处理数十个或数百个同时连接，因此1值通常不足。 例如，当使用ab对具有100个并发HTTP 1.0客户端的这些示例程序执行负载测试时，任何小于50的积压值通常会导致性能下降。
TCP选项
TCP_CORK选项可用于“填充”消息，直到它们准备好发送为止。 如示例5的第34和40行所示，此选项对于使用HTTP / 1.1流水线的HTTP服务器可能是一个不错的选择。
// Example 5
import socket, select
EOL1 = b\'\n\n\'
EOL2 = b\'\n\r\n\'
response = b\'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n\'
response += b\'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n\'
response += b\'Hello, world!\'
serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serversocket.bind((\'0.0.0.0\', 8080))
serversocket.listen(1)
serversocket.setblocking(0)
epoll = select.epoll()
epoll.register(serversocket.fileno(), select.EPOLLIN)
try:
connections = {}; requests = {}; responses = {}
while True:
events = epoll.poll(1)
for fileno, event in events:
if fileno == serversocket.fileno():
connection, address = serversocket.accept()
connection.setblocking(0)
epoll.register(connection.fileno(), select.EPOLLIN)
connections[connection.fileno()] = connection
requests[connection.fileno()] = b\'\'
responses[connection.fileno()] = response
elif event & select.EPOLLIN:
requests[fileno] += connections[fileno].recv(1024)
if EOL1 in requests[fileno] or EOL2 in requests[fileno]:
epoll.modify(fileno, select.EPOLLOUT)
connections[fileno].setsockopt(socket.IPPROTO_TCP, socket.TCP_CORK, 1)
print(\'-\'*40 + \'\n\' + requests[fileno].decode()[:-2])
elif event & select.EPOLLOUT:
byteswritten = connections[fileno].send(responses[fileno])
responses[fileno] = responses[fileno][byteswritten:]
if len(responses[fileno]) == 0:
connections[fileno].setsockopt(socket.IPPROTO_TCP, socket.TCP_CORK, 0)
epoll.modify(fileno, 0)
connections[fileno].shutdown(socket.SHUT_RDWR)
elif event & select.EPOLLHUP:
epoll.unregister(fileno)
connections[fileno].close()
del connections[fileno]
finally:
epoll.unregister(serversocket.fileno())
epoll.close()
serversocket.close()
另一方面，TCP_NODELAY选项可用于告诉操作系统任何传递给socket.send（）的数据都应立即发送给客户端，而不要被操作系统缓冲。 如示例6的第14行所示，此选项可能是用于SSH客户端或其他“实时”应用程序的不错的选择。
// Example 6
import socket, select
EOL1 = b\'\n\n\'
EOL2 = b\'\n\r\n\'
response = b\'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n\'
response += b\'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n\'
response += b\'Hello, world!\'
serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serversocket.bind((\'0.0.0.0\', 8080))
serversocket.listen(1)
serversocket.setblocking(0)
serversocket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
epoll = select.epoll()
epoll.register(serversocket.fileno(), select.EPOLLIN)
try:
connections = {}; requests = {}; responses = {}
while True:
events = epoll.poll(1)
for fileno, event in events:
if fileno == serversocket.fileno():
connection, address = serversocket.accept()
connection.setblocking(0)
epoll.register(connection.fileno(), select.EPOLLIN)
connections[connection.fileno()] = connection
requests[connection.fileno()] = b\'\'
responses[connection.fileno()] = response
elif event & select.EPOLLIN:
requests[fileno] += connections[fileno].recv(1024)
if EOL1 in requests[fileno] or EOL2 in requests[fileno]:
epoll.modify(fileno, select.EPOLLOUT)
print(\'-\'*40 + \'\n\' + requests[fileno].decode()[:-2])
elif event & select.EPOLLOUT:
byteswritten = connections[fileno].send(responses[fileno])
responses[fileno] = responses[fileno][byteswritten:]
if len(responses[fileno]) == 0:
epoll.modify(fileno, 0)
connections[fileno].shutdown(socket.SHUT_RDWR)
elif event & select.EPOLLHUP:
epoll.unregister(fileno)
connections[fileno].close()
del connections[fileno]
finally:
epoll.unregister(serversocket.fileno())
epoll.close()
serversocket.close()
源代码
本页上的示例位于公共领域，可以下载。