Python 模組教學 - selectors

Posted on  Jun 5, 2024  in  Python 程式設計 - 中階  by  Amo Chen  ‐ 5 min read

Python 是 1 個追求易用、易學、 battery included 的程式語言， Python 有一些模組會把既有的模組包裝成高階(high-level)模組，除了更易於使用之外，因為一些實作細節或最佳實務也都幫忙打點好了，所以能讓開發者寫出更優雅、簡潔的程式碼。

本文將介紹 selectors 模組，該模組是基於 select 模組的高階模組，也是做 I/O 多工會使用的模組。

本文環境

  • Python 3

前言

本文是「用 Python 學網路程式設計重要概念，從單執行緒到 I/O 多工(I/O multiplexing)」的延伸教學，強烈建議不知道 I/O 多工的讀者先閱讀該文，理解何謂 I/O 多工之後，才能夠對本文有更好的掌握力。

Selectors 模組簡介

This module allows high-level and efficient I/O multiplexing, built upon the select module primitives. Users are encouraged to use this module instead, unless they want precise control over the OS-level primitives used.

Python 原本內建可以處理 I/O multiplexing 的 select 模組，不過 Python 3.4 之後新增更易於使用的 selectors 模組，該模組雖然也是處理 I/O multiplexing 的模組，但其實是 select 模組的再包裝，開發者同樣可以寫出具有效率，但更簡潔的 I/O multiplexing 的程式碼，甚至可以不用特別處理在什麼作業系統應該使用何種 system call, selectors 模組預設就能幫開發者做決定。

也因為如此， Python 官方文件鼓勵開發者們盡量使用 selectors 模組，除非有需求是 selectors 模組無法滿足的，才需要使用 select 模組。

p.s. concurrent.futures 也屬於高階模組，該模組是 ~multiprocessing~ 模組與 threading 模組的再包裝

selectors 模組如何使用？

selectors 模組提供以下 5 種類別，每 1 個類別對應的都是 1 個 system call:

  • SelectSelector
  • PollSelector
  • EpollSelector
  • DevpollSelector
  • KqueueSelector

不過並不是每個作業系統都支援這些 system calls, 以下是各個作業系統支援的情形：

  • Windows 僅支援 select()
  • Linux 支援 select(), poll()
  • Linux 核心版本 2.5.45 (含 2.5.45) 以上支援 epoll()
  • Solaris 支援 devpoll()
  • FreeBSD 支援 kqueue()

如同前文所述，使用 selectors 模組時，我們不需要記住在何種作業系統應該使用哪個 system call, selectors 模組提供開發者 1 個稱為 DefaultSelector 的類別，該模組會自動幫開發者使用最合適的 system call, 因此官方文件建議開發者使用 DefaultSelector 作為優先選擇。

使用 DefaultSelector 的方式如下，僅需要 1 行即可（扣掉 import selectors ）：

import selectors

selector = selectors.DefaultSelector()

至於 DefaultSelector 會使用什麼 system call, 我們可以從它的程式碼看到：

if _can_use('kqueue'):
    DefaultSelector = KqueueSelector
elif _can_use('epoll'):
    DefaultSelector = EpollSelector
elif _can_use('devpoll'):
    DefaultSelector = DevpollSelector
elif _can_use('poll'):
    DefaultSelector = PollSelector
else:
    DefaultSelector = SelectSelector

原則上是按照 kqueue ⭢ epoll ⭢ devpoll ⭢ poll ⭢ select 的順序逐一嘗試（從順序上也可以看到是從效率好的 system call 到效率相對不好的），如果作業系統有支援的話，就會使用它作為 DefaultSelector 。

如果你有指定使用何種 system call 的話，就直接使用對應的類別即可，例如想使用 select() system call 的話：

sel = SelectSelector()

如何接收 I/O Events?

select 模組也實作各種 system calls 所定義的 I/O events, 例如 EPOLLIN , EPOLLOUT , POLLIN , POLLOUT 等等，但是 selector 模組統一簡化為 2 個 I/O events:

  • EVENT_READ
  • EVENT_WRITE

