在Python中如何使用Linux的epoll

在Python中如何使用Linux的epoll

目錄

1. 序言
2. 阻塞socket編程示例
3. 異步socket的好處以及Linux epoll
4. 帶epoll的異步socket編程示例
5. 性能注意事項
6. 源代碼

 

序言

從2.6開始，Python包含了訪問Linux epoll庫的API。這篇文章用幾個簡單的python 3例子來展示下這個API。歡迎大家質疑和反饋。

阻塞socket編程示例

示例1用python3.0搭建了一個簡單的服務：在8080端口監聽HTTP請求，把它打印到控制台，並返回一個HTTP響應消息給客戶端。

1. 第9行：創建服務器socket。
2. 第10行：允許在11行使用bind()來監聽指定端口，即使這個端口最近被其他程序監聽。沒有這個設置的話，服務不能運行，直到一兩分鍾后，這個端口不再被之前的程序使用。
3. 第11行：監聽這台機器所有可用的IPv4地址上面的8080端口。
4. 第12行：通知服務端socket開始接受來自客戶端的連接。
5. 第 14行：這行代碼直到接收到一個客戶端連接才會完成。這時，服務端socket會在服務端機器上面創建一個新的socket，用來和客戶端通信。這個新的 socket在代碼里面就是accept()調用返回的clientconnection 對象。返回的address對象代表着客戶端的IP和端口。
6. 第15-17行：組裝從客戶端傳輸過來的數據，直到HTTP請求完成。HTTP協議可以參考這里。
7. 第18行：把請求打印到控制台，驗證操作是否正確。
8. 第19行：發送響應回客戶端。
9. 第20-22行：關閉和客戶端的連接以及服務端監聽socket。

官方howto中對python socket編程有更詳細的描述。

Example 1 (All examples use Python 3)

1  import socket
2
3  EOL1 = b'\n\n'
4  EOL2 = b'\n\r\n'
5  response  = b'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n'
6  response += b'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n'
7  response += b'Hello, world!'
8
9  serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
10  serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
11  serversocket.bind(('0.0.0.0', 8080))
12  serversocket.listen(1)
13
14  connectiontoclient, address = serversocket.accept()
15  request = b''
16  while EOL1 not in request and EOL2 not in request:
17     request += connectiontoclient.recv(1024)
18  print(request.decode())
19  connectiontoclient.send(response)
20  connectiontoclient.close()
21
22  serversocket.close()

 

示例2在15行增加了一個循環來不斷的處理來自客戶端的連接，直到用戶中斷（比如鍵盤中斷）。這個例子更清楚的說明服務端socket從不和客戶端交換數據。相反的，它接收客戶端的連接，然后在這台服務器上面創建一個新的socket用來和客戶端通信。

在23-24行的finally語句，可以確保服務端負責監聽的socket會關閉，即使有異常發生。

Example 2

1  import socket
2
3  EOL1 = b'\n\n'
4  EOL2 = b'\n\r\n'
5  response  = b'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n'
6  response += b'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n'
7  response += b'Hello, world!'
8
9  serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
10  serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
11  serversocket.bind(('0.0.0.0', 8080))
12  serversocket.listen(1)
13
14  try:
15     while True:
16        connectiontoclient, address = serversocket.accept()
17        request = b''
18        while EOL1 not in request and EOL2 not in request:
19            request += connectiontoclient.recv(1024)
20        print('-'*40 + '\n' + request.decode()[:-2])
21        connectiontoclient.send(response)
22        connectiontoclient.close()
23  finally:
24     serversocket.close()

 

異步socket的好處以及Linux epoll

示 例2中的socket叫做阻塞socket，因為python程序會停止運行，直到一個event發生。16行的accept()調用會阻塞，直到接收到 一個客戶端連接。19行的recv()調用會阻塞，直到這次接收客戶端數據完成（或者沒有更多的數據要接收）。21行的send()調用也會阻塞，直到將 這次需要返回給客戶端的數據都放到Linux的發送緩沖隊列中。

