signal — 為非同步事件設定處理程式

原始碼: Lib/signal.py


此模組提供了在 Python 中使用訊號處理程式的機制。

通用規則

signal.signal() 函式允許定義自定義處理程式,以便在收到訊號時執行。 安裝了少量預設處理程式:SIGPIPE 被忽略(因此管道和套接字上的寫入錯誤可以作為普通的 Python 異常報告),如果父程序沒有更改 SIGINT,則將其轉換為 KeyboardInterrupt 異常。

一旦設定了特定訊號的處理程式,它將保持安裝狀態,直到顯式重置(無論底層實現如何,Python 都會模擬 BSD 樣式介面),但 SIGCHLD 的處理程式除外,它遵循底層實現。

在 WebAssembly 平臺上,訊號是模擬的,因此行為有所不同。 並非所有函式和訊號都可以在這些平臺上使用。

Python 訊號處理程式的執行

Python 訊號處理程式不會在底層 (C) 訊號處理程式內部執行。相反,底層訊號處理程式會設定一個標誌,該標誌告訴 虛擬機器 在稍後的某個時間點(例如,在下一個 位元組碼 指令)執行相應的 Python 訊號處理程式。這會產生以下後果

  • 捕獲 C 程式碼中的無效操作引起的 SIGFPESIGSEGV 等同步錯誤沒有意義。 Python 將從訊號處理程式返回到 C 程式碼,C 程式碼很可能會再次引發相同的訊號,從而導致 Python 明顯掛起。 從 Python 3.3 開始,您可以使用 faulthandler 模組來報告同步錯誤。

  • 完全在 C 中實現的長時間執行的計算(例如對大量文字進行正則表示式匹配)可能會不受任何訊號的影響而執行任意長的時間。 計算完成後,將呼叫 Python 訊號處理程式。

  • 如果處理程式引發異常,則會在主執行緒中“憑空”引發該異常。 有關討論,請參閱下面的說明

訊號和執行緒

Python 訊號處理程式始終在主直譯器的 Python 主執行緒中執行,即使訊號是在另一個執行緒中接收的。 這意味著訊號不能用作執行緒間通訊的手段。 您可以使用 threading 模組中的同步原語代替。

此外,只允許主直譯器的主執行緒設定新的訊號處理程式。

模組內容

在 3.5 版本中更改: 下面列出的訊號 (SIG*)、處理程式 (SIG_DFLSIG_IGN) 和訊號掩碼 (SIG_BLOCKSIG_UNBLOCKSIG_SETMASK) 相關常量被轉換為 列舉(分別為 SignalsHandlersSigmasks)。 getsignal()pthread_sigmask()sigpending()sigwait() 函式返回可讀的 列舉 作為 Signals 物件。

signal 模組定義了三個列舉

class signal.Signals

SIG* 常量和 CTRL_* 常量的 enum.IntEnum 集合。

3.5 版本中新增。

class signal.Handlers

常量 SIG_DFLSIG_IGNenum.IntEnum 集合。

3.5 版本中新增。

class signal.Sigmasks

常量 SIG_BLOCKSIG_UNBLOCKSIG_SETMASKenum.IntEnum 集合。

可用性: Unix。

有關更多資訊,請參閱手冊頁 sigprocmask(2)pthread_sigmask(3)

3.5 版本中新增。

signal 模組中定義的變數有:

signal.SIG_DFL

這是兩種標準訊號處理選項之一;它將簡單地執行該訊號的預設功能。例如,在大多數系統上,SIGQUIT 的預設操作是轉儲核心並退出,而 SIGCHLD 的預設操作是簡單地忽略它。

signal.SIG_IGN

這是另一種標準訊號處理程式,它將簡單地忽略給定的訊號。

signal.SIGABRT

來自 abort(3) 的中止訊號。

signal.SIGALRM

來自 alarm(2) 的定時器訊號。

可用性: Unix。

signal.SIGBREAK

來自鍵盤的中斷(CTRL + BREAK)。

可用性: Windows。

