python下使用epoll

Reference： http://blog.csdn.net/hehe123456ZXC/article/details/52526670

 

由於最近想學習如何用epoll寫服務器, 因而找到了一篇介紹的文章. 由於我最近一直看不進技術文章, 因而打算經過翻譯來強迫本身學習. 原文在這裏:

http://scotdoyle.com/python-epoll-howto.html

文章裏面的代碼下載地址：

http://scotdoyle.com/python-epoll-examples.tar.gz

介紹 
從2.6版本開始, Python 提供了使用Linux epoll 的功能. 這篇文章經過3個例子來大體介紹如何使用它. 歡迎提問和反饋.

阻塞式socket通信

第一個例子是一個簡單的python3.0版本的服務器代碼, 監聽8080端口的http請求, 打印結果到命令行, 迴應http response給客戶端.

行 9: 創建服務器的socket 
行 10: 容許11行的bind()操做, 即便其餘程序也在監聽一樣的端口. 否則的話, 這個程序只能在其餘程序中止使用這個端口以後的1到2分鐘後才能執行. 
行 11: 綁定socket到這臺機器上全部IPv4地址上的8080端口. 
行 12: 告訴服務器開始響應從客戶端過來的鏈接請求. 
行 14: 程序會一直停在這裏, 直到創建了一個鏈接. 這個時候, 服務器socket會創建一個新的socket, 用來和客戶端通信. 這個新的socket是accept()的返回值, address對象標示了客戶端的IP地址和端口. 
行 15-17: 接收數據, 直到一個完整的http請求被接收完畢. 這是一個簡單的http服務器實現. 
行 18: 爲了方便驗證, 打印客戶端過來的請求到命令行. 
行 19: 發送迴應. 
行 20-22: 關閉鏈接, 以及服務器的監聽socket.

Example1:

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行加上了一個循環, 用來循環處理客戶端請求, 直到咱們中斷這個過程(在命令行下面輸入鍵盤中斷, 好比Ctrl-C). 這個例子更明顯地表示出來了, 服務器socket並無用來作數據處理, 而是接受服務器過來的鏈接, 而後創建一個新的socket, 用來和客戶端通信.

最後的23-24行確保服務器的監聽socket最後老是close掉, 即便出現了異常.

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

 

 

異步socket和linux epoll的優點 
第2個例子裏面的socket採用的是阻塞方式, 由於python解釋器在出現事件以前都處在中止狀態. 16行的accept()一直阻塞, 直到新的鏈接進來. 19行的recv()也是一直阻塞, 直到從客戶端收到數據(或者直到沒有數據能夠接收). 21行的send()也一直阻塞, 直到全部須要發送給客戶端的數據都交給了linux內核的發送隊列.

當一個程序採用阻塞socket的時候, 它常常採用一個線程(甚至一個進程)一個socket通信的模式. 主線程保留服務器監聽socket, 接受進來的鏈接, 一次接受一個鏈接, 而後把生成的socket交給一個分離的線程去作交互. 由於一個線程只和一個客戶端通信, 在任何位置的阻塞都不會形成問題. 阻塞自己不會影響其餘線程的工做.

多線程阻塞socket模式代碼清晰, 可是有幾個缺陷, 可能很難確保線程間資源共享工做正常, 可能在只有一個CPU的機器上效率低下.

C10K(單機1萬鏈接問題!) 探討了其餘處理並行socket通信的模式. 一種是採用異步socket. socket不會阻塞, 直到特定事件發生. 程序在異步socket上面進行一個特定操做, 而且當即獲得一個結果, 無論執行成功或者失敗. 而後讓程序決定下一步怎麼作. 由於異步socket是非阻塞的, 咱們能夠不採用多線程. 全部的事情均可以在一個線程裏面完成. 雖然這種模式有它須要面對的問題, 它對於特定程序來講仍是不錯的選擇. 也能夠和多線程合起來使用: 單線程的異步socket能夠看成服務器上面處理網絡的一個模塊, 而線程能夠用來訪問阻塞式的資源, 好比數據庫.

Linux 2.6有一些方式來管理異步socket, python API可以用的有3種: select, poll和epoll. epoll和poll比select性能更好, 由於python程序不須要爲了特定的事件去查詢單獨的socket, 而是依賴操做系統來告訴你什麼socket產生了什麼事件. epoll比poll性能更好, 由於它不須要每次python程序查詢的時候, 操做系統都去檢查全部的socket, 在事件產生的時候, linux跟蹤他們, 而後在python程序調用的時候, 返回具體的列表. 因此epoll在大量(上千)並行鏈接下, 是一種更有效率, 伸縮性更強的機制.

採用epoll的異步socket編程示例

採用epoll的程序通常這樣操做:

創建一個epoll對象 
告訴epoll對象, 對於一些socket監控一些事件. 
問epoll, 從上次查詢以來什麼socket產生了什麼事件. 
針對這些socket作特定操做. 
告訴epoll, 修改監控socket和/或監控事件. 
重複第3步到第5步, 直到結束. 
銷燬epoll對象. 
採用異步socket的時候第3步重複了第2步的事情. 這裏的程序更復雜, 由於一個線程須要和多個客戶端交互.