當 一個程序使用阻塞socket時，常常使用一個線程（甚至是一個專門的程序）來進行各個socket之間的通信。主程序線程會包含接收客戶端連接的服務端 監聽socket。這個socket一次接收一個客戶端連接，把連接傳給另外一個線程新建的socket去處理。因為這些線程每個只和一個客戶端通信，所 以處理時即便在某幾個點阻塞也沒有關系。這種阻塞並不會對其他線程的處理造成任何影響。

使用多線程、阻塞socket來處理的話，代碼會很直觀，但是也會有不少缺陷。它很難確保線程共享資源沒有問題。而且這種編程風格的程序在只有一個CPU的電腦上面效率更低。

C10K問題探討了一些替代選擇，其一是使用異步socket。 這種socket只有在一些event觸發時才會阻塞。相反，程序在異步socket上面執行一個動作，會立即被告知這個動作是否成功。程序會根據這個信 息決定怎么繼續下面的操作由於異步socket是非阻塞的，就沒有必要再來使用多線程。所有的工作都可以在一個線程中完成。這種單線程模式有它自己的挑 戰，但可以成為很多方案不錯的選擇。它也可以結合多線程一起使用：單線程使用異步socket用於處理服務器的網絡部分，多線程可以用來訪問其他阻塞資 源，比如數據庫。

Linux的2.6內核有一系列機制來管理異 步socket，其中3個有對應的Python的API：select、poll和epoll。epoll和pool比select更好，因為 Python程序不需要檢查每一個socket感興趣的event。相反，它可以依賴操作系統來告訴它哪些socket可能有這些event。epoll 比pool更好，因為它不要求操作系統每次都去檢查python程序需要的所有socket感興趣的event。而是Linux在event發生的時候會 跟蹤到，並在Python需要的時候返回一個列表。因此epoll對於大量（成千上萬）並發socket連接，是更有效率和可擴展的機制，可以看這里的圖片。

帶epoll的異步socket編程示例

程序中使用epoll的順序大都如下：

1. 創建一個epoll對象
2. 告訴epoll對象監控指定socket的指定event
3. 詢問epoll對象自從上次查詢以后有哪些socket可能有指定的event發生
4. 在這些socket上面執行一些動作
5. 告訴epool對象去修改socket列表和（或者）event監控
6. 重復步驟3到5，直到完成
7. 銷毀epoll對象

示例3重復了示例2的功能，同時使用異步socket。這個程序更為復雜，因為一個線程要交錯與多個客戶端通信。

1. 第1行：select模塊包含epoll功能。
2. 第13行：因為socket默認是阻塞的，所以需要使用非阻塞（異步）模式。
3. 第15行：創建一個epoll對象。
4. 第16行：在服務端socket上面注冊對讀event的關注。一個讀event隨時會觸發服務端socket去接收一個socket連接。
5. 第19行：字典connections映射文件描述符（整數）到其相應的網絡連接對象。
6. 第21行：查詢epoll對象，看是否有任何關注的event被觸發。參數“1”表示，我們會等待1秒來看是否有event發生。如果有任何我們感興趣的event發生在這次查詢之前，這個查詢就會帶着這些event的列表立即返回。
7. 第22行：event作為一個序列（fileno，event code）的元組返回。fileno是文件描述符的代名詞，始終是一個整數。
8. 第23行：如果一個讀event在服務端sockt發生，就會有一個新的socket連接可能被創建。
9. 第25行：設置新的socket為非阻塞模式。
10. 第26行：為新的socket注冊對讀（EPOLLIN）event的關注。
11. 第31行：如果發生一個讀event，就讀取從客戶端發送過來的新數據。
12. 第33行：一旦完成請求已收到，就注銷對讀event的關注，注冊對寫（EPOLLOUT）event的關注。寫event發生的時候，會回復數據給客戶端。
13. 第34行：打印完整的請求，證明雖然與客戶端的通信是交錯進行的，但數據可以作為一個整體來組裝和處理。
14. 第35行：如果一個寫event在一個客戶端socket上面發生，它會接受新的數據以便發送到客戶端。
15. 第36-38行：每次發送一部分響應數據，直到完整的響應數據都已經發送給操作系統等待傳輸給客戶端。
16. 第39行：一旦完整的響應數據發送完成，就不再關注讀或者寫event。
17. 第40行：如果一個連接顯式關閉，那么socket shutdown是可選的。本示例程序這樣使用，是為了讓客戶端首先關閉。shutdown調用會通知客戶端socket沒有更多的數據應該被發送或接收，並會讓功能正常的客戶端關閉自己的socket連接。
18. 第41行：HUP（掛起）event表明客戶端socket已經斷開（即關閉），所以服務端也需要關閉。沒有必要注冊對HUP event的關注。在socket上面，它們總是會被epoll對象注冊。
19. 第42行：注銷對此socket連接的關注。
20. 第43行：關閉socket連接。
21. 第18-45行：使用try-catch，因為該示例程序最有可能被KeyboardInterrupt異常中斷。
22. 第46-48行：打開的socket連接不需要關閉，因為Python會在程序結束的時候關閉。這里顯式關閉是一個好的代碼習慣。

 