signal.SIGBUS

匯流排錯誤(錯誤的記憶體訪問)。

可用性: Unix。

signal.SIGCHLD

子程序停止或終止。

可用性: Unix。

signal.SIGCLD

別名為 SIGCHLD

可用性: 非 macOS。

signal.SIGCONT

如果程序當前已停止,則繼續該程序

可用性: Unix。

signal.SIGFPE

浮點異常。例如,除以零。

另請參閱

當除法或模運算的第二個引數為零時,會引發 ZeroDivisionError

signal.SIGHUP

在控制終端上檢測到結束通話或控制程序死亡。

可用性: Unix。

signal.SIGILL

非法指令。

signal.SIGINT

來自鍵盤的中斷(CTRL + C)。

預設操作是引發 KeyboardInterrupt

signal.SIGKILL

殺死訊號。

它不能被捕獲、阻塞或忽略。

可用性: Unix。

signal.SIGPIPE

管道破裂:寫入沒有讀取器的管道。

預設操作是忽略訊號。

可用性: Unix。

signal.SIGSEGV

段錯誤:無效的記憶體引用。

signal.SIGSTKFLT

協處理器上的堆疊故障。Linux 核心不會引發此訊號:它只能在使用者空間中引發。

可用性: Linux。

在訊號可用的架構上。 有關更多資訊,請參閱 man 手冊 signal(7)

在 3.11 版本中新增。

signal.SIGTERM

終止訊號。

signal.SIGUSR1

使用者定義的訊號 1。

可用性: Unix。

signal.SIGUSR2

使用者定義的訊號 2。

可用性: Unix。

signal.SIGWINCH

視窗調整大小訊號。

可用性: Unix。

SIG*

所有訊號編號都是以符號方式定義的。 例如,結束通話訊號定義為 signal.SIGHUP;變數名稱與 C 程式中使用的名稱相同,如 <signal.h> 中所示。 ‘signal()’ 的 Unix man 手冊列出了現有訊號(在某些系統上是 signal(2),在其他系統上,列表在 signal(7) 中)。 請注意,並非所有系統都定義了相同的訊號名稱集;此模組僅定義系統定義的那些名稱。

signal.CTRL_C_EVENT

Ctrl+C 按鍵事件對應的訊號。 此訊號只能與 os.kill() 一起使用。

可用性: Windows。

在 3.2 版本中新增。

signal.CTRL_BREAK_EVENT

Ctrl+Break 按鍵事件對應的訊號。 此訊號只能與 os.kill() 一起使用。

可用性: Windows。

在 3.2 版本中新增。

signal.NSIG

比最高訊號編號大 1 的數字。使用 valid_signals() 獲取有效的訊號編號。

signal.ITIMER_REAL

以即時方式遞減間隔計時器,並在到期時傳遞 SIGALRM

signal.ITIMER_VIRTUAL

僅當程序正在執行時才遞減間隔計時器,並在到期時傳遞 SIGVTALRM。

signal.ITIMER_PROF

當程序執行以及系統代表程序執行時,都遞減間隔計時器。與 ITIMER_VIRTUAL 結合使用,此計時器通常用於分析應用程式在使用者空間和核心空間中花費的時間。到期時傳遞 SIGPROF。

signal.SIG_BLOCK

pthread_sigmask()how 引數的可能值,指示要阻止訊號。

在 3.3 版本中新增。

signal.SIG_UNBLOCK

pthread_sigmask()how 引數的可能值,指示要取消阻止訊號。

在 3.3 版本中新增。

signal.SIG_SETMASK

pthread_sigmask()how 引數的可能值,指示要替換訊號掩碼。

在 3.3 版本中新增。

signal 模組定義了一個異常:

exception signal.ItimerError

當底層 setitimer()getitimer() 實現發生錯誤時引發此異常。如果向 setitimer() 傳遞了無效的間隔定時器或負時間,則會引發此錯誤。此錯誤是 OSError 的子型別。