行 1: select模塊帶有epoll功能 
行 13: 由於socket默認是阻塞的, 咱們須要設置成非阻塞(異步)模式. 
行 15: 創建一個epoll對象. 
行 16: 註冊服務器socket, 監聽讀取事件. 服務器socket接收一個鏈接的時候, 產生一個讀取事件. 
行 19: connections表映射文件描述符(file descriptors, 整型)到對應的網絡鏈接對象上面. 
行 21: epoll對象查詢一下是否有感興趣的事件發生, 參數1說明咱們最多等待1秒的時間. 若是有對應事件發生, 馬上會返回一個事件列表. 
行 22: 返回的events是一個(fileno, event code)tuple列表. fileno是文件描述符, 是一個整型數. 
行 23: 若是是服務器socket的事件, 那麼須要針對新的鏈接創建一個socket. 
行 25: 設置socket爲非阻塞模式. 
行 26: 註冊socket的read(EPOLLIN)事件. 
行 31: 若是讀取事件發生, 從客戶端讀取新數據. 
行 33: 一旦完整的http請求接收到, 取消註冊讀取事件, 註冊寫入事件(EPOLLOUT), 寫入事件在可以發送數據回客戶端的時候產生. 
行 34: 打印完整的http請求, 展現即便通信是交錯的, 數據自己是做爲一個完整的信息組合和處理的. 
行 35: 若是寫入事件發生在一個客戶端socket上面, 咱們就能夠發送新數據到客戶端了. 
行s 36-38: 一次發送一部分返回數據, 直到全部數據都交給操做系統的發送隊列. 
行 39: 一旦全部的返回數據都發送完, 取消監聽讀取和寫入事件. 
行 40: 若是鏈接被明確關閉掉, 這一步是可選的. 這個例子採用這個方法是爲了讓客戶端首先斷開, 告訴客戶端沒有數據須要發送和接收了, 而後讓客戶端斷開鏈接. 
行 41: HUP(hang-up)事件表示客戶端斷開了鏈接(好比 closed), 因此服務器這端也會斷開. 不須要註冊HUP事件, 由於它們都會標示到註冊在epoll的socket. 
行 42: 取消註冊. 
行 43: 斷開鏈接. 
行s 18-45: 在這裏的異常捕捉的做用是, 咱們的例子老是採用鍵盤中斷來中止程序執行. 
行s 46-48: 雖然開啓的socket不須要手動關閉, 程序退出的時候會自動關閉, 明確寫出來這樣的代碼, 是更好的編碼風格.

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有2種模式, 邊沿觸發(edge-triggered)和狀態觸發(level-triggered). 邊沿觸發模式下, epoll.poll()在讀取/寫入事件發生的時候只返回一次, 程序必須在後續調用epoll.poll()以前處理完對應事件的全部的數據. 當從一個事件中獲取的數據被用完了, 更多在socket上的處理會產生異常. 相反, 在狀態觸發模式下面, 重複調用epoll.poll()只會返回重複的事件, 直到全部對應的數據都處理完成. 通常狀況下不產生異常.

好比, 一個服務器socket註冊了讀取事件, 邊沿觸發程序須要調用accept創建新的socket鏈接直到一個socket.error錯誤產生, 而後狀態觸發下只須要處理一個單獨的accept(), 而後繼續epoll查詢新的事件來判斷是否有新的accept須要操做.

例子3採用默認的狀態觸發模式, 例子4展現如何用邊沿觸發模式. 例子4中的25, 36和45行引入了循環, 直到錯誤產生(或者全部的數據都處理完了), 32, 38 和48行捕捉socket異常. 最後16, 28, 41 和51行添加EPOLLET mask用來設置邊沿觸發.

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

 

由於比較相似, 狀態觸發常常用在轉換採用select/poll模式的程序上面, 邊沿觸發用在程序員不須要或者不但願操做系統來管理事件狀態的場合上面. 
除了這兩種模式之外, socket常常註冊爲EPOLLONESHOT event mask, 當用到這個選項的時候, 事件只有效一次, 以後會自動從監控的註冊列表中移除.

性能考慮

Listen Backlog Queue Size

在1-4的例子中, 12行顯示了調用serversocket.listen()方法. 參數是監聽等待隊列的大小. 它告訴了操做系統, 在python代碼accept前, 緩存多少TCP/IP鏈接在隊列中. 每次python代碼調用accept()的時候, 一個鏈接從隊列中移除, 爲新的鏈接進來空出一個位置. 若是隊列滿了, 新的鏈接自動放棄, 給客戶端帶來沒必要要的網絡延遲. 一個生產環境下的服務器常常處理幾十或者幾百的同時鏈接數, 因此參數不該該設置爲1. 好比, 當採用 ab 來對這些測試程序進行併發100個http1.0客戶端請求時, 少於50的參數容易形成性能降低.

TCP Options

TCP_CORK 參數能夠設置緩存消息直到一塊兒被髮送, 這個選項, 在例子5的34和40行, 適合給一個實現 http/1.1 pipelining 的服務器來使用.

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

 

源碼在這裏：

http://scotdoyle.com/python-epoll-examples.tar.gz