sellect、poll、epoll

時間 2019-11-30

標籤 sellect poll epoll 简体版

原文原文鏈接

http://www.cnblogs.com/alex3714/p/4372426.htmlhtml

select
select最先於1983年出如今4.2BSD中，它經過一個select()系統調用來監視多個文件描述符的數組，當select()返回後，該數組中就緒的文件描述符便會被內核修改標誌位，使得進程能夠得到這些文件描述符從而進行後續的讀寫操做。python

select目前幾乎在全部的平臺上支持，其良好跨平臺支持也是它的一個優勢，事實上從如今看來，這也是它所剩很少的優勢之一。數組

select的一個缺點在於單個進程可以監視的文件描述符的數量存在最大限制，在Linux上通常爲1024，不過能夠經過修改宏定義甚至從新編譯內核的方式提高這一限制。緩存

另外，select()所維護的存儲大量文件描述符的數據結構，隨着文件描述符數量的增大，其複製的開銷也線性增加。同時，因爲網絡響應時間的延遲使得大量TCP鏈接處於非活躍狀態，但調用select()會對全部socket進行一次線性掃描，因此這也浪費了必定的開銷。服務器

poll
poll在1986年誕生於System V Release 3，它和select在本質上沒有多大差異，可是poll沒有最大文件描述符數量的限制。網絡

poll和select一樣存在一個缺點就是，包含大量文件描述符的數組被總體複製於用戶態和內核的地址空間之間，而不論這些文件描述符是否就緒，它的開銷隨着文件描述符數量的增長而線性增大。數據結構

另外，select()和poll()將就緒的文件描述符告訴進程後，若是進程沒有對其進行IO操做，那麼下次調用select()和poll()的時候將再次報告這些文件描述符，因此它們通常不會丟失就緒的消息，這種方式稱爲水平觸發（Level Triggered）。併發

epoll
直到Linux2.6纔出現了由內核直接支持的實現方法，那就是epoll，它幾乎具有了以前所說的一切優勢，被公認爲Linux2.6下性能最好的多路I/O就緒通知方法。app

epoll能夠同時支持水平觸發和邊緣觸發（Edge Triggered，只告訴進程哪些文件描述符剛剛變爲就緒狀態，它只說一遍，若是咱們沒有采起行動，那麼它將不會再次告知，這種方式稱爲邊緣觸發），理論上邊緣觸發的性能要更高一些，可是代碼實現至關複雜。socket

epoll一樣只告知那些就緒的文件描述符，並且當咱們調用epoll_wait()、create、regester、unregest得到就緒文件描述符時，返回的不是實際的描述符，而是一個表明就緒描述符數量的值，你只須要去epoll指定的一個數組中依次取得相應數量的文件描述符便可，這裏也使用了內存映射（mmap）技術，這樣便完全省掉了這些文件描述符在系統調用時複製的開銷。

另外一個本質的改進在於epoll採用基於事件的就緒通知方式。在select/poll中，進程只有在調用必定的方法後，內核纔對全部監視的文件描述符進行掃描，而epoll事先經過epoll_ctl()來註冊一個文件描述符，一旦基於某個文件描述符就緒時，內核會採用相似callback的回調機制，迅速激活這個文件描述符，當進程調用epoll_wait()時便獲得通知。

Python select

Python的select()方法直接調用操做系統的IO接口，它監控sockets,open files, and pipes(全部帶fileno()方法的文件句柄)什麼時候變成readable 和writeable, 或者通訊錯誤，select()使得同時監控多個鏈接變的簡單，而且這比寫一個長循環來等待和監控多客戶端鏈接要高效，由於select直接經過操做系統提供的C的網絡接口進行操做，而不是經過Python的解釋器。

注意：Using Python’s file objects with select() works for Unix, supported under Windows yet.

接下來經過echo server例子以瞭解select 是如何經過單進程實現同時處理多個非阻塞的socket鏈接的

select_server:

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

# 導入模塊
import select
import socket
import sys
import queue

# 建立TCP/IP socket套接字實例server
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 設置 socket爲非阻塞
server.setblocking(False)

## 綁定server socket地址(ip、端口)
server_address = ('localhost', 10000)       #設置服務端地址（即IP和端口）
print(sys.stderr, 'starting up on %s port %s' % server_address)     # 打印啓動信息(ip、端口)
server.bind(server_address)     # server socket實例綁定地址