3.3 版本新增: 此錯誤以前是 IOError 的子型別,現在是 OSError 的別名。

signal 模組定義了以下函式

signal.alarm(time)

如果 time 為非零值,此函式請求在 time 秒後向程序傳送 SIGALRM 訊號。任何先前計劃的警報都會被取消(任何時候只能計劃一個警報)。返回的值是任何先前設定的警報被傳遞之前的秒數。如果 time 為零,則不會計劃警報,並且任何計劃的警報都會被取消。如果返回值是零,則當前沒有計劃任何警報。

可用性: Unix。

有關更多資訊,請參閱手冊頁 alarm(2)

signal.getsignal(signalnum)

返回訊號 signalnum 的當前訊號處理程式。 返回的值可以是可呼叫的 Python 物件,或者特殊值 signal.SIG_IGNsignal.SIG_DFLNone 之一。 這裡, signal.SIG_IGN 表示該訊號先前被忽略, signal.SIG_DFL 表示先前正在使用預設的訊號處理方式,而 None 表示先前的訊號處理程式不是從 Python 安裝的。

signal.strsignal(signalnum)

返回訊號 signalnum 的描述,例如 SIGINT 的 “中斷”。 如果 signalnum 沒有描述,則返回 None 。如果 signalnum 無效,則引發 ValueError

3.8 版本新增。

signal.valid_signals()

返回此平臺上有效的訊號編號集合。如果系統保留某些訊號以供內部使用,則此值可能小於 range(1, NSIG)

3.8 版本新增。

signal.pause()

使程序休眠直到收到訊號;然後將呼叫適當的處理程式。不返回任何值。

可用性: Unix。

有關更多資訊,請參閱手冊頁 signal(2)

另請參閱 sigwait(), sigwaitinfo(), sigtimedwait()sigpending()

signal.raise_signal(signum)

向呼叫程序傳送訊號。不返回任何值。

3.8 版本新增。

signal.pidfd_send_signal(pidfd, sig, siginfo=None, flags=0)

將訊號 sig 傳送到由檔案描述符 pidfd 引用的程序。Python 當前不支援 siginfo 引數;它必須是 Noneflags 引數是為將來的擴充套件提供的;目前沒有定義標誌值。

有關更多資訊,請參閱 pidfd_send_signal(2) 手冊頁。

可用性: Linux >= 5.1,Android >= build-time API 級別 31

3.9 版本新增。

signal.pthread_kill(thread_id, signalnum)

將訊號 signalnum 傳送到執行緒 thread_id,即與呼叫者在同一程序中的另一個執行緒。目標執行緒可以執行任何程式碼(Python 或非 Python)。但是,如果目標執行緒正在執行 Python 直譯器,則 Python 訊號處理程式將由主直譯器的主執行緒執行。因此,向特定 Python 執行緒傳送訊號的唯一目的是強制執行的系統呼叫以 InterruptedError 失敗。

使用 threading.get_ident()ident 屬性 threading.Thread 物件來獲取 thread_id 的合適值。

如果 signalnum 為 0,則不會發送訊號,但仍會執行錯誤檢查;這可用於檢查目標執行緒是否仍在執行。

引發帶有引數 thread_id, signalnum審計事件 signal.pthread_kill

可用性: Unix。

有關更多資訊,請參閱手冊頁 pthread_kill(3)

另請參閱 os.kill()

在 3.3 版本中新增。

signal.pthread_sigmask(how, mask)

獲取和/或更改呼叫執行緒的訊號掩碼。訊號掩碼是當前阻止向呼叫者傳遞的訊號集。以訊號集的形式返回舊的訊號掩碼。

呼叫的行為取決於 how 的值,如下所示。

  • SIG_BLOCK: 被阻止的訊號集是當前集和 mask 引數的並集。

  • SIG_UNBLOCK: 從當前被阻止的訊號集中刪除 mask 中的訊號。可以嘗試取消阻止未被阻止的訊號。

  • SIG_SETMASK: 被阻止的訊號集設定為 mask 引數。