Example 3

1  import socket, select
2
3  EOL1 = b'\n\n'
4  EOL2 = b'\n\r\n'
5  response  = b'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n'
6  response += b'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n'
7  response += b'Hello, world!'
8
9  serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
10  serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
11  serversocket.bind(('0.0.0.0', 8080))
12  serversocket.listen(1)
13  serversocket.setblocking(0)
14
15  epoll = select.epoll()
16  epoll.register(serversocket.fileno(), select.EPOLLIN)
17
18  try:
19     connections = {}; requests = {}; responses = {}
20     while True:
21        events = epoll.poll(1)
22        for fileno, event in events:
23           if fileno == serversocket.fileno():
24              connection, address = serversocket.accept()
25              connection.setblocking(0)
26              epoll.register(connection.fileno(), select.EPOLLIN)
27              connections[connection.fileno()] = connection
28              requests[connection.fileno()] = b''
29              responses[connection.fileno()] = response
30           elif event & select.EPOLLIN:
31              requests[fileno] += connections[fileno].recv(1024)
32              if EOL1 in requests[fileno] or EOL2 in requests[fileno]:
33                 epoll.modify(fileno, select.EPOLLOUT)
34                 print('-'*40 + '\n' + requests[fileno].decode()[:-2])
35           elif event & select.EPOLLOUT:
36              byteswritten = connections[fileno].send(responses[fileno])
37              responses[fileno] = responses[fileno][byteswritten:]
38              if len(responses[fileno]) == 0:
39                 epoll.modify(fileno, 0)
40                 connections[fileno].shutdown(socket.SHUT_RDWR)
41           elif event & select.EPOLLHUP:
42              epoll.unregister(fileno)
43              connections[fileno].close()
44              del connections[fileno]
45  finally:
46     epoll.unregister(serversocket.fileno())
47     epoll.close()
48     serversocket.close()

 

epoll有兩種操作模式，稱為邊沿觸發和水平觸發 。在邊沿觸發模式中，epoll.poll（）在讀或者寫event在socket上面發生后，將只會返回一次event。調用epoll.poll() 的程序必須處理所有和這個event相關的數據，隨后的epoll.poll()調用不會再有這個event的通知。當一個特定event的數據耗盡時， 進一步嘗試操作socket將導致一個異常。相反，在水平觸發模式下，重復調用epoll.poll（）會重復通知關注的event，直到與該event 有關的所有數據都已被處理。在水平模式下通常沒有異常。

例如， 假設一個服務端socket已經為一個epoll對象注冊了讀event。在邊沿觸發模式下，程序需要一直accept()新的socket連接，直到一 個socket.error的異常發生。而在水平觸發模式下，一個accept()調用后，epoll對象會被服務端socket再次詢問是否有新的 event，以確定下一個accept()是否應該被調用。

示 例3使用水平觸發模式，這是操作的默認模式。示例4演示了如何使用邊沿觸發模式。在示例4中，第25，36和45行引入循環，直到出現異常才退出（或者所 有其他已知的數據都被處理）。第32，38和48行捕獲預期的socket異常。最后，第16，28，41和51行添加EPOLLET掩碼，用來設置為邊 沿觸發模式。