## 監聽客戶端鏈接,最多容許5個併發鏈接
server.listen(5)
'''
select()方法接收並監控3個通訊列表， 第一個是全部的輸入的data,就是指外部發過來的數據，第2個是監控和接收全部要發出去的data(outgoing data),第3個監控錯誤信息，接下來咱們須要建立2個列表來包含輸入和輸出信息來傳給select().
'''
'''
select.select()
select(rlist, wlist, xlist, timeout=None)
select(rlist, wlist, xlist[, timeout]) -> (rlist, wlist, xlist)
做用：等到一個或多個文件描述符準備好某種I/O
select() 方法中接收並監控三個通訊列表，
        rlist可讀列表，全部輸入的數據，監控外部發起的鏈接及發過來的數據；
        wlist可寫列表，監控並接收全部要發出的數據
        xlist錯誤信息列表，監控錯誤信息

'''
# Sockets from which we expect to read定義可讀列表，保存server socket及已創建的socket鏈接，首先把server socket放進去
inputs = [server]

# Sockets to which we expect to write定義可寫列表，保存須要發送數據的socket鏈接
outputs = []

'''
全部客戶端的進來的鏈接和數據將會被server的主循環程序放在上面的list中處理，咱們如今的server端須要等待鏈接可寫（writable)以後才能過來，而後接收數據並返回(所以不是在接收到數據以後就馬上返回)，由於每一個鏈接要把輸入或輸出的數據先緩存到queue裏，而後再由select取出來再發出去。

Connections are added to and removed from these lists by the server main loop. Since this version of the server is going to wait for a socket to become writable before sending any data (instead of immediately sending the reply), each output connection needs a queue to act as a buffer for the data to be sent through it.
'''
# 定義發送消息緩衝區（字典類型，每一個鏈接與數據爲其中一個key:value對，如socket_connect:'recv_data'），服務端收到的數據不是即收即發的，收到的數據先放到緩衝區,等到掃描可寫時再從緩衝區的隊列中讀取併發送
message_queues = {}

'''
The main portion of the server program loops, calling select() to block and wait for network activity.

下面是此程序的主循環，調用select()時會阻塞和等待直到新的鏈接和數據進來
'''