mask 是一組訊號編號(例如 {signal.SIGINT, signal.SIGTERM})。使用 valid_signals() 獲取包含所有訊號的完整掩碼。

例如,signal.pthread_sigmask(signal.SIG_BLOCK, []) 讀取呼叫執行緒的訊號掩碼。

SIGKILLSIGSTOP 無法被阻止。

可用性: Unix。

有關更多資訊,請參閱手冊頁 sigprocmask(2)pthread_sigmask(3)

另請參閱 pause(), sigpending()sigwait()

在 3.3 版本中新增。

signal.setitimer(which, seconds, interval=0.0)

設定由 which 指定的給定間隔計時器(signal.ITIMER_REAL, signal.ITIMER_VIRTUALsignal.ITIMER_PROF 中的一個),在 seconds 秒後觸發(接受浮點數,與 alarm() 不同),之後每隔 interval 秒觸發一次(如果 interval 非零)。 可以透過將 seconds 設定為零來清除由 which 指定的間隔計時器。

當間隔計時器觸發時,會向程序傳送一個訊號。傳送的訊號取決於正在使用的計時器; signal.ITIMER_REAL 將傳遞 SIGALRMsignal.ITIMER_VIRTUAL 傳送 SIGVTALRM,並且 signal.ITIMER_PROF 將傳遞 SIGPROF

舊值以元組形式返回:(延遲, 間隔)。

嘗試傳遞無效的間隔計時器將導致 ItimerError

可用性: Unix。

signal.getitimer(which)

返回由 which 指定的給定間隔計時器的當前值。

可用性: Unix。

signal.set_wakeup_fd(fd, *, warn_on_full_buffer=True)

將喚醒檔案描述符設定為 fd。當收到訊號時,訊號編號將作為單個位元組寫入 fd。庫可以使用它來喚醒輪詢或選擇呼叫,從而允許完全處理訊號。

返回舊的喚醒 fd(如果未啟用檔案描述符喚醒,則返回 -1)。如果 fd 為 -1,則停用檔案描述符喚醒。如果不是 -1,則 fd 必須是非阻塞的。庫有責任在再次呼叫 poll 或 select 之前從 fd 中刪除任何位元組。

啟用執行緒時,此函式只能從 主直譯器的主執行緒 呼叫;嘗試從其他執行緒呼叫它會導致引發 ValueError 異常。

有兩種常見的方式來使用此函式。在這兩種方法中,你都使用 fd 在訊號到達時喚醒,但它們在如何確定 哪個哪些 訊號已到達方面有所不同。

在第一種方法中,我們從 fd 的緩衝區中讀取資料,位元組值會給你訊號編號。這很簡單,但在極少數情況下可能會遇到問題:通常 fd 的緩衝區空間有限,如果太多訊號到達太快,那麼緩衝區可能會被填滿,並且可能會丟失一些訊號。如果使用此方法,則應設定 warn_on_full_buffer=True,這至少會在訊號丟失時導致向 stderr 列印警告。

在第二種方法中,我們僅將喚醒 fd 用於喚醒,而忽略實際的位元組值。在這種情況下,我們關心的只是 fd 的緩衝區是空的還是非空的;完整的緩衝區根本不表示問題。如果使用此方法,則應設定 warn_on_full_buffer=False,以便使用者不會因虛假的警告訊息而感到困惑。

在 3.5 版本中更改: 在 Windows 上,該函式現在也支援套接字控制代碼。

在 3.7 版本中更改: 添加了 warn_on_full_buffer 引數。

signal.siginterrupt(signalnum, flag)

更改系統呼叫重啟行為:如果 flagFalse,則當被訊號 signalnum 中斷時,系統呼叫將重新啟動,否則系統呼叫將被中斷。不返回任何內容。

可用性: Unix。

有關更多資訊,請參閱手冊頁 siginterrupt(3)

請注意,使用 signal() 安裝訊號處理程式將透過隱式呼叫 siginterrupt() 併為給定訊號使用 true flag 值,將重啟行為重置為可中斷。