Example 4

1  import socket, select
2
3  EOL1 = b'\n\n'
4  EOL2 = b'\n\r\n'
5  response  = b'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n'
6  response += b'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n'
7  response += b'Hello, world!'
8
9  serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
10  serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
11  serversocket.bind(('0.0.0.0', 8080))
12  serversocket.listen(1)
13  serversocket.setblocking(0)
14
15  epoll = select.epoll()
16  epoll.register(serversocket.fileno(), select.EPOLLIN | select.EPOLLET)
17
18  try:
19     connections = {}; requests = {}; responses = {}
20     while True:
21        events = epoll.poll(1)
22        for fileno, event in events:
23           if fileno == serversocket.fileno():
24              try:
25                 while True:
26                    connection, address = serversocket.accept()
27                    connection.setblocking(0)
28                    epoll.register(connection.fileno(), select.EPOLLIN | select.EPOLLET)
29                    connections[connection.fileno()] = connection
30                    requests[connection.fileno()] = b''
31                    responses[connection.fileno()] = response
32              except socket.error:
33                 pass
34           elif event & select.EPOLLIN:
35              try:
36                 while True:
37                    requests[fileno] += connections[fileno].recv(1024)
38              except socket.error:
39                 pass
40              if EOL1 in requests[fileno] or EOL2 in requests[fileno]:
41                 epoll.modify(fileno, select.EPOLLOUT | select.EPOLLET)
42                 print('-'*40 + '\n' + requests[fileno].decode()[:-2])
43           elif event & select.EPOLLOUT:
44              try:
45                 while len(responses[fileno]) > 0:
46                    byteswritten = connections[fileno].send(responses[fileno])
47                    responses[fileno] = responses[fileno][byteswritten:]
48              except socket.error:
49                 pass
50              if len(responses[fileno]) == 0:
51                 epoll.modify(fileno, select.EPOLLET)
52                 connections[fileno].shutdown(socket.SHUT_RDWR)
53           elif event & select.EPOLLHUP:
54              epoll.unregister(fileno)
55              connections[fileno].close()
56              del connections[fileno]
57  finally:
58     epoll.unregister(serversocket.fileno())
59     epoll.close()
60     serversocket.close()

 

這兩種模式是類似的，水平觸發模式常被用在移植使用select或者poll機制的應用程序時，而邊沿觸發模式可以用在當程序員不需要或不想要操作系統協助管理event狀態時。

除了這兩種操作模式，epoll對象也可以注冊socket使用EPOLLONESHOTevent掩碼。當使用這個選項時，注冊的event只適用於一個epoll.poll()調用，調用之后它會自動從被監視的socket注冊列表中移除。

性能注意事項

監聽緩沖區隊列大小

在 示例1-4中，第12行都調用了serversocket.listen()方法。此方法的參數就是監聽緩沖區隊列的大小。它告訴操作系統可以接收多少 TCP/IP連接，並放到緩沖區隊列中等待Pytohn程序接收。Python程序每次在服務端socket上面調用accept()，就會有一個連接從 緩沖區隊列中移除，一個新的連接可以進入緩沖區隊列。如果隊列已滿，新的連接都會被忽略，這會對網絡連接的客戶端造成不必要的延遲。在生產服務器上，通常 要處理幾十或幾百個並發連接，所以值1通常是不夠的。比如，當使用ab模擬100個並發HTTP 1.0客戶端，對上面的幾個示例進行負載測試，如果緩沖區隊列的值小於50，就會引起性能下降。

TCP選項

TCP_CORK選項可以用來“封存”消息，直到他們准備好發送。如示例5的第34行和第40行所示，這個選項對於使用HTTP/1.1流水線技術的HTTP服務端來說，可能是一個很好的選擇。

Example 5