## 程序主循環
while inputs:       # 循環可讀列表

    # Wait for at least one of the sockets to be ready for processing
    print( '\nwaiting for the next event')          # 打印提示信息
    '''
    當你把inputs,outputs,exceptional(這裏跟inputs共用)傳給select()後，它返回3個新的list，咱們上面將他們分別賦值爲readable,writable,exceptional, 全部在readable list中的socket鏈接表明有數據可接收(recv),全部在writable list中的存放着你能夠對其進行發送(send)操做的socket鏈接，當鏈接通訊出現error時會把error寫到exceptional列表中。

    select() returns three new lists, containing subsets of the contents of the lists passed in. All of the sockets in the readable list have incoming data buffered and available to be read. All of the sockets in the writable list have free space in their buffer and can be written to. The sockets returned in exceptional have had an error (the actual definition of 「exceptional condition」 depends on the platform).
    '''
    '''把 inputs、outputs、exceptional（與inputs共用）列表傳給select(),
    用readable, writable, exceptional來承接select()返回的3個新列表,
    readable 列表中的server表示接收新socket鏈接，其餘的socket鏈接表示有數據可接受
    writable列表中的socket鏈接表示有數據要發送
    exceptional列表的socket表示在通訊過程當中出現有錯誤
    '''
    readable, writable, exceptional = select.select(inputs, outputs, inputs, 2)     #設置2秒超時，默認阻塞的

    '''
    Readable list 中的socket 能夠有3種可能狀態，第一種是若是這個socket是main "server" socket,它負責監聽客戶端的鏈接，若是這個main server socket出如今readable裏，那表明這是server端已經ready來接收一個新的鏈接進來了，爲了讓這個main server能同時處理多個鏈接，在下面的代碼裏，咱們把這個main server的socket設置爲非阻塞模式。

The 「readable」 sockets represent three possible cases. If the socket is the main 「server」 socket, the one being used to listen for connections, then the 「readable」 condition means it is ready to accept another incoming connection. In addition to adding the new connection to the list of inputs to monitor, this section sets the client socket to not block.
    '''

    # Handle inputs處理可讀列表
    '''
    可讀列表中有三種狀態：
    一、server socket監聽 新的socket鏈接
    二、已創建的socket鏈接有數據發過來
    三、客戶端斷開鏈接（這時候recv()的數據爲空）
    '''
    for s in readable:

        if s is server: # socket鏈接爲main "server" socket,表示有新的socket要鏈接進來
            # A "readable" server socket is ready to accept a connection
            connection, client_address = s.accept()     #接收外部鏈接
            print('new connection from', client_address)        #打印客戶地址(ip,port)
            connection.setblocking(False)       #設置新鏈接的socket爲非阻塞
            inputs.append(connection)       #把新鏈接的socket追加到可讀列表

            # Give the connection a queue for data we want to send 在數據緩存區中爲新鏈接的socket分配列隊，以便緩存須要發送給該socket的數據
            message_queues[connection] = queue.Queue()
        else:   #非server socket，即外部已經創建的socket鏈接，這時候就接收數據
            data = s.recv(1024)
            '''
            第二種狀況是這個socket是已經創建了的鏈接，它把數據發了過來，這個時候你就能夠經過recv()來接收它發過來的數據，而後把接收到的數據放到queue裏，這樣你就能夠把接收到的數據再傳回給客戶端了。

The next case is an established connection with a client that has sent data. The data is read with recv(), then placed on the queue so it can be sent through the socket and back to the client.
            '''
            # 這裏作一個簡單的處理，就是客戶發什麼數據過來，服務端就再發什麼數據給客戶。
            # 若是是WEB服務器則，須要對外部發來的data進行解析，如外部請求是讀取一個圖片，那麼服務端就須要把圖片內容放到該socket鏈接在緩衝區中對應的對象中
            if data:        # 有數據，數據爲非空
                # A readable client socket has data
                print(sys.stderr, 'received "%s" from %s' % (data, s.getpeername()) )       #打印接收到的數據及地址信息
                message_queues[s].put(data)     #把接收到的數據放到緩衝區中該鏈接的隊列中
                # Add output channel for response
                # 把該socket鏈接添加到可寫列表中，等到遍歷可寫列表時再發送上面存放的數據
                if s not in outputs:
                    outputs.append(s)
            else:       # 客戶端已經斷開鏈接，因此 recv()接收數據就爲空，服務端須要把該客戶的鏈接關閉（這裏還須要刪除可讀、可寫列表中的該socket鏈接，及緩衝區中的數據）
                '''
                第三種狀況就是這個客戶端已經斷開了，因此你再經過recv()接收到的數據就爲空了，因此這個時候你就能夠把這個跟客戶端的鏈接關閉了
                A readable socket without data available is from a client that has disconnected, and the stream is ready to be closed.
                '''
                # Interpret empty result as closed connection
                print('closing', client_address, 'after reading no data')   #打印客戶端地址信息
                # Stop listening for input on the connection
                if s in outputs:        #若是可寫列表中有該socket鏈接，
                    outputs.remove(s)    # 可寫列表中刪除該socket鏈接
                inputs.remove(s)    # 可讀列表中刪除該socket鏈接
                s.close()           # 關閉該socket鏈接

                # Remove message queue
                del message_queues[s]       # 消息緩衝區中刪除給該socket鏈接分區的queue隊列

    '''
    對於writable list中的socket，也有2種狀態，若是這個客戶端鏈接在跟它對應的queue裏有數據，就把這個數據取出來再發回給這個客戶端，不然就把這個鏈接從output list中移除，這樣下一次循環select()調用時檢測到outputs list中沒有這個鏈接，那就會認爲這個鏈接還處於非活動狀態

There are fewer cases for the writable connections. If there is data in the queue for a connection, the next message is sent. Otherwise, the connection is removed from the list of output connections so that the next time through the loop select() does not indicate that the socket is ready to send data.
    '''
    # Handle outputs處理可寫列表
    '''
    在可寫列表中的socke鏈接有兩種狀態：
    一、有數據要發送
    二、無數據要發送（客戶端爲非活躍狀態）
    '''
    for s in writable:
        try:
            next_msg = message_queues[s].get_nowait()           # 獲取發送數據
        except queue.Empty:     # 無數據要發送
            # No messages waiting so stop checking for writability.
            print('output queue for', s.getpeername(), 'is empty')     #打印客戶地址等提示信息
            outputs.remove(s)       # 可寫列表中刪除該socket鏈接，這樣作也能讓可寫列表不變得那麼大，即不保存非活躍的socket鏈接
        else:        # 有數據要發送
            print( 'sending "%s" to %s' % (next_msg, s.getpeername()))      #打印發送數據及地址信息
            s.send(next_msg)        # 發送數據

    '''
    最後，若是在跟某個socket鏈接通訊過程當中出了錯誤，就把這個鏈接對象在inputs\outputs\message_queue中都刪除，再把鏈接關閉掉
    '''
    # Handle "exceptional conditions"處理錯誤列表
    # 中止監聽錯誤列表中的socket鏈接
    for s in exceptional:
        print('handling exceptional condition for', s.getpeername() )       #打印地址等提示信息
        inputs.remove(s)        # 可讀列表中刪除該socket鏈接
        if s in outputs:        # 可寫列表中刪除該socket鏈接
            outputs.remove(s)
        s.close()       # 閉關該socket鏈接

        # 消息緩存區中刪除該socket對應的queue隊列
        del message_queues[s]