這 2 個 I/O events 分別代表可讀取、可寫入 2 個狀態。

例如，我們設定接收可讀的 I/O event 的話，可以呼叫 selector 的 register(fileobj, events, data=None) 方法，例如設定只接收 EVENT_READ 事件：

import selectors

sel = SelectSelector()
sel.register(
    sock,
    selectors.EVENT_READ,
)

如果想開始接收 event 的話，則不管使用何種類別，都是統一呼叫 select() 方法：

while True:
    events = sel.select()
    for key, mask in events:
        pass # you can handle the event here

select() 方法會回傳 1 個 list, list 內每個值都是 (key, events) 格式的 tuple, 其中 key 是 1 個名稱為 SelectorKey 的 namedtuple, 裡面包含 fileobject, fd, events, data 4 個屬性，這 4 個屬性其實是在呼叫 register 時被包裝到 SelectorKey 裡（原始碼），並不是隨著 event 產生：

def register(self, fileobj, events, data=None):
    if (not events) or (events & ~(EVENT_READ | EVENT_WRITE)):
        raise ValueError("Invalid events: {!r}".format(events))

    key = SelectorKey(fileobj, self._fileobj_lookup(fileobj), events, data)

    if key.fd in self._fd_to_key:
        raise KeyError("{!r} (FD {}) is already registered"
                       .format(fileobj, key.fd))

    self._fd_to_key[key.fd] = key
    return key

所以回來詳細談談 register() 方法。

register(fileobj, events, data=None) 方法的 3 個參數分別是

  1. fileobj

    file object, 是具有 fileno() 方法的物件， fileno() 方法也必須回傳代表 file descriptor 的整數。

  2. events

    開發者想接收的 I/O events, 如果想接收多個 events, 同樣可以使用 | 分隔多個 events 表示，例如 sel.register(fd, EVENT_READ | EVENT_WRITE) 。

  3. data

    預設為 None , 是當 event 發生時，要跟著附加在 SelectorKey 裡的資料，它會被放在 data 這個屬性裡，它可以是包含 function 的任意值。

我們用以下範例理解 data 參數的作用，下列是 1 個很簡單的 socket server, 執行之後會監聽 127.0.0.1 的 1234 通訊埠，一旦觸發 EVENT_READ 時，就會列印 event 裡的 data, 而這個 data 對應的恰好是我們呼叫 register() 方法時所設定的 {'foo': 'bar'} ：

import selectors
import socket

sock = socket.socket()
sock.bind(('localhost', 1234))
sock.listen(100)
sock.setblocking(False)

sel = selectors.DefaultSelector()
sel.register(
    sock,
    selectors.EVENT_READ,
    data={'foo', 'bar'}
)

while True:
    events = sel.select()
    for key, mask in events:
        print(key.data)
    break

上述範例執行之後，可以用以下 telnet 指令測試：

$ telnet 127.0.0.1 1234

執行成功的話，將可以看到 socket server 列印以下字串：

{'foo', 'bar'}

這就是 data 參數的作用，除了能放特定資料之外，也可以像 Python 官方範例一樣，設定 data 為 1 個函式，如此一來，它就能夠做到 event callback 的功能：

import selectors
import socket

sel = selectors.DefaultSelector()

def accept(sock, mask):
    conn, addr = sock.accept()  # Should be ready
    print('accepted', conn, 'from', addr)
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)

def read(conn, mask):
    data = conn.recv(1000)  # Should be ready
    if data:
        print('echoing', repr(data), 'to', conn)
        conn.send(data)  # Hope it won't block
    else:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()

sock = socket.socket()
sock.bind(('localhost', 1234))
sock.listen(100)
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)

while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj, mask)

p.s. 上述程式同樣可以使用指令 telnet 127.0.0.1 1234 測試

簡而言之， data 參數的設計使得使用 selectors 模組能寫出比用 select 模組更簡潔的程式碼！

判別是何種 I/O event

