signal
--- 設定非同步事件的處理函式¶
原始碼:Lib/signal.py
本模組提供於 Python 中使用訊號處理程式的機制。
一般規則¶
signal.signal()
函式允許定義自訂的處理程式,會在收到訊號時執行。我們安裝了少數的預設處理程式:SIGPIPE
會被忽略 (所以管道和 socket 上的寫入錯誤可以當作一般的 Python 例外報告),而 SIGINT
(如果父行程沒有改變它的話)會被轉換成 KeyboardInterrupt
例外。
特定訊號的處理程式一旦被設定,就會一直被安裝,直到被明確地重設為止 (不管底層的實作為何,Python 皆模擬出 BSD 風格的介面),但 SIGCHLD
的處理程式除外,它會跟隨底層的實作。
在 WebAssembly 平台上,訊號是模擬出來的,故行為不同。有幾個函式和訊號在這些平台上是不可用的。
Python 訊號處理程式的執行¶
Python 訊號處理程式不會在低階(C 語言)訊號處理程式中執行。相反地,低階訊號處理程式會設定一個旗標,告訴虛擬機在稍後執行相對應的 Python 訊號處理程式(例如在下一個 bytecode 指令)。這會有一些後果:
捕捉像
SIGFPE
或SIGSEGV
這類由 C 程式碼中無效操作所引起的同步錯誤是沒有意義的。Python 將從訊號處理程式中回傳到 C 程式碼,而 C 程式碼很可能再次引發相同的訊號,導致 Python 明顯假當機 (hang)。從 Python 3.3 開始,你可以使用faulthandler
模組來報告同步錯誤。純粹以 C 實作的長時間計算(例如在大量文字上的正規表示式比對)可能會不間斷地運行任意長度的時間而不考慮收到的任何訊號。當計算完成時,Python 訊號處理程式會被呼叫。
如果處理程式引發例外,就會在主執行緒中「憑空」產生例外。請參閱下面的說明。
訊號和執行緒¶
Python 訊號處理程式總是在主直譯器的主 Python 執行緒中執行,即使訊號是在另一個執行緒中接收到的。這意味著訊號不能用來做為執行緒間通訊的方式。你可以使用 threading
模組的同步原語 (synchronization primitive) 來代替。
此外,只有主直譯器的主執行緒才被允許設定新的訊號處理程式。
模組內容¶
在 3.5 版的變更: 下面列出的訊號 (SIG*)、處理器(SIG_DFL
、SIG_IGN
)和訊號遮罩 (sigmask)(SIG_BLOCK
、SIG_UNBLOCK
、SIG_SETMASK
)的相關常數被轉換成 enums
(Signals
、Handlers
和 Sigmasks
)。getsignal()
、pthread_sigmask()
、sigpending()
和 sigwait()
函式會回傳可被人類閱讀的枚舉
作為 Signals
物件。
訊號模組定義了三個枚舉:
- 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
的預設動作是轉儲 (dump) 核心並退出,而SIGCHLD
的預設動作是直接忽略。
- signal.SIG_IGN¶
這是另一個標準的訊號處理程式,會直接忽略給定的訊號。
- signal.SIGFPE¶
浮點運算例外。例如除以零。
也參考
ZeroDivisionError
會在除法或模運算 (modulo operation) 的第二個引數為零時引發。
- signal.SIGILL¶
非法指令。
- signal.SIGINT¶
從鍵盤中斷 (CTRL + C)。
預設動作是引發
KeyboardInterrupt
。
- signal.SIGSEGV¶
記憶體區段錯誤 (segmentation fault):無效記憶體參照。
- signal.SIGSTKFLT¶
輔助處理器 (coprocessor) 上的堆疊錯誤 (stack fault)。Linux 核心不會引發此訊號:它只能在使用者空間 (user space) 中引發。
在 3.11 版被加入.
- signal.SIGTERM¶
終止訊號。
- SIG*
所有的訊號編號都是以符號定義的。例如,掛斷訊號被定義為
signal.SIGHUP
;變數名稱與 C 程式中使用的名稱相同,可在<signal.h>
中找到。Unix 線上手冊 'signal()
' 列出了現有的訊號(在某些系統上是 signal(2),在其他系統上是在 signal(7) 中)。請注意,並非所有系統都會定義同一套訊號名稱;只有那些由系統所定義的名稱才會由這個模組定義。
- signal.CTRL_C_EVENT¶
與 Ctrl+C 擊鍵 (keystroke) 事件相對應的訊號。此訊號只能與
os.kill()
搭配使用。適用: Windows.
在 3.2 版被加入.
- signal.NSIG¶
比最高編號訊號的編號多一。使用
valid_signals()
來取得有效的訊號編號。
- 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
的子型別。
signal
模組定義了下列函式:
- signal.alarm(time)¶
如果 time 非零,則此函式會要求在 time 秒後傳送
SIGALRM
訊號給該行程。任何先前排程 (scheduled) 的警報都會被取消(任何時候都只能排程一個警報)。回傳值是先前設定的警報原本再等多久就會被傳送的秒數。如果 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
的 "Interrupt"。如果 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)¶
傳送訊號 sig 到檔案描述器 pidfd 所指的行程。Python 目前不支援 siginfo 參數;它必須是
None
。flags 引數是提供給未來的擴充;目前沒有定義旗標值。更多資訊請見 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_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 非零)。which 指定的間隔計時器可透過將 seconds 設定為零來清除它。當間隔計時器啟動時,一個訊號會被傳送給行程。傳送的訊號取決於使用的計時器;
signal.ITIMER_REAL
會傳送SIGALRM
,signal.ITIMER_VIRTUAL
會傳送SIGVTALRM
,而signal.ITIMER_PROF
會傳送SIGPROF
。舊值會以一個元組回傳:(delay, interval)。
嘗試傳入無效的間隔計時器會導致
ItimerError
。適用: Unix.
- signal.set_wakeup_fd(fd, *, warn_on_full_buffer=True)¶
設定喚醒檔案描述器為 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)¶
改變系統呼叫重新啟動的行為:如果 flag 是
False
,系統呼叫會在被訊號 signalnum 中斷時重新啟動,否則系統呼叫會被中斷。不會回傳任何東西。適用: Unix.
更多資訊請見 siginterrupt(3) 線上手冊。
請注意,使用
signal()
安裝訊號處理程式,會透過隱式呼叫siginterrupt()
來將重新啟動的行為重設為可中斷,且指定訊號的 flag 值為 true。
- signal.signal(signalnum, handler)¶
將訊號 signalnum 的處理程式設定為函式 handler。handler 可以是帶兩個引數的可呼叫 Python 物件(見下面),或是特殊值
signal.SIG_IGN
或signal.SIG_DFL
之一。先前的訊號處理程式將會被回傳(請參閱上面getsignal()
的說明)。(更多資訊請參閱 Unix 線上手冊 :signal(2))。當啟用執行緒時,這個函式只能從主直譯器的主執行緒來呼叫;嘗試從其他執行緒呼叫它將會引發
ValueError
例外。handler 被呼叫時有兩個引數:訊號編號和目前的堆疊 frame(
None
或一個 frame 物件;關於 frame 物件的描述,請參閱型別階層中的描述或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!")
# 設定訊號處理程式與五秒警報
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...")