python IO 多路復用 select poll epoll

 

select 

select 原理

select 是通過系統調用來監視着一個由多個文件描述符（file descriptor）組成的數組，當select()返回后，數組中就緒的文件描述符會被內核修改標記位(其實就是一個整數)，使得進程可以獲得這些文件描述符從而進行后續的讀寫操作。select飾通過遍歷來監視整個數組的，而且每次遍歷都是線性的。

 

select 優點

select目前幾乎在所有的平台上支持，良好跨平台性。

 

select 缺點

• 每次調用select，都需要把fd集合從用戶態拷貝到內核態，這個開銷在fd很多的時候會很大
• 單個進程能夠監視的fd數量存在最大限制，在linux上默認為1024（可以通過修改宏定義或者重新編譯內核的方式提升這個限制）
• 並且由於select的fd是放在數組中，並且每次都要線性遍歷整個數組，當fd很多的時候，開銷也很大

 

python  select 

調用select的函數為r, w, e = select.select(rlist, wlist, xlist[, timeout])，前三個參數都分別是三個列表，數組中的對象均為waitable object:均是整數的文件描述符（file descriptor）或者一個擁有返回文件描述符方法fileno()的對象；

• rlist: 等待讀就緒的list
• wlist: 等待寫就緒的list
• errlist: 等待“異常”的list

 

select方法用來監視文件描述符，如果文件描述符發生變化，則獲取該描述符。

1、這三個list可以是一個空的list，但是接收3個空的list是依賴於系統的(在Linux上是可以接受的，但是在window上是不可以的)。

2、當 rlist  序列中的描述符發生可讀時（accetp和read），則獲取發生變化的描述符並添加到 r  序列中

3、當 wlist  序列中含有描述符時，則將該序列中所有的描述符添加到 w  序列中

4、當 errlist 序列中的句柄發生錯誤時，則將該發生錯誤的句柄添加到 e 序列中

5、當 超時時間 未設置，則select會一直阻塞，直到監聽的描述符發生變化

    當 超時時間 ＝  1 時，那么如果監聽的句柄均無任何變化，則select會阻塞  1  秒，之后返回三個空列表，如果監聽的描述符（fd）有變化，則直接執行。

6、在list中可以接受Ptython的的 file對象(比如 sys.stdin，或者會被 open()和 os.open()返回的object)，socket object將會返回 socket.socket()。也可以自定義類，只要有一個合適的 fileno()的方法(需要真實返回一個文件描述符，而不是一個隨機的整數)。

 

 

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import select, socket

response = b"hello world"

#創建一個server socket
serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serversocket.bind(('localhost', 8080))
serversocket.listen(1)
serversocket.setblocking(0)

inputs = [serversocket, ]

while True:
    rlist, wlist, xlist = select.select(inputs, [], [])
    for sock in rlist:
        ＃ server socket讀就緒
        if sock == serversocket:
            con, addr = serversocket.accept()
            ＃將這個connection添加到讀就緒中
            inputs.append(con)
        else:
            data = sock.recv(1024)
            if data:
                sock.send(response)
                ＃從讀就緒的list中刪除
                inputs.remove(sock)
                sock.close()

 

poll

poll的原理

poll本質上和select沒有區別，只是沒有了最大連接數(linux上默認1024個)的限制，原因是它基於鏈表存儲的。

poll的缺點

poll除了沒有了最大連接數的缺點，其他都和select一樣

在Python中調用poll

• select.poll()，返回一個poll的對象，支持注冊和注銷文件描述符。

• poll.register(fd[, eventmask])注冊一個文件描述符，注冊后，可以通過poll()方法來檢查是否有對應的I/O事件發生。fd可以是i 個整數，或者有返回整數的fileno()方法對象。如果File對象實現了fileno()，也可以當作參數使用。

• eventmask是一個你想去檢查的事件類型，它可以是常量POLLIN, POLLPRI和 POLLOUT的組合。如果缺省，默認會去檢查所有的3種事件類型。

事件常量	意義
POLLIN	有數據讀取
POLLPRT	有數據緊急讀取
POLLOUT	准備輸出:輸出不會阻塞
POLLERR	某些錯誤情況出現
POLLHUP	掛起
POLLNVAL	無效請求:描述無法打開