其實 I/O events 是 bitmask, 所以要判斷是何種 I/O event 的話，可以使用 & 運算子，例如：

while True:
    events = sel.select()
    for key, mask in events:
        if mask & selectors.EVENT_READ:
            print('Got EVENT_READ')
        callback = key.data
        callback(key.fileobj, mask)

萬一我想接收 EVENT_READ, EVENT_WRITE 以外的 I/O events 呢？

雖然 select 模組針對各個 system call 定義很多 eventmasks 可以使用，例如還有 POLLERR , POLLPRI , EPOLLERR 等等，不過 selectors 模組將 eventmasks 簡化為 2 種，也就是前文所提到的 EVENT_READ 與 EVENT_WRITE , 這 2 個 eventmasks 分別對應各個 system call 的其中幾個 eventmasks 。

以下是它們實作上的對應。

PollSelector

EVENT_READ ⭢ select.POLLIN

EVENT_WRITE ⭢ select.POLLOUT

EpollSelector

EVENT_READ ⭢ select.EPOLLIN

EVENT_WRITE ⭢ select.EPOLLOUT

DevpollSelector

EVENT_READ ⭢ select.POLLIN

EVENT_WRITE ⭢ select.POLLOUT

KqueueSelector

EVENT_READ ⭢ select.KQ_FILTER_READ (filter) + select.KQ_EV_ADD (flag)

EVENT_WRITE ⭢ select.KQ_FILTER_WRITE (filter) + select.KQ_EV_ADD (flag)

萬一你想接收額外的 events, 那麼只能使用低階的 select 模組了，因為在 register() 方法中都會檢查 events 必須是 EVENT_READ 或者 EVENT_WRITE 才行（原始碼），如果使用其他的 eventmask 將會導致 ValueError 例外錯誤發生：

def register(self, fileobj, events, data=None):
    if (not events) or (events & ~(EVENT_READ | EVENT_WRITE)):
        raise ValueError("Invalid events: {!r}".format(events))

這也是使用 selectors 模組較為可惜的地方。

p.s. 你也可以使用繼承的方式複寫 register() 方法，實作能夠關注不同 events 的方法

取消關注 I/O events

如果要取消接收 I/O events 則是呼叫 unregister(fileobj) 方法，例如：

import selectors
import socket

sock = socket.socket()
sock.bind(('localhost', 1234))
sock.listen(100)
sock.setblocking(False)

sel = selectors.DefaultSelector()
sel.register(sock, selectors.EVENT_READ)

sel.unregister(sock)

修改關注的 I/O events

如果要修改關注的 I/O events 則是呼叫 modify(fileobj, events, data=None) 方法，它的使用方法與 register() 方法完全一樣。

This is equivalent to BaseSelector.unregister(fileobj) followed by BaseSelector.register(fileobj, events, data), except that it can be implemented more efficiently.

實際上 modify() 的運作是先取消關注，再 register() 一次。

總結

以上就是關於 selectors 模組的介紹啦！

Enjoy!

References

https://docs.python.org/3/library/selectors.html

python
Facebook Threads X

對抗久坐職業傷害

研究指出每天增加 2 小時坐著的時間，會增加大腸癌、心臟疾病、肺癌的風險，也造成肩頸、腰背疼痛等常見問題。

然而對抗這些問題，卻只需要工作時定期休息跟伸展身體即可！

你想輕鬆改變現狀嗎？試試看我們的 PomodoRoll 番茄鐘吧！ PomodoRoll 番茄鐘會根據你所設定的專注時間，定期建議你 1 項辦公族適用的伸展運動，幫助你打敗久坐所帶來的傷害！

贊助我們的創作

看完這篇文章了嗎? 休息一下，喝杯咖啡吧！

如果你覺得 MyApollo 有讓你獲得實用的資訊，希望能看到更多的技術分享，邀請你贊助我們一杯咖啡，讓我們有更多的動力與精力繼續提供高品質的文章，感謝你的支持！

你可能也會感興趣的文章