1  import socket, select
2
3  EOL1 = b'\n\n'
4  EOL2 = b'\n\r\n'
5  response  = b'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n'
6  response += b'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n'
7  response += b'Hello, world!'
8
9  serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
10  serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
11  serversocket.bind(('0.0.0.0', 8080))
12  serversocket.listen(1)
13  serversocket.setblocking(0)
14
15  epoll = select.epoll()
16  epoll.register(serversocket.fileno(), select.EPOLLIN)
17
18 try:
19     connections = {}; requests = {}; responses = {}
20     while True:
21        events = epoll.poll(1)
22        for fileno, event in events:
23           if fileno == serversocket.fileno():
24              connection, address = serversocket.accept()
25              connection.setblocking(0)
26              epoll.register(connection.fileno(), select.EPOLLIN)
27              connections[connection.fileno()] = connection
28              requests[connection.fileno()] = b''
29              responses[connection.fileno()] = response
30           elif event & select.EPOLLIN:
31              requests[fileno] += connections[fileno].recv(1024)
32              if EOL1 in requests[fileno] or EOL2 in requests[fileno]:
33                 epoll.modify(fileno, select.EPOLLOUT)
34                 connections[fileno].setsockopt(socket.IPPROTO_TCP, socket.TCP_CORK, 1)
35                 print('-'*40 + '\n' + requests[fileno].decode()[:-2])
36           elif event & select.EPOLLOUT:
37              byteswritten = connections[fileno].send(responses[fileno])
38              responses[fileno] = responses[fileno][byteswritten:]
39              if len(responses[fileno]) == 0:
40                 connections[fileno].setsockopt(socket.IPPROTO_TCP, socket.TCP_CORK, 0)
41                 epoll.modify(fileno, 0)
42                 connections[fileno].shutdown(socket.SHUT_RDWR)
43           elif event & select.EPOLLHUP:
44              epoll.unregister(fileno)
45              connections[fileno].close()
46              del connections[fileno]
47  finally:
48     epoll.unregister(serversocket.fileno())
49     epoll.close()
50     serversocket.close()

另一方面， TCP_NODELAY選項可以用來告訴操作系統，任何傳遞給socket.send()的數據，不再緩存，要立即發送給客戶端。如示例6的14行所示，這個選項對於使用一個SSH客戶端或其他“實時”應用來說，可能是一個很好的選擇。

Example 6

1  import socket, select
2
3  EOL1 = b'\n\n'
4  EOL2 = b'\n\r\n'
5  response  = b'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n'
6  response += b'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n'
7  response += b'Hello, world!'
8
9  serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
10  serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
11  serversocket.bind(('0.0.0.0', 8080))
12  serversocket.listen(1)
13  serversocket.setblocking(0)
14  serversocket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
15
16  epoll = select.epoll()
17  epoll.register(serversocket.fileno(), select.EPOLLIN)
18
19 try:
20     connections = {}; requests = {}; responses = {}
21     while True:
22        events = epoll.poll(1)
23        for fileno, event in events:
24           if fileno == serversocket.fileno():
25              connection, address = serversocket.accept()
26              connection.setblocking(0)
27              epoll.register(connection.fileno(), select.EPOLLIN)
28              connections[connection.fileno()] = connection
29              requests[connection.fileno()] = b''
30              responses[connection.fileno()] = response
31           elif event & select.EPOLLIN:
32              requests[fileno] += connections[fileno].recv(1024)
33              if EOL1 in requests[fileno] or EOL2 in requests[fileno]:
34                 epoll.modify(fileno, select.EPOLLOUT)
35                 print('-'*40 + '\n' + requests[fileno].decode()[:-2])
36           elif event & select.EPOLLOUT:
37              byteswritten = connections[fileno].send(responses[fileno])
38              responses[fileno] = responses[fileno][byteswritten:]
39              if len(responses[fileno]) == 0:
40                 epoll.modify(fileno, 0)
41                 connections[fileno].shutdown(socket.SHUT_RDWR)
42           elif event & select.EPOLLHUP:
43              epoll.unregister(fileno)
44              connections[fileno].close()
45              del connections[fileno]
46  finally:
47     epoll.unregister(serversocket.fileno())
48     epoll.close()
49     serversocket.close()

源代碼

此頁面上的示例不受版權限制，這里提供下載 。