第172回では複数のCLIを順番に実行し、第173回では二重起動を防ぎ、第174回では一時失敗だけを再試行しました。
ただし、タイムアウト時に直接の子プロセスだけを終了すると、その子が起動した処理は残ることがあります。そのまま再試行すれば、新旧の処理が重なります。
今回は子コマンドを新しいプロセスグループで起動し、タイムアウト時に同じグループへSIGTERMを送ります。猶予後も残る処理にはSIGKILLを送り、直接の子を回収して終了するラッパーを作ります。
対象はPython 3.12以上を実行するLinuxなど、POSIXシグナルとos.killpg()を利用できる環境です。外部ライブラリは使いません。
プロセスグループを使う理由
Popen.terminate()が停止するのは直接の子です。子が同じプロセスグループ内に孫プロセスを起動しても、その孫には信号が届きません。
start_new_session=Trueを指定すると、POSIX環境では子の実行前にsetsid()が呼ばれます。直接の子は新しいセッションとプロセスグループのリーダーになるため、そのPIDをグループIDとしてos.killpg()へ渡せます。
ただし、別のグループやセッションへ移動した子孫には届きません。外部サービスへの送信や作成済みファイルも取り消せません。
今回の実行契約
| 項目 | 仕様 |
|---|---|
| 子コマンド | --より後ろの引数配列を、シェルを介さず実行する |
| タイムアウト | --timeoutで1～86400秒を指定する |
| 終了猶予 | --graceで0～60秒を指定する。既定値は5秒 |
| 標準入力 | DEVNULLへ接続し、対話入力を待たせない |
| 標準出力・標準エラー出力 | 子コマンドの出力先を引き継ぐ。ラッパーの診断は標準エラー出力へ書く |
| タイムアウト時 | グループへSIGTERMを送り、猶予後にSIGKILLを送る |
| Ctrl+C | 子の終了待ち中に受けた場合は同じ後始末を行い、成功時は130を返す |
| 確認範囲 | SIGKILL後、直接の子が5秒以内に終了したことまで確認する |
--timeoutはPopen.wait()の待機時間です。プロセスの起動時間、終了猶予、回収待ちは含みません。ラッパー全体の厳密な実行時間上限ではありません。
終了コード124は、タイムアウト後に直接の子を回収できたことを表します。プロセスグループ全体の消滅を保証する値ではありません。後始末に失敗した場合は125を返します。
1．実行ラッパーを保存する
次のコードをgroup_timeout.pyとして保存します。
#!/usr/bin/env python3
import argparse
import os
import signal
import subprocess
import sys
import time
TIMEOUT_EXIT = 124
CLEANUP_EXIT = 125
CANNOT_EXECUTE_EXIT = 126
NOT_FOUND_EXIT = 127
INTERRUPTED_EXIT = 130
KILL_WAIT_SECONDS = 5
def integer_between(minimum, maximum):
def parse(text):
try:
value = int(text)
except ValueError as exc:
raise argparse.ArgumentTypeError(
"整数で指定してください"
) from exc
if not minimum <= value <= maximum:
raise argparse.ArgumentTypeError(
f"{minimum}から{maximum}までの整数で指定してください"
)
return value
return parse
def parse_args(argv):
parser = argparse.ArgumentParser(
description="子コマンドを新しいプロセスグループで実行します"
)
parser.add_argument(
"--timeout",
required=True,
type=integer_between(1, 86400),
help="終了待ちのタイムアウト（秒、1から86400）",
)
parser.add_argument(
"--grace",
default=5,
type=integer_between(0, 60),
help="SIGTERM後の猶予（秒、0から60、既定値5）",
)
parser.add_argument("command", nargs=argparse.REMAINDER)
args = parser.parse_args(argv)
if len(args.command) < 2 or args.command[0] != "--":
parser.error("子コマンドの前に -- を置いてください")
args.command = args.command[1:]
return args
def report(message):
print(
f"group_timeout: {message}",
file=sys.stderr,
flush=True,
)
def send_group(process_group_id, signal_number):
try:
os.killpg(process_group_id, signal_number)
except ProcessLookupError:
return False
return True
def wait_grace(seconds):
deadline = time.monotonic() + seconds
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
return
try:
time.sleep(min(0.1, remaining))
except KeyboardInterrupt:
report("追加の割り込みを受けたため猶予を打ち切ります")
return
def reap_after_kill(process):
deadline = time.monotonic() + KILL_WAIT_SECONDS
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
return False
try:
process.wait(timeout=remaining)
return True
except subprocess.TimeoutExpired:
return False
except KeyboardInterrupt:
continue
def signal_group_and_reap_child(process, grace):
process_group_id = process.pid
if send_group(process_group_id, signal.SIGTERM):
report(
f"process_group={process_group_id} signal=SIGTERM"
)
wait_grace(grace)
if send_group(process_group_id, signal.SIGKILL):
report(
f"process_group={process_group_id} signal=SIGKILL"
)
return reap_after_kill(process)
def cleanup(process, grace):
try:
reaped = signal_group_and_reap_child(process, grace)
except OSError as exc:
report(f"プロセスグループへの信号送信に失敗しました: {exc}")
try:
process.kill()
process.wait(timeout=KILL_WAIT_SECONDS)
except (
OSError,
subprocess.TimeoutExpired,
KeyboardInterrupt,
):
pass
return False
if not reaped:
report(
"SIGKILL後も直接の子プロセスの終了を確認できませんでした"
)
return reaped
def main(argv=None):
args = parse_args(argv)
if os.name != "posix" or not hasattr(os, "killpg"):
report("os.killpgを使えるPOSIX環境専用です")
return 2
try:
process = subprocess.Popen(
args.command,
stdin=subprocess.DEVNULL,
stdout=None,
stderr=None,
shell=False,
start_new_session=True,
)
except FileNotFoundError as exc:
report(f"実行ファイルが見つかりません: {exc}")
return NOT_FOUND_EXIT
except PermissionError as exc:
report(f"実行権限がありません: {exc}")
return CANNOT_EXECUTE_EXIT
except OSError as exc:
report(f"子コマンドを起動できません: {exc}")
return CANNOT_EXECUTE_EXIT
try:
return_code = process.wait(timeout=args.timeout)
except subprocess.TimeoutExpired:
report(f"timeout={args.timeout}s")
if not cleanup(process, args.grace):
return CLEANUP_EXIT
return TIMEOUT_EXIT
except KeyboardInterrupt:
report("keyboard_interrupt")
if not cleanup(process, args.grace):
return CLEANUP_EXIT
return INTERRUPTED_EXIT
if return_code < 0:
return 128 + (-return_code)
return return_code
if __name__ == "__main__":
raise SystemExit(main())
猶予中に直接の子が終了しても、指定時間が終わるまで待ちます。親だけがSIGTERMで終了し、同じグループの孫が残る場合にも、その後のSIGKILLを送るためです。
shell=Falseなので、ワイルドカード、パイプ、リダイレクトは解釈されません。ただし、利用者が自由にコマンド名や引数を指定できる設計にすれば、任意のプログラムや危険なオプションを実行されます。実運用では許可するコマンドを固定してください。
2．正常終了と子コマンドの失敗を確認する
まず正常終了を確認します。
python3 group_timeout.py --timeout 5 --grace 1 -- \
python3 -c 'print("ok")'
echo $?
標準出力にはok、終了コードには0が表示されます。
子コマンドが終了コード7を返す場合も確認します。
python3 group_timeout.py --timeout 5 -- \
python3 -c 'raise SystemExit(7)'
echo $?
終了コードは7です。子がシグナル番号Nで終了した場合は、Popenが返す負の値を128 + Nへ変換します。
3．SIGTERMを無視する孫プロセスで確認する
次のコードをstubborn_tree.pyとして保存します。孫はSIGTERMを無視します。親は、孫がシグナルハンドラの設定を終えたことを確認してからPIDを表示します。
#!/usr/bin/env python3
import os
import signal
import subprocess
import sys
import time
from pathlib import Path
if sys.argv[1:] == ["grandchild"]:
signal.signal(signal.SIGTERM, signal.SIG_IGN)
print("ready", flush=True)
while True:
time.sleep(1)
if sys.argv[1:]:
raise SystemExit("usage: python3 stubborn_tree.py")
grandchild = subprocess.Popen(
[
sys.executable,
str(Path(__file__).resolve()),
"grandchild",
],
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
text=True,
)
ready_stream = grandchild.stdout
if ready_stream is None:
raise RuntimeError("grandchild stdout is unavailable")
ready = ready_stream.readline()
ready_stream.close()
if ready != "ready\n":
return_code = grandchild.wait()
raise SystemExit(
f"grandchild setup failed: return_code={return_code}"
)
print(
f"parent_pid={os.getpid()} grandchild_pid={grandchild.pid}",
flush=True,
)
while True:
time.sleep(1)
2秒でタイムアウトさせ、SIGTERMから1秒後にSIGKILLへ進めます。
python3 group_timeout.py --timeout 2 --grace 1 -- \
python3 stubborn_tree.py
echo $?
出力はおおむね次の形になります。PIDは実行ごとに変わります。
parent_pid=<親PID> grandchild_pid=<孫PID>
group_timeout: timeout=2s
group_timeout: process_group=<親PID> signal=SIGTERM
group_timeout: process_group=<親PID> signal=SIGKILL
124
124で確認できるのは直接の子の回収までです。表示されたPIDを使い、直後にプロセス表も調べます。
ps -o pid=,ppid=,pgid=,stat=,args= \
-p <親PID>,<孫PID>
何も表示されなければ、両方ともプロセス表から消えています。状態欄がZなら実行は終わっていますが、回収待ちです。それ以外の状態で対象のコマンドが残っている場合は、自動再試行せず、PGIDと起動方法を確認します。
4．終了コードを運用へ渡す
| 終了コード | 意味 |
|---|---|
| 0 | 子コマンドが正常終了した |
| 子コマンドの非ゼロ値 | 通常終了した子の値をそのまま返した |
| 2 | 引数が不正、または対象外の環境だった |
| 124 | タイムアウト後、直接の子の終了を確認した |
| 125 | 信号送信に失敗したか、直接の子の終了を確認できなかった |
| 126 | 権限などの理由で子コマンドを起動できなかった |
| 127 | 実行ファイルが見つからなかった |
| 130 | 終了待ち中にCtrl+Cを受け、後始末を完了した |
| 128 + N | 子がシグナル番号Nで終了した |
存在しない実行ファイルは127になります。
python3 group_timeout.py --timeout 5 -- \
./does-not-exist-for-test
echo $?
--timeout 0、範囲外の--grace、子コマンド前の--がない呼び出しは、実行前に引数エラーとなります。
124～127や130は子コマンド自身も返せます。終了コードだけで原因を判定する運用では、これらをラッパー用として子側の契約から除外してください。ラッパーが検出した事象には、標準エラー出力へgroup_timeout:で始まる診断も出ます。
このラッパーが扱えない範囲
- 子孫が
setsid()やsetpgid()で別のグループへ移動すると、今回のkillpg()では停止できない - 別ホスト、コンテナ管理機能、ジョブキューへ渡した処理は対象外である
- 直接の子が正常終了し、バックグラウンド処理だけを残した場合は停止処理を行わない
SIGKILLでは終了処理を実行できないため、一時ファイルや未完成の出力が残り得る- 外部APIへの登録、メール送信、確定済みデータなどの副作用は取り消さない
- ラッパー自身への
SIGTERM、SIGHUP、SIGKILL、電源断の後始末は実装していない - 起動中や後始末中に繰り返し届く
SIGINTを、あらゆる位置で処理する設計ではない
子コマンドは、自分が起動した処理を待ってから正常終了する設計にします。出力は一時ファイルへ書き、検証後に置き換えるなど、途中終了を前提にしてください。
再試行と組み合わせる順序
- ワークフロー全体の排他ロックを取得する
- 再試行ラッパーが1回分の試行を始める
- 今回のラッパーが、その1回にタイムアウトを設定する
- 実際の子CLIを新しいプロセスグループで起動する
終了コード124を無条件に再試行対象へ加えてはいけません。外部更新の直後に応答待ちで止まった可能性があるためです。処理IDなどによる重複防止があり、同じ入力を再実行しても結果が壊れない場合に限って再試行します。
このラッパーが管理するのは、タイムアウト時に信号を送る範囲と、直接の子の回収です。プロセスグループ全体の消滅、処理結果の取り消し、再実行の安全性は別に確認します。