signal --- 設定非同步事件的處理函式

原始碼:Lib/signal.py


本模組提供於 Python 中使用訊號處理程式的機制。

一般規則

signal.signal() 函式允許定義自訂的處理程式,會在收到訊號時執行。我們安裝了少數的預設處理程式:SIGPIPE 會被忽略 (所以管道和 socket 上的寫入錯誤可以當作一般的 Python 例外報告),而 SIGINT(如果父行程沒有改變它的話)會被轉換成 KeyboardInterrupt 例外。

特定訊號的處理程式一旦被設定,就會一直被安裝,直到被明確地重設為止 (不管底層的實作為何,Python 皆模擬出 BSD 風格的介面),但 SIGCHLD 的處理程式除外,它會跟隨底層的實作。

在 WebAssembly 平台上,訊號是模擬出來的,故行為不同。有幾個函式和訊號在這些平台上是不可用的。

Python 訊號處理程式的執行

Python 訊號處理程式不會在低階(C 語言)訊號處理程式中執行。相反地,低階訊號處理程式會設定一個旗標,告訴虛擬機在稍後執行相對應的 Python 訊號處理程式(例如在下一個 bytecode 指令)。這會有一些後果:

  • 捕捉像 SIGFPESIGSEGV 這類由 C 程式碼中無效操作所引起的同步錯誤是沒有意義的。Python 將從訊號處理程式中回傳到 C 程式碼,而 C 程式碼很可能再次引發相同的訊號,導致 Python 明顯假當機 (hang)。從 Python 3.3 開始,你可以使用 faulthandler 模組來報告同步錯誤。

  • 純粹以 C 實作的長時間計算(例如在大量文字上的正規表示式比對)可能會不間斷地運行任意長度的時間而不考慮收到的任何訊號。當計算完成時,Python 訊號處理程式會被呼叫。

  • 如果處理程式引發例外,就會在主執行緒中「憑空」產生例外。請參閱下面的說明

訊號和執行緒

Python 訊號處理程式總是在主直譯器的主 Python 執行緒中執行,即使訊號是在另一個執行緒中接收到的。這意味著訊號不能用來做為執行緒間通訊的方式。你可以使用 threading 模組的同步原語 (synchronization primitive) 來代替。

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

模組內容

在 3.5 版的變更: 下面列出的訊號 (SIG*)、處理器(SIG_DFLSIG_IGN)和訊號遮罩 (sigmask)(SIG_BLOCKSIG_UNBLOCKSIG_SETMASK)的相關常數被轉換成 enumsSignalsHandlersSigmasks)。getsignal()pthread_sigmask()sigpending()sigwait() 函式會回傳可被人類閱讀的枚舉作為 Signals 物件。

訊號模組定義了三個枚舉:

class signal.Signals

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

在 3.5 版被加入.

class signal.Handlers

SIG_DFLSIG_IGN 常數的 enum.IntEnum 集合。

在 3.5 版被加入.

class signal.Sigmasks

SIG_BLOCKSIG_UNBLOCKSIG_SETMASK 常數的 enum.IntEnum 集合。

可用性: Unix.

更多資訊請見 sigprocmask(2)pthread_sigmask(3) 線上手冊。

在 3.5 版被加入.

signal 模組中定義的變數有:

signal.SIG_DFL

這是兩種標準訊號處理選項之一;它會簡單地執行訊號的預設功能。例如,在大多數系統上,SIGQUIT 的預設動作是轉儲 (dump) 核心並退出,而 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 的別名。

可用性: not macOS.

signal.SIGCONT

如果目前行程是被停止的,則繼續運行

可用性: Unix.

signal.SIGFPE

浮點運算例外。例如除以零。

也參考

ZeroDivisionError 會在除法或模運算 (modulo operation) 的第二個引數為零時引發。

signal.SIGHUP

偵測到控制終端機掛斷 (hangup) 或控制行程死亡。

可用性: Unix.

signal.SIGILL

非法指令。

signal.SIGINT