signal.signal(signalnum, handler)

將訊號 signalnum 的處理程式設定為函式 handlerhandler 可以是接受兩個引數的可呼叫 Python 物件(見下文),或者可以是特殊值 signal.SIG_IGNsignal.SIG_DFL。將返回之前的訊號處理程式(請參閱上面 getsignal() 的描述)。(有關更多資訊,請參閱 Unix 手冊頁 signal(2)。)

啟用執行緒時,此函式只能從 主直譯器的主執行緒 呼叫;嘗試從其他執行緒呼叫它會導致引發 ValueError 異常。

呼叫 handler 時會傳入兩個引數:訊號編號和當前堆疊幀(None 或幀物件;有關幀物件的描述,請參見 型別層次結構中的描述 或檢視 inspect 模組中的屬性描述)。

在 Windows 上,只能使用 SIGABRT, SIGFPE, SIGILL, SIGINT, SIGSEGV, SIGTERMSIGBREAK 呼叫 signal()。在任何其他情況下都會引發 ValueError。請注意,並非所有系統都定義了相同的訊號名稱集;如果訊號名稱未定義為 SIG* 模組級常量,則會引發 AttributeError

signal.sigpending()

檢查等待傳遞到呼叫執行緒的訊號集(即,在阻塞時引發的訊號)。返回待處理訊號集。

可用性: Unix。

有關更多資訊,請參閱手冊頁 sigpending(2)

另請參閱 pause()pthread_sigmask()sigwait()

在 3.3 版本中新增。

signal.sigwait(sigset)

暫停呼叫執行緒的執行,直到傳遞訊號集 sigset 中指定的某個訊號。 該函式接受該訊號(將其從待處理訊號列表中移除),並返回訊號編號。

可用性: Unix。

更多資訊請參閱手冊頁 sigwait(3)

另請參閱 pause()pthread_sigmask()sigpending()sigwaitinfo()sigtimedwait()

在 3.3 版本中新增。

signal.sigwaitinfo(sigset)

暫停呼叫執行緒的執行,直到傳遞訊號集 sigset 中指定的某個訊號。該函式接受該訊號並將其從待處理訊號列表中移除。如果 sigset 中的某個訊號已經處於呼叫執行緒的待處理狀態,該函式將立即返回有關該訊號的資訊。不會為傳遞的訊號呼叫訊號處理程式。如果該函式被 sigset 中未包含的訊號中斷,則會引發 InterruptedError 異常。

返回值是一個表示 siginfo_t 結構中包含的資料的物件,即: si_signo, si_code, si_errno, si_pid, si_uid, si_status, si_band

可用性: Unix。

更多資訊請參閱手冊頁 sigwaitinfo(2)

另請參閱 pause()sigwait()sigtimedwait()

在 3.3 版本中新增。

在 3.5 版本中更改: 如果被 sigset 中未包含的訊號中斷,並且訊號處理程式沒有引發異常,則該函式現在會重試(有關基本原理,請參閱 PEP 475)。

signal.sigtimedwait(sigset, timeout)

sigwaitinfo() 類似,但它接受一個額外的 timeout 引數來指定超時。 如果將 timeout 指定為 0,則執行輪詢。 如果發生超時,則返回 None

可用性: Unix。

更多資訊請參閱手冊頁 sigtimedwait(2)

另請參閱 pause()sigwait()sigwaitinfo()

在 3.3 版本中新增。

在 3.5 版本中更改: 如果被 sigset 中未包含的訊號中斷,並且訊號處理程式沒有引發異常,則該函式現在會使用重新計算的 timeout 值進行重試(有關基本原理,請參閱 PEP 475)。

示例

這是一個最小的示例程式。它使用 alarm() 函式來限制等待開啟檔案的時間; 如果該檔案用於可能未開啟的序列裝置,則這很有用,通常會導致 os.open() 無限期地掛起。 解決方案是在開啟檔案之前設定一個 5 秒的警報;如果操作時間過長,將傳送警報訊號,並且處理程式將引發異常。

import signal, os

def handler(signum, frame):
    signame = signal.Signals(signum).name
    print(f'Signal handler called with signal {signame} ({signum})')
    raise OSError("Couldn't open device!")

# Set the signal handler and a 5-second alarm
signal.signal(signal.SIGALRM, handler)
signal.alarm(5)

# This open() may hang indefinitely
fd = os.open('/dev/ttyS0', os.O_RDWR)

signal.alarm(0)          # Disable the alarm

關於 SIGPIPE 的注意事項

當標準輸出的接收者提前關閉時,將程式的輸出透過管道傳輸到 head(1) 等工具會導致向你的程序傳送 SIGPIPE 訊號。 這會導致類似 BrokenPipeError: [Errno 32] Broken pipe 這樣的異常。 要處理這種情況,請如下所示包裝你的入口點以捕獲此異常

import os
import sys

def main():
    try:
        # simulate large output (your code replaces this loop)
        for x in range(10000):
            print("y")
        # flush output here to force SIGPIPE to be triggered
        # while inside this try block.
        sys.stdout.flush()
    except BrokenPipeError:
        # Python flushes standard streams on exit; redirect remaining output
        # to devnull to avoid another BrokenPipeError at shutdown
        devnull = os.open(os.devnull, os.O_WRONLY)
        os.dup2(devnull, sys.stdout.fileno())
        sys.exit(1)  # Python exits with error code 1 on EPIPE

if __name__ == '__main__':
    main()

不要將 SIGPIPE 的處置設定為 SIG_DFL,以避免 BrokenPipeError。 這樣做會導致你的程式在程式仍在向其中寫入內容時,每當任何套接字連線中斷時意外退出。

關於訊號處理程式和異常的注意事項

如果訊號處理程式引發異常,則該異常將傳播到主執行緒,並且可能在任何 位元組碼 指令之後引發。 最值得注意的是, KeyboardInterrupt 可能在執行期間的任何時候出現。 大多數 Python 程式碼(包括標準庫)都無法對此進行可靠的防禦,因此, KeyboardInterrupt(或任何其他由訊號處理程式導致的異常)在極少數情況下可能會使程式處於意外狀態。

為了說明這個問題,請考慮以下程式碼

class SpamContext:
    def __init__(self):
        self.lock = threading.Lock()

    def __enter__(self):
        # If KeyboardInterrupt occurs here, everything is fine
        self.lock.acquire()
        # If KeyboardInterrupt occurs here, __exit__ will not be called
        ...
        # KeyboardInterrupt could occur just before the function returns

    def __exit__(self, exc_type, exc_val, exc_tb):
        ...
        self.lock.release()

對於許多程式,特別是那些只想在 KeyboardInterrupt 時退出的程式,這不是問題,但複雜或需要高可靠性的應用程式應避免從訊號處理程式引發異常。 它們還應避免捕獲 KeyboardInterrupt 作為正常關閉的手段。 相反,它們應安裝自己的 SIGINT 處理程式。 下面是一個 HTTP 伺服器的示例,它避免了 KeyboardInterrupt

import signal
import socket
from selectors import DefaultSelector, EVENT_READ
from http.server import HTTPServer, SimpleHTTPRequestHandler

interrupt_read, interrupt_write = socket.socketpair()

def handler(signum, frame):
    print('Signal handler called with signal', signum)
    interrupt_write.send(b'\0')
signal.signal(signal.SIGINT, handler)

def serve_forever(httpd):
    sel = DefaultSelector()
    sel.register(interrupt_read, EVENT_READ)
    sel.register(httpd, EVENT_READ)

    while True:
        for key, _ in sel.select():
            if key.fileobj == interrupt_read:
                interrupt_read.recv(1)
                return
            if key.fileobj == httpd:
                httpd.handle_request()

print("Serving on port 8000")
httpd = HTTPServer(('', 8000), SimpleHTTPRequestHandler)
serve_forever(httpd)
print("Shutdown...")