はじめに
本記事は、「手間いらずにPythonでシグナルをExceptionっぽく簡単に扱えるようにしたい」という方向けです。
バリバリ本気でシグナルを使いこなしたい方は公式ドキュメント(signal)を参考にしてください。
TL;DR
ここに使い方が書いてます。
ソースコードはここに記載の内容をコピペしてください。
(220320追記) 内容一部修正してGitHubに公開しました.
記事の内容とちょっぴり違いますがよければこちらもご利用ください.
https://github.com/nadu-festival/sigasexc
モチベーション
「シグナルで割り込みかけられる処理を、例外処理っぽく書けたらめっちゃ楽なのになぁ」という考えから、シグナル処理を例外っぽく書けるモジュールを作りました。
通常シグナルは使わないことの方が圧倒的に多いですが、使いたいときになって調べて、Pythonっぽくない書き方を要求されるので、使うのに少し困惑します。
対して、自前でプロセスを立ち上げて協調的な処理をしたりしたり、他言語で作ったプロセスと協調的に処理をしたい場合、シグナルは比較的便利だったりします。例えば、CとPythonで協調的に処理をしたい場合に、プロセス間通信はPOSIX IPCやsocketを使えば大概はできますが、異常発生時の割り込み処理や、異常発生に対して強制終了したい場合等はシグナルで書いた方が圧倒的に使いやすいです。
通常のシグナルの使い方
まず、通常のシグナルの使い方をざっと復習します。
Pythonのバージョンは3.8.5を利用します。
signal.alarmを使う例
サンプルとして用いるコードは、公式ドキュメント(signal)に記載のサンプルを少し改変したものです。
import signal
import time
# シグナルハンドラを定義
def handler(signum, frame):
print(f'handlerが呼び出されました(signum={signum})')
raise OSError("シグナルの割り込みがありました")
# signal.SIGALRMのシグナルハンドラを登録
signal.signal(signal.SIGALRM, handler)
# 5秒後にsignal.SIGALRMで割り込み予約
signal.alarm(5)
# 時間のかかる処理／割り込みによって中断したい処理
while True:
print("hellow.")
time.sleep(0.7)
# signal.SIGALRMのアラーム設定を解除
signal.alarm(0)
上記のサンプルでは、signal.alarm関数を用いて指定秒数後にsignal.SIGALRMで割り込む様に予約をしています。
そして、signal.SIGALRMによる割り込みに対し、シグナルハンドラhandlerが呼び出され、OSError例外が送出されるという流れです。
この例は、非常にシンプルながら強力で、単純に時間のかかる処理を一定時間で強制的に終わらせたりするにはこれだけで充分です。
ただし、問題点として、signal.alarm関数で設定できるシグナルはsignal.SIGALRM限定で、かつ、signal.alarmによるアラームは一つしか登録できません。
また、signal.SIGALRMは他の機能でも利用されていることがあり、干渉しないようにする必要があります。
そのため、他プロセスからのシグナル割り込みに利用するには不適切です。
ユーザ定義シグナルsignal.SIGUSR1を使う例
次は、signal.SIGUSR1を使った例です。
下記は、子プロセスを用いてユーザ定義シグナルsignal.SIGUSR1を5秒後に発生させ、それをシグナルハンドラhandlerで受け取る例です。
import signal
import os
import time
from multiprocessing import Process
# シグナルハンドラを定義
def handler(signum, frame):
print(f'handlerが呼び出されました(signum={signum})')
raise OSError("シグナルの割り込みがありました")
# signal.SIGUSR1のシグナルハンドラを登録
signal.signal(signal.SIGUSR1, handler)
def timer_procedure():
# 5秒間処理を停止
time.sleep(5)
# 親プロセスに対してsignal.SIGUSR1で割り込みを発生させる
os.kill(os.getppid(), signal.SIGUSR1)
timer = Process(target=timer_procedure)
timer.start()
# 時間のかかる処理／割り込みによって中断したい処理
while True:
print("hellow.")
time.sleep(0.7)
timer.join()
この例では、signal.alarmの代わりに、指定秒数後にos.killを用いて、TimerThreadからシグナル割り込みを発生させています。
この様に、ユーザ定義シグナルsignal.SIGUSR1を用いることで、他機能と干渉する可能性を低くすることができます。
また、この例のように子プロセスからシグナル割り込みを発生させずとも、別のターミナルを開いてそこからkillコマンドを実行したりしてもちゃんと動きます。
シグナルハンドラいつまで設定してるの問題
シグナルハンドラは、一度設定すると、上から別のハンドラを設定しない限り、設定されたままになります。
そのため、本来意図しないタイミングでシグナルを受け取ってしまい、割り込みを想定していない箇所で割り込み処理が走ってしまうことがあります。
これの対策としては、シグナルによる割り込み監視領域から出るときに、signal.signal関数を用いて、シグナルハンドラにsignal.SIG_IGNまたはsignal.SIG_DFLを設定します。
ただし、これはハンドラの上書きを行っているだけであるため、元々別のハンドラが設定されていた場合、元々のハンドラは行方不明に・・・。
真面目に書くのであれば、signal.signal関数の返り値を保持しておいて、割り込み監視領域から出るタイミングでハンドラをもとに戻す・・・ということをするのでしょうが、そこまでするのであればwith句で管理したいよねえ・・・。
ハンドラの定義箇所の問題
signalモジュールの使用上仕方がないことですが、シグナル割り込みによって実行する処理は、ハンドラ設定時に設定する必要があります。
しかし、もっと簡単にシグナルを使いたいPython使いとしては、可能な限り「処理の流れそのままに」書きたいのです。
なんなら、「シグナルハンドラ」の存在など気にせず書きたい。
理想の使い方を追い求め
ここまでの内容を踏まえて、こんな感じに書けたら非常に使いやすい／読みやすいと思うモジュールを作成しました。
import signal
import time
from signal_listener import SignalListener, SignalInterrupt
# 監視対象のシグナルを引数に、監視用オブジェクトの定義
listener = SignalListener(signal.SIGUSR1, signal.SIGUSR2)
try:
# シグナル割り込みを例外に置き換えて送出する領域
with listener.listen():
# シグナル割り込みによって中断したい処理
while True:
print("hellow.")
time.sleep(0.7)
# with句終了と同時に、元々設定されていたシグナルハンドラが再設定される
except SignalInterrupt.SIGUSR1:
print("SIGUSR1による割り込みが発生しました")
except SignalInterrupt.SIGUSR2:
print("SIGUSR2による割り込みが発生しました")
else:
print("シグナル割り込みは発生しませんでした")
finally:
print("シグナル割り込みの有無にかかわらず実行する処理")
どうですかこれ。
めちゃくちゃ読みやすくないですか？
推しポイントとしては、
- シグナルハンドラを設定する代わりに、対応する例外が送出される
- 例外監視領域の処理は中断される
- シグナル割り込みを想定／期待している箇所が明確
- シグナルハンドラの登録+削除の手間いらず。
withがやります。 - 割り込み処理が処理順（上→下）の流れで読める
です。
作ったソースコード
今回作成したコードですが、あえて3ファイルに分けて作っています。
from signal_listener.sigexc import SignalInterrupt
from signal_listener.signal_listener import SignalListener
import signal
class SignalInterrupt(Exception):
__subclasses = {}
def __init__(self, signum, sigframe, *args, **kwargs):
super(SignalInterrupt, self).__init__(
signum=signum, sigframe=sigframe, *args, **kwargs
)
@classmethod
def subclass(cls, sig, alias=None):
if isinstance(sig, signal.Signals):
alias = (sig.name if alias is None else alias)
signum = sig.value
else:
signum = sig
if signal.NSIG <= signum:
raise OSError(22, 'Invalid argument')
if signum not in cls.__subclasses:
def _sub_init_func(self, sigframe, *args, **kwargs):
super().__init__(signum, sigframe, *args, **kwargs)
if not alias:
alias = f"SIG{signum:02d}"
subcls_name = f"{cls.__name__}.{alias}"
_sub_cls = type(subcls_name, (cls,), {"__init__": _sub_init_func})
cls.__subclasses[signum] = _sub_cls
setattr(cls, alias, cls.__subclasses[signum])
return cls.__subclasses[signum]
for sig in signal.Signals:
SignalInterrupt.subclass(sig)
from abc import ABCMeta
import signal
from signal_listener.sigexc import SignalInterrupt
class SignalHandlerContext(metaclass=ABCMeta):
def __init__(self, sigset, handlers=None):
self.sigset = sigset
if handlers is None:
handlers = {}
self.handlers = handlers
def __raise_handler__(self, signum, sigframe):
"""シグナルを例外に置き換えて送出する"""
sigexc = SignalInterrupt.subclass(signum)
raise sigexc(sigframe)
def __enter__(self):
self._old_handler = {}
for sig in self.sigset:
if sig in self.handlers:
new_handler = self.handlers[sig]
else:
new_handler = self.__raise_handler__
self._old_handler[sig] = signal.signal(sig, new_handler)
def __exit__(self, exc_type, exc_value, traceback):
for sig in self.sigset:
signal.signal(sig, self._old_handler[sig])
class SignalListener(metaclass=ABCMeta):
def __init__(self, *signals):
self.sigset = set(signals)
def listen(self, handlers=None):
return SignalHandlerContext(self.sigset, handlers=handlers)
def sigwait(self):
return signal.sigwait(self.sigset)
def sigwaitinfo(self):
return signal.sigwaitinfo(self.sigset)
def sigtimedwait(self, timeout):
return signal.sigtimedwait(self.sigset, timeout)
ちなみに、理想の使い方を追い求めのコードは、signal_listenerディレクトリと同じ階層に置けば動作します。
終わりに
シグナルの扱いがめんどくさいという方はぜひ使ってみてほしいです。
そして感想もしくは苦情をぜひともお願いいたします。