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
) 相關常量被轉換為 列舉
(分別為 Signals
、Handlers
和 Sigmasks
)。 getsignal()
、pthread_sigmask()
、sigpending()
和 sigwait()
函式返回可讀的 列舉
作為 Signals
物件。
signal 模組定義了三個列舉
- class signal.Signals¶
SIG* 常量和 CTRL_* 常量的
enum.IntEnum
集合。3.5 版本中新增。
- class signal.Handlers¶
常量
SIG_DFL
和SIG_IGN
的enum.IntEnum
集合。3.5 版本中新增。
- class signal.Sigmasks¶
常量
SIG_BLOCK
、SIG_UNBLOCK
和SIG_SETMASK
的enum.IntEnum
集合。可用性: 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.SIGSTKFLT¶
協處理器上的堆疊故障。Linux 核心不會引發此訊號:它只能在使用者空間中引發。
在 3.11 版本中新增。
- signal.SIGTERM¶
終止訊號。
- SIG*
所有訊號編號都是以符號方式定義的。 例如,結束通話訊號定義為
signal.SIGHUP
;變數名稱與 C 程式中使用的名稱相同,如<signal.h>
中所示。 ‘signal()
’ 的 Unix man 手冊列出了現有訊號(在某些系統上是 signal(2),在其他系統上,列表在 signal(7) 中)。 請注意,並非所有系統都定義了相同的訊號名稱集;此模組僅定義系統定義的那些名稱。
- signal.NSIG¶
比最高訊號編號大 1 的數字。使用
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()¶
使程序休眠直到收到訊號;然後將呼叫適當的處理程式。不返回任何值。
另請參閱
sigwait()
,sigwaitinfo()
,sigtimedwait()
和sigpending()
。
- signal.raise_signal(signum)¶
向呼叫程序傳送訊號。不返回任何值。
3.8 版本新增。
- signal.pidfd_send_signal(pidfd, sig, siginfo=None, flags=0)¶
將訊號 sig 傳送到由檔案描述符 pidfd 引用的程序。Python 當前不支援 siginfo 引數;它必須是
None
。flags 引數是為將來的擴充套件提供的;目前沒有定義標誌值。有關更多資訊,請參閱 pidfd_send_signal(2) 手冊頁。
可用性: Linux >= 5.1,Android >=
build-time
API 級別 313.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, [])
讀取呼叫執行緒的訊號掩碼。SIGKILL
和SIGSTOP
無法被阻止。可用性: 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(如果未啟用檔案描述符喚醒,則返回 -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()
安裝訊號處理程式將透過隱式呼叫siginterrupt()
併為給定訊號使用 true flag 值,將重啟行為重置為可中斷。
- signal.signal(signalnum, handler)¶
將訊號 signalnum 的處理程式設定為函式 handler。handler 可以是接受兩個引數的可呼叫 Python 物件(見下文),或者可以是特殊值
signal.SIG_IGN
或signal.SIG_DFL
。將返回之前的訊號處理程式(請參閱上面getsignal()
的描述)。(有關更多資訊,請參閱 Unix 手冊頁 signal(2)。)啟用執行緒時,此函式只能從 主直譯器的主執行緒 呼叫;嘗試從其他執行緒呼叫它會導致引發
ValueError
異常。呼叫 handler 時會傳入兩個引數:訊號編號和當前堆疊幀(
None
或幀物件;有關幀物件的描述,請參見 型別層次結構中的描述 或檢視inspect
模組中的屬性描述)。在 Windows 上,只能使用
SIGABRT
,SIGFPE
,SIGILL
,SIGINT
,SIGSEGV
,SIGTERM
或SIGBREAK
呼叫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...")