signal
— 為非同步事件設定處理程式¶
原始碼: Lib/signal.py
此模組提供了在 Python 中使用訊號處理程式的機制。
通用規則¶
signal.signal()
函式允許定義自定義處理程式,以便在接收到訊號時執行。安裝了少量預設處理程式:SIGPIPE
被忽略(因此管道和套接字上的寫入錯誤可以作為普通的 Python 異常報告),如果父程序沒有更改 SIGINT
,則它會被轉換為 KeyboardInterrupt
異常。
一旦為特定訊號設定了處理程式,它就會一直保持安裝狀態,直到顯式重置(Python 模擬 BSD 風格的介面,無論底層實現如何),但 SIGCHLD
的處理程式除外,它遵循底層實現。
在 WebAssembly 平臺上,訊號是模擬的,因此行為有所不同。這些平臺上不提供某些函式和訊號。
Python 訊號處理程式的執行¶
Python 訊號處理程式不在低階 (C) 訊號處理程式內部執行。相反,低階訊號處理程式會設定一個標誌,告訴 虛擬機器 在稍後(例如在下一個 位元組碼 指令)執行相應的 Python 訊號處理程式。這會帶來以下後果:
捕獲由 C 程式碼中的無效操作引起的同步錯誤(如
SIGFPE
或SIGSEGV
)意義不大。Python 將從訊號處理程式返回到 C 程式碼,C 程式碼很可能會再次引發相同的訊號,導致 Python 似乎掛起。從 Python 3.3 開始,你可以使用faulthandler
模組報告同步錯誤。純粹用 C 實現的長時間執行的計算(例如對大量文字進行正則表示式匹配)可能會不間斷地執行任意長時間,無論接收到任何訊號。Python 訊號處理程式將在計算完成後被呼叫。
如果處理程式引發異常,它將在主執行緒中“憑空”引發。有關討論,請參閱下面的註釋。
訊號與執行緒¶
Python 訊號處理程式始終在主直譯器的主 Python 執行緒中執行,即使訊號是在另一個執行緒中接收到的。這意味著訊號不能用作執行緒間通訊的手段。你可以改用 threading
模組中的同步原語。
此外,只有主直譯器的主執行緒才能設定新的訊號處理程式。
模組內容¶
版本 3.5 中已更改: 下面列出的訊號 (SIG*)、處理程式 (SIG_DFL
, SIG_IGN
) 和訊號掩碼 (SIG_BLOCK
, SIG_UNBLOCK
, SIG_SETMASK
) 相關常量已轉換為 enums
(分別為 Signals
、Handlers
和 Sigmasks
)。getsignal()
、pthread_sigmask()
、sigpending()
和 sigwait()
函式返回人類可讀的 enums
作為 Signals
物件。
訊號模組定義了三個列舉
- class signal.Signals¶
enum.IntEnum
集合,包含 SIG* 常量和 CTRL_* 常量。在 3.5 版本加入。
- class signal.Handlers¶
enum.IntEnum
集合,包含常量SIG_DFL
和SIG_IGN
。在 3.5 版本加入。
- class signal.Sigmasks¶
enum.IntEnum
集合,包含常量SIG_BLOCK
、SIG_UNBLOCK
和SIG_SETMASK
。可用性: Unix。
有關更多資訊,請參閱手冊頁 sigprocmask(2) 和 pthread_sigmask(3)。
在 3.5 版本加入。
signal
模組中定義的變數是
- signal.SIG_DFL¶
這是兩個標準訊號處理選項之一;它將簡單地執行訊號的預設功能。例如,在大多數系統上,
SIGQUIT
的預設操作是轉儲核心並退出,而SIGCHLD
的預設操作是簡單地忽略它。
- signal.SIG_IGN¶
這是另一個標準訊號處理程式,它將簡單地忽略給定的訊號。
- signal.SIGFPE¶
浮點異常。例如,除以零。
參見
當除法或模運算的第二個引數為零時,會引發
ZeroDivisionError
。
- signal.SIGILL¶
非法指令。
- signal.SIGINT¶
來自鍵盤的中斷 (CTRL + C)。
預設操作是引發
KeyboardInterrupt
。
- signal.SIGSEGV¶
段錯誤:無效記憶體引用。
- signal.SIGSTOP¶
停止執行(無法捕獲或忽略)。
- signal.SIGSTKFLT¶
協處理器上的棧錯誤。Linux 核心不會引發此訊號:它只能在使用者空間中引發。
在 3.11 版本中新增。
- signal.SIGTERM¶
終止訊號。
- SIG*
所有訊號編號都以符號形式定義。例如,掛起訊號定義為
signal.SIGHUP
;變數名與 C 程式中使用的名稱相同,如<signal.h>
中所示。Unix 手冊頁 'signal' 列出了現有訊號(在某些系統上是 signal(2),在其他系統上列表在 signal(7) 中)。請注意,並非所有系統都定義相同的訊號名稱集;此模組僅定義系統定義的那些名稱。
- signal.NSIG¶
比最高訊號編號多一。使用
valid_signals()
獲取有效訊號編號。
- 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
的子型別。
signal
模組定義了以下函式
- signal.alarm(time)¶
如果 time 非零,則此函式請求在 time 秒後向程序傳送
SIGALRM
訊號。任何先前計劃的警報都將被取消(一次只能計劃一個警報)。返回值是任何先前設定的警報被髮送之前的秒數。如果 time 為零,則不計劃警報,並且任何已計劃的警報都將被取消。如果返回值為零,則當前沒有計劃警報。
- signal.getsignal(signalnum)¶
返回訊號 signalnum 的當前訊號處理程式。返回值可能是一個可呼叫的 Python 物件,或者特殊值
signal.SIG_IGN
、signal.SIG_DFL
或None
。這裡,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()¶
使程序休眠直到接收到訊號;然後將呼叫相應的處理程式。不返回任何內容。
- signal.raise_signal(signum)¶
向呼叫程序傳送訊號。不返回任何內容。
在 3.8 版本加入。
- signal.pidfd_send_signal(pidfd, sig, siginfo=None, flags=0)¶
向檔案描述符 pidfd 指代的程序傳送訊號 sig。Python 當前不支援 siginfo 引數;它必須為
None
。flags 引數用於未來的擴充套件;目前沒有定義任何標誌值。有關更多資訊,請參閱 pidfd_send_signal(2) 手冊頁。
在 3.9 版本中新增。
- signal.pthread_kill(thread_id, signalnum)¶
向執行緒 thread_id 傳送訊號 signalnum,該執行緒是呼叫者程序中的另一個執行緒。目標執行緒可以執行任何程式碼(Python 或非 Python)。但是,如果目標執行緒正在執行 Python 直譯器,則 Python 訊號處理程式將由主直譯器的主執行緒執行。因此,向特定 Python 執行緒傳送訊號的唯一目的是強制執行的系統呼叫以
InterruptedError
失敗。使用
threading.get_ident()
或threading.Thread
物件的ident
屬性來獲取適合 thread_id 的值。如果 signalnum 為 0,則不傳送訊號,但仍執行錯誤檢查;這可用於檢查目標執行緒是否仍在執行。
引發 審計事件
signal.pthread_kill
,引數為thread_id
、signalnum
。可用性: 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, [])
讀取呼叫執行緒的訊號掩碼。可用性: Unix。
有關更多資訊,請參閱手冊頁 sigprocmask(2) 和 pthread_sigmask(3)。
另請參閱
pause()
、sigpending()
和sigwait()
。在 3.3 版本加入。
- signal.setitimer(which, seconds, interval=0.0)¶
設定由 which 指定的給定間隔計時器(
signal.ITIMER_REAL
、signal.ITIMER_VIRTUAL
或signal.ITIMER_PROF
之一),使其在 seconds 秒後觸發(接受浮點數,與alarm()
不同),之後每隔 interval 秒觸發一次(如果 interval 非零)。透過將 seconds 設定為零,可以清除由 which 指定的間隔計時器。當間隔計時器觸發時,會向程序傳送一個訊號。傳送的訊號取決於所使用的計時器;
signal.ITIMER_REAL
將傳送SIGALRM
,signal.ITIMER_VIRTUAL
傳送SIGVTALRM
,而signal.ITIMER_PROF
將傳送SIGPROF
。舊值作為元組返回:(延遲,間隔)。
嘗試傳遞無效的間隔計時器將導致
ItimerError
。可用性: Unix。
- signal.set_wakeup_fd(fd, *, warn_on_full_buffer=True)¶
將喚醒檔案描述符設定為 fd。當程式註冊的訊號處理程式收到訊號時,訊號編號將作為單個位元組寫入 fd。如果未為關心的訊號註冊訊號處理程式,則不會向喚醒 fd 寫入任何內容。這可用於庫喚醒 poll 或 select 呼叫,從而允許完全處理訊號。
返回舊的喚醒 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)¶
更改系統呼叫重啟行為:如果 flag 為
False
,則系統呼叫在被訊號 signalnum 中斷時將重啟,否則系統呼叫將被中斷。不返回任何內容。可用性: Unix。
有關更多資訊,請參閱手冊頁 siginterrupt(3)。
請注意,使用
signal()
安裝訊號處理程式將透過隱式呼叫帶有給定訊號的 true flag 值的siginterrupt()
將重啟行為重置為可中斷。
- signal.signal(signalnum, handler)¶
將訊號 signalnum 的處理程式設定為函式 handler。handler 可以是接受兩個引數的可呼叫 Python 物件(見下文),或者是特殊值
signal.SIG_IGN
或signal.SIG_DFL
。將返回先前的訊號處理程式(參見上面getsignal()
的描述)。(有關更多資訊,請參見 Unix 手冊頁 signal(2)。)啟用執行緒時,此函式只能從主直譯器的主執行緒呼叫;嘗試從其他執行緒呼叫它將導致引發
ValueError
異常。handler 使用兩個引數呼叫:訊號編號和當前堆疊幀(
None
或幀物件;有關幀物件的描述,請參閱型別層次結構中的描述或inspect
模組中的屬性描述)。在 Windows 上,
signal()
只能與SIGABRT
、SIGFPE
、SIGILL
、SIGINT
、SIGSEGV
、SIGTERM
或SIGBREAK
一起使用。在任何其他情況下,都將引發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
處理程式。下面是一個避免 KeyboardInterrupt
的 HTTP 伺服器示例
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...")