Python——在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()

View Code

 

 

示例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()

View Code

 

異步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()

View Code

 

 

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()

 

View Code

 

這兩種模式是相似的，水平觸發模式常被用在移植使用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()

View Code

 

另外一方面， 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()

View Code

 

源代碼

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