從鍵盤中斷 (CTRL + C)。

預設動作是引發 KeyboardInterrupt

signal.SIGKILL

殺死訊號。

它無法被捕捉、阻擋或忽略。

可用性: Unix.

signal.SIGPIPE

管道中斷 (broken pipe):寫到沒有讀取器 (reader) 的管道。

預設動作是忽略訊號。

可用性: Unix.

signal.SIGPROF

Profiling timer expired.

可用性: Unix.

signal.SIGQUIT

Terminal quit signal.

可用性: Unix.

signal.SIGSEGV

記憶體區段錯誤 (segmentation fault):無效記憶體參照。

signal.SIGSTOP

Stop executing (cannot be caught or ignored).

signal.SIGSTKFLT

輔助處理器 (coprocessor) 上的堆疊錯誤 (stack fault)。Linux 核心不會引發此訊號:它只能在使用者空間 (user space) 中引發。

可用性: Linux.

在訊號可用的架構上。請參閱 signal(7) 線上手冊以取得更多資訊。

在 3.11 版被加入.

signal.SIGTERM

終止訊號。

signal.SIGUSR1

使用者定義訊號 1。

可用性: Unix.

signal.SIGUSR2

使用者定義訊號 2。

可用性: Unix.

signal.SIGVTALRM

Virtual timer expired.

可用性: Unix.

signal.SIGWINCH

視窗調整大小訊號。

可用性: Unix.

SIG*

All the signal numbers are defined symbolically. For example, the hangup signal is defined as signal.SIGHUP; the variable names are identical to the names used in C programs, as found in <signal.h>. The Unix man page for 'signal' lists the existing signals (on some systems this is signal(2), on others the list is in signal(7)). Note that not all systems define the same set of signal names; only those names defined by the system are defined by this module.

signal.CTRL_C_EVENT

Ctrl+C 擊鍵 (keystroke) 事件相對應的訊號。此訊號只能與 os.kill() 搭配使用。

可用性: Windows.

在 3.2 版被加入.

signal.CTRL_BREAK_EVENT

Ctrl+Break 擊鍵事件相對應的訊號。此訊號只能與 os.kill() 搭配使用。

可用性: Windows.

在 3.2 版被加入.

signal.NSIG

比最高編號訊號的編號多一。使用 valid_signals() 來取得有效的訊號編號。

signal.ITIMER_REAL

即時減少間隔計時器 (interval timer),並在到期時送出 SIGALRM

signal.ITIMER_VIRTUAL

僅在執行行程時遞減間隔計時器,並在到期時傳送 SIGVTALRM。

signal.ITIMER_PROF

當行程執行或系統代表行程執行時,都會減少間隔計時器。與 ITIMER_VIRTUAL 配合,這個計時器通常用來分析 (profile) 應用程式在使用者空間與核心空間 (kernel space) 所花費的時間。SIGPROF 會在到期時傳送。

signal.SIG_BLOCK

pthread_sigmask()how 參數的可能值,表示訊號將被阻檔。

在 3.3 版被加入.

signal.SIG_UNBLOCK

pthread_sigmask()how 參數的可能值,表示訊號將被解除阻檔。

在 3.3 版被加入.

signal.SIG_SETMASK

pthread_sigmask()how 參數的可能值,表示訊號遮罩 (signal mask) 要被取代。

在 3.3 版被加入.

signal 模組定義了一個例外:

exception signal.ItimerError

setitimer()getitimer() 底層實作發生錯誤時引發訊號。如果傳給 setitimer() 的是無效的間隔計時器或負數時間,則預期會發生此錯誤。這個錯誤是 OSError 的子型別。

在 3.3 版被加入: 此錯誤過去是 IOError 的子型別,現在是 OSError 的別名。

signal 模組定義了下列函式:

signal.alarm(time)

如果 time 非零,則此函式會要求在 time 秒後傳送 SIGALRM 訊號給該行程。任何先前排程 (scheduled) 的警報都會被取消(任何時候都只能排程一個警報)。回傳值是先前設定的警報原本再等多久就會被傳送的秒數。如果 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 的 "Interrupt"。如果 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 level 31