server運行結果：

<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> starting up on localhost port 10000
waiting for the next event

waiting for the next event

new connection from ('127.0.0.1', 60486)
waiting for the next event

new connection from ('127.0.0.1', 60487)
waiting for the next event

new connection from ('127.0.0.1', 60488)
waiting for the next event

new connection from ('127.0.0.1', 60489)
waiting for the next event

<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'This is the message.'" from ('127.0.0.1', 60486)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'This is the message.'" from ('127.0.0.1', 60487)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'This is the message.'" from ('127.0.0.1', 60488)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'This is the message.'" from ('127.0.0.1', 60489)
waiting for the next event

sending "b'This is the message.'" to ('127.0.0.1', 60486)
sending "b'This is the message.'" to ('127.0.0.1', 60487)
sending "b'This is the message.'" to ('127.0.0.1', 60488)
sending "b'This is the message.'" to ('127.0.0.1', 60489)
waiting for the next event

output queue for ('127.0.0.1', 60486) is empty
output queue for ('127.0.0.1', 60487) is empty
output queue for ('127.0.0.1', 60488) is empty
output queue for ('127.0.0.1', 60489) is empty
waiting for the next event

<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'It whill be send.'" from ('127.0.0.1', 60486)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'It whill be send.'" from ('127.0.0.1', 60487)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'It whill be send.'" from ('127.0.0.1', 60488)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'It whill be send.'" from ('127.0.0.1', 60489)
waiting for the next event

sending "b'It whill be send.'" to ('127.0.0.1', 60486)
sending "b'It whill be send.'" to ('127.0.0.1', 60487)
sending "b'It whill be send.'" to ('127.0.0.1', 60488)
sending "b'It whill be send.'" to ('127.0.0.1', 60489)
waiting for the next event

<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'in parts.'" from ('127.0.0.1', 60486)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'in parts.'" from ('127.0.0.1', 60487)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'in parts.'" from ('127.0.0.1', 60488)
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> received "b'in parts.'" from ('127.0.0.1', 60489)
waiting for the next event

sending "b'in parts.'" to ('127.0.0.1', 60486)
sending "b'in parts.'" to ('127.0.0.1', 60487)
sending "b'in parts.'" to ('127.0.0.1', 60488)
sending "b'in parts.'" to ('127.0.0.1', 60489)
waiting for the next event

closing ('127.0.0.1', 60489) after reading no data
waiting for the next event

waiting for the next event

client運行結果：

<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> connecting to localhost port 10000
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503): sending "This is the message."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504): sending "This is the message."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505): sending "This is the message."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506): sending "This is the message."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503) received "b'This is the message.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504) received "b'This is the message.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505) received "b'This is the message.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506) received "b'This is the message.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503): sending "It whill be send."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504): sending "It whill be send."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505): sending "It whill be send."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506): sending "It whill be send."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503) received "b'It whill be send.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504) received "b'It whill be send.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505) received "b'It whill be send.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506) received "b'It whill be send.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503): sending "in parts."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504): sending "in parts."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505): sending "in parts."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506): sending "in parts."
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60503) received "b'in parts.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60504) received "b'in parts.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60505) received "b'in parts.'"
<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'> ('127.0.0.1', 60506) received "b'in parts.'"

epoll_socket_server:

#!/use/local/env python
# -*- coding:utf-8 -*-

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

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

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。