• poll.modify(fd, eventmask) 修改一個已經存在的fd，和poll.register(fd, eventmask)有相同的作用。如果去嘗試修改一個未經注冊的fd，會引起一個errno為ENOENT的IOError。
• poll.unregister(fd)從poll對象中注銷一個fd。嘗試去注銷一個未經注冊的fd，會引起KeyError。
• poll.poll([timeout])去檢測已經注冊了的文件描述符。會返回一個可能為空的list，list中包含着(fd, event)這樣的二元組。 fd是文件描述符， event是文件描述符對應的事件。如果返回的是一個空的list，則說明超時了且沒有文件描述符有事件發生。timeout的單位是milliseconds，如果設置了timeout，系統將會等待對應的時間。如果timeout缺省或者是None，這個方法將會阻塞直到對應的poll對象有一個事件發生。

 

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import select, socket

response = b"hello world"

serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serversocket.bind(('192.168.199.197', 8080))
serversocket.listen(1)
serversocket.setblocking(0)

#
poll = select.poll()
poll.register(serversocket.fileno(), select.POLLIN)

connections = {}
while True:
    for fd, event in poll.poll():
        if event == select.POLLIN:
            if fd == serversocket.fileno():
                con, addr = serversocket.accept()
                poll.register(con.fileno(), select.POLLIN)
                connections[con.fileno()] = con
            else:
                con = connections[fd]
                data = con.recv(1024)
                if data:
                    poll.modify(con.fileno(), select.POLLOUT)
        elif event == select.POLLOUT:
            con = connections[fd]
            con.send(response)
            poll.unregister(con.fileno())
            con.close()

 

epoll

epoll的原理及改進

在linux2.6（准確來說是2.5.44）由內核直接支持的方法。epoll解決了select和poll的缺點。

• 對於第一個缺點，epoll的解決方法是每次注冊新的事件到epoll中，會把所有的fd拷貝進內核，而不是在等待的時候重復拷貝，保證了每個fd在整個過程中只會拷貝1次。
• 對於第二個缺點，epoll沒有這個限制，它所支持的fd上限是最大可以打開文件的數目，具體數目可以cat /proc/sys/fs/file-max查看，一般來說這個數目和系統內存關系比較大。
• 對於第三個缺點，epoll的解決方法不像select和poll每次對所有fd進行遍歷輪詢所有fd集合，而是在注冊新的事件時，為每個fd指定一個回調函數，當設備就緒的時候，調用這個回調函數，這個回調函數就會把就緒的fd加入一個就緒表中。（所以epoll實際只需要遍歷就緒表）。

epoll同時支持水平觸發和邊緣觸發：

• 水平觸發（level－triggered）：只要滿足條件，就觸發一個事件(只要有數據沒有被獲取，內核就不斷通知你)。e.g:在水平觸發模式下，重復調用epoll.poll()會重復通知關注的event，直到與該event有關的所有數據都已被處理。(select, poll是水平觸發, epoll默認水平觸發)
• 邊緣觸發（edge－triggered）：每當狀態變化時，觸發一個事件。e.g:在邊沿觸發模式中，epoll.poll()在讀或者寫event在socket上面發生后，將只會返回一次event。調用epoll.poll()的程序必須處理所有和這個event相關的數據，隨后的epoll.poll()調用不會再有這個event的通知。

在Python中調用epoll

• select.epoll([sizehint=-1])返回一個epoll對象。

• eventmask

事件常量	意義
EPOLLIN	讀就緒
EPOLLOUT	寫就緒
EPOLLPRI	有數據緊急讀取
EPOLLERR	assoc. fd有錯誤情況發生
EPOLLHUP	assoc. fd發生掛起
EPOLLRT	設置邊緣觸發(ET)（默認的是水平觸發）
EPOLLONESHOT	設置為 one-short 行為，一個事件(event)被拉出后，對應的fd在內部被禁用
EPOLLRDNORM	和 EPOLLIN 相等
EPOLLRDBAND	優先讀取的數據帶(data band)
EPOLLWRNORM	和 EPOLLOUT 相等
EPOLLWRBAND	優先寫的數據帶(data band)
EPOLLMSG	忽視

• epoll.close()關閉epoll對象的文件描述符。
• epoll.fileno返回control fd的文件描述符number。
• epoll.fromfd(fd)用給予的fd來創建一個epoll對象。
• epoll.register(fd[, eventmask])在epoll對象中注冊一個文件描述符。（如果文件描述符已經存在，將會引起一個IOError）
• epoll.modify(fd, eventmask)修改一個已經注冊的文件描述符。
• epoll.unregister(fd)注銷一個文件描述符。
• epoll.poll(timeout=-1[, maxevnets=-1])等待事件，timeout(float)的單位是秒（second）。

 

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