在 3.9 版被加入.

signal.pthread_kill(thread_id, signalnum)

將訊號 signalnum 傳送給與呼叫者在同一行程中的另一個執行緒 thread_id。目標執行緒能執行任何程式碼(無論為 Python 與否)。但是,如果目標執行緒正執行 Python 直譯器,Python 訊號處理程式將會由主直譯器的主執行緒來執行。因此,向特定 Python 執行緒發送訊號的唯一意義是強制執行中的系統呼叫以 InterruptedError 方式失敗。

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

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

引發一個附帶引數 thread_idsignalnum稽核事件 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_REALsignal.ITIMER_VIRTUALsignal.ITIMER_PROF 之一)並在*seconds*(接受浮點數,與 alarm() 不同)之後啟動,在之後的每 interval 秒啟動一次(如果 interval 非零)。which 指定的間隔計時器可透過將 seconds 設定為零來清除它。

當間隔計時器啟動時,一個訊號會被傳送給行程。傳送的訊號取決於使用的計時器;signal.ITIMER_REAL 會傳送 SIGALRMsignal.ITIMER_VIRTUAL 會傳送 SIGVTALRM,而 signal.ITIMER_PROF 會傳送 SIGPROF

舊值會以一個元組回傳:(delay, interval)。

嘗試傳入無效的間隔計時器會導致 ItimerError

可用性: Unix.

signal.getitimer(which)

回傳由 which 指定之間隔計時器的目前值。

可用性: Unix.

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

設定喚醒檔案描述器為 fd。當接收到其處理程式已有被註冊的訊號時,訊號編號會以單一位元組寫入 fd。如果你沒有為你在意的訊號註冊處理程式,就不會寫入到喚醒 fd。這可被函式庫用來喚醒輪詢 (wakeup a poll) 或 select 呼叫,讓訊號得以完全處理。

回傳舊的喚醒 fd(如果檔案描述器喚醒未啟用,則回傳 -1)。如果 fd 為 -1,則會停用檔案描述器喚醒。如果不是 -1,fd 必須是非阻塞的。在再次呼叫輪詢或 select 之前,由函式庫來決定是否移除 fd 中的任何位元組。

當啟用執行緒時,這個函式只能從主直譯器的主執行緒來呼叫;嘗試從其他執行緒呼叫它將會引發 ValueError 例外。

使用這個函式有兩種常見的方式。在這兩種方法中,當訊號抵達時,你都要使用 fd 來喚醒,但它們的不同之處在於如何判斷哪個或哪些訊號有抵達。

在第一種方法中,我們從 fd 的緩衝區中讀取資料,而位元組值則提供訊號編號。這個方法很簡單,但在少數情況下可能會遇到問題:一般來說,fd 的緩衝區空間有限,如果太多訊號來得太快,那麼緩衝區可能會滿,而有些訊號可能會遺失。如果你使用這種方法,那麼你應該設定 warn_on_full_buffer=True,這至少會在訊號丟失時將警告印出到 stderr。

在第二種方法中,我們只會將喚醒 fd 用於喚醒,而忽略實際的位元組值。在這種情況下,我們只在乎 fd 的緩衝區是空或非空;即便緩衝區滿了也不代表有問題。如果你使用這種方法,那麼你應該設定 warn_on_full_buffer=False,這樣你的使用者就不會被虛假的警告訊息所混淆。

在 3.5 版的變更: 在 Windows 上,此功能現在也支援 socket 處理程式。

在 3.7 版的變更: 新增 warn_on_full_buffer 參數。

signal.siginterrupt(signalnum, flag)

改變系統呼叫重新啟動的行為:如果 flagFalse,系統呼叫會在被訊號 signalnum 中斷時重新啟動,否則系統呼叫會被中斷。不會回傳任何東西。

可用性: Unix.

更多資訊請見 siginterrupt(3) 線上手冊。

請注意,使用 signal() 安裝訊號處理程式,會透過隱式呼叫 siginterrupt() 來將重新啟動的行為重設為可中斷,且指定訊號的 flag 值為 true。

signal.signal(signalnum, handler)

將訊號 signalnum 的處理程式設定為函式 handlerhandler 可以是帶兩個引數的可呼叫 Python 物件(見下面),或是特殊值 signal.SIG_IGNsignal.SIG_DFL 之一。先前的訊號處理程式將會被回傳(請參閱上面 getsignal() 的說明)。(更多資訊請參閱 Unix 線上手冊 signal(2))。

當啟用執行緒時,這個函式只能從主直譯器的主執行緒來呼叫;嘗試從其他執行緒呼叫它將會引發 ValueError 例外。

handler 被呼叫時有兩個引數:訊號編號和目前的堆疊 frame(None 或一個 frame 物件;關於 frame 物件的描述,請參閱型別階層中的描述inspect 模組中的屬性描述)。

在 Windows 上,signal() 只能在使用 SIGABRTSIGFPESIGILLSIGINTSIGSEGVSIGTERMSIGBREAK 時呼叫。在其他情況下會引發 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

The return value is an object representing the data contained in the siginfo_t structure, namely: 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!")

# 設定訊號處理程式與五秒警報
signal.signal(signal.SIGALRM, handler)
signal.alarm(5)

# 這個 open() 可能無限期地被擱置
fd = os.open('/dev/ttyS0', os.O_RDWR)

signal.alarm(0)          # 停用警報

關於 SIGPIPE 的說明

將程式的輸出管道化到 head(1) 之類的工具,會在你的行程的標準輸出接收器提早關閉時,導致 SIGPIPE 訊號傳送給你的行程。這會導致類似 BrokenPipeError: [Errno 32] Broken pipe 的例外。要處理這種情況,請將你的進入點包裝成如下的樣子來捕捉這個例外:

import os
import sys

def main():
    try:
        # 模擬大量輸出(你的程式取代此迴圈)
        for x in range(10000):
            print("y")
        # 在這裡清除輸出以強制 SIGPIPE 在這個 try 區塊
        # 中被觸發
        sys.stdout.flush()
    except BrokenPipeError:
        # Python 在退出時清除標準串流;為剩下的輸出重新導向
        # 至 devnull 來避免關閉時的 BrokenPipeError
        devnull = os.open(os.devnull, os.O_WRONLY)
        os.dup2(devnull, sys.stdout.fileno())
        sys.exit(1)  # Python 在 EPIPE 時以錯誤碼 1 退出

if __name__ == '__main__':
    main()

不要為了避免 BrokenPipeError 而將 SIGPIPE 之處置 (disposition) 設定為 SIG_DFL。這樣做會導致你的程式在寫入任何 socket 連線時被中斷而意外退出。

訊號處理程式與例外的說明

如果訊號處理程式產生例外,例外會傳送到主執行緒並可能在任何 bytecode 指令之後發生。最值得注意的是,KeyboardInterrupt 可能在執行過程中的任何時候出現。大多數 Python 程式碼,包括標準函式庫,都無法避免這種情況,因此 KeyboardInterrupt(或任何其他由訊號處理程式產生的例外)可能會在罕見的情況下使程式處於預期之外的狀態。

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

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

    def __enter__(self):
        # 如果 KeyboardInterrupt 在此發生則一切正常
        self.lock.acquire()
        # 如果 KeyboardInterrupt 在此發生,__exit__ 將不會被呼叫
        ...
        # KeyboardInterrupt 可能在函式回傳之前發生

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

對許多程式來說,尤其是那些只想在 KeyboardInterrupt 時退出的程式,這並不是問題,但是對於複雜或需要高可靠性的應用程式來說,應該避免從訊號處理程式產生例外。它們也應該避免將捕獲 KeyboardInterrupt 作為一種優雅關閉 (gracefully shutting down) 的方式。相反地,它們應該安裝自己的 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...")