第172回では、複数のPython CLIを設定順に実行し、失敗した時点で後続処理を止めました。ただし、手動実行と定期実行が重なると、ワークフロー全体が二つ同時に動く可能性があります。
今回は、同じLinuxホスト上で、同じロックを使うコマンドを一つだけ実行できるラッパーを作ります。Python標準ライブラリのfcntl.flock()で非待機ロックを取得し、すでに実行中なら子コマンドを起動せず、終了コード75で停止します。
この仕組みが防ぐのは同時実行です。完了後の再実行、外部APIへの二重送信、途中まで作成されたファイルの取り消しは扱いません。
順番制御と排他制御は別に考える
| 仕組み | 防ぐ問題 | 防げない問題 |
|---|---|---|
| 第172回の順次実行器 | 実行順の乱れ、失敗後の後続実行 | ワークフロー全体の二重起動 |
| 今回の排他ラッパー | 同じロックを使うコマンドの同時実行 | 完了後の再実行、ラッパーを通さない実行 |
| 業務側の重複防止 | 同じ処理IDの再登録や二重送信 | 異なる処理同士の資源競合 |
請求、通知、外部API更新などには、処理IDを使った重複防止も別に必要です。
今回の実行契約
| 項目 | 仕様 |
|---|---|
| 動作環境 | Python 3.11以上、Linux、ローカルファイルシステム |
| ロック | 指定ファイルへ排他的な非待機ロックを取得する |
| 競合時 | 子コマンドを起動せず、標準エラー出力へ1行JSONを出して終了コード75 |
| 子コマンド | シェルを介さず、--より後ろの引数配列を実行する |
| 標準入力 | DEVNULLへ接続し、対話入力を待たせない |
| 標準出力・標準エラー出力 | 子コマンドの出力先をそのまま引き継ぐ |
| 終了コード | 子コマンドの値を引き継ぐ。シグナル終了は128 + シグナル番号へ変換する |
| ロックファイル | 終了後も削除しない。ファイルの存在ではなく、カーネルが管理するロック状態で判定する |
| 対象外 | ネットワークファイルシステム、複数ホストの分散ロック、タイムアウト、再試行、ロールバック |
flock()のロックは協調型です。このラッパーを通さない処理までは止められません。非待機ロックとpass_fdsの動作は、fcntl.flockの公式ドキュメントとsubprocess.Popenの公式ドキュメントで確認できます。
1．排他実行ラッパーを作る
次のコードをsingle_run.pyとして保存します。ロックファイルの親ディレクトリは事前に作成し、そのディレクトリへ書き込める利用者を限定してください。
from __future__ import annotations
import argparse
import errno
import fcntl
import json
import os
import stat
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import NoReturn, TextIO
EXIT_USAGE = 2
EXIT_RUNTIME = 3
EXIT_BUSY = 75
def emit_json(stream: TextIO, payload: dict[str, object]) -> None:
stream.write(
json.dumps(payload, ensure_ascii=False, separators=(',', ':')) + '\n'
)
stream.flush()
class AppError(Exception):
def __init__(self, exit_code: int, payload: dict[str, object]) -> None:
super().__init__(str(payload.get('reason', 'error')))
self.exit_code = exit_code
self.payload = payload
class JsonArgumentParser(argparse.ArgumentParser):
def error(self, message: str) -> NoReturn:
emit_json(
sys.stderr,
{
'status': 'error',
'reason': 'invalid_arguments',
'message': message,
},
)
raise SystemExit(EXIT_USAGE)
def parse_args(
argv: list[str] | None = None,
) -> tuple[Path, str, list[str]]:
raw = list(sys.argv[1:] if argv is None else argv)
parser = JsonArgumentParser(
usage='%(prog)s --lock LOCK_FILE -- COMMAND [ARG ...]',
description='同じロックを使うコマンドの同時実行を防ぎます。',
)
parser.add_argument('--lock', required=True, type=Path)
if '--' not in raw:
if '-h' in raw or '--help' in raw:
parser.parse_args(raw)
parser.error('区切りの -- と実行コマンドが必要です')
separator = raw.index('--')
namespace = parser.parse_args(raw[:separator])
command = raw[separator + 1:]
if not command:
parser.error('-- の後に実行コマンドを指定してください')
display_path = os.fspath(namespace.lock)
return namespace.lock, display_path, command
def write_lock_record(lock_fd: int, record: dict[str, object]) -> None:
data = (
json.dumps(record, ensure_ascii=False, separators=(',', ':')) + '\n'
).encode('utf-8')
os.ftruncate(lock_fd, 0)
os.lseek(lock_fd, 0, os.SEEK_SET)
remaining = memoryview(data)
while remaining:
written = os.write(lock_fd, remaining)
if written == 0:
raise OSError(errno.EIO, 'lock metadata write returned zero bytes')
remaining = remaining[written:]
os.fsync(lock_fd)
def acquire_lock(lock_path: Path, display_path: str) -> int:
flags = os.O_RDWR | os.O_CREAT
flags |= getattr(os, 'O_CLOEXEC', 0)
flags |= getattr(os, 'O_NOFOLLOW', 0)
try:
lock_fd = os.open(lock_path, flags, 0o600)
except FileNotFoundError as exc:
raise AppError(
EXIT_RUNTIME,
{
'status': 'error',
'reason': 'lock_parent_not_found',
'lock': display_path,
},
) from exc
except PermissionError as exc:
raise AppError(
EXIT_RUNTIME,
{
'status': 'error',
'reason': 'lock_permission_denied',
'lock': display_path,
},
) from exc
except OSError as exc:
raise AppError(
EXIT_RUNTIME,
{
'status': 'error',
'reason': 'lock_open_failed',
'lock': display_path,
},
) from exc
try:
try:
mode = os.fstat(lock_fd).st_mode
except OSError as exc:
raise AppError(
EXIT_RUNTIME,
{
'status': 'error',
'reason': 'lock_stat_failed',
'lock': display_path,
},
) from exc
if not stat.S_ISREG(mode):
raise AppError(
EXIT_RUNTIME,
{
'status': 'error',
'reason': 'lock_not_regular_file',
'lock': display_path,
},
)
try:
fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError as exc:
busy_numbers = {
errno.EACCES,
errno.EAGAIN,
errno.EWOULDBLOCK,
}
if exc.errno in busy_numbers:
raise AppError(
EXIT_BUSY,
{
'status': 'busy',
'reason': 'lock_held',
'lock': display_path,
},
) from exc
raise AppError(
EXIT_RUNTIME,
{
'status': 'error',
'reason': 'lock_acquire_failed',
'lock': display_path,
},
) from exc
record = {
'format_version': 1,
'owner_pid': os.getpid(),
'acquired_at_utc': datetime.now(timezone.utc)
.isoformat(timespec='seconds')
.replace('+00:00', 'Z'),
}
try:
write_lock_record(lock_fd, record)
except OSError as exc:
raise AppError(
EXIT_RUNTIME,
{
'status': 'error',
'reason': 'lock_metadata_write_failed',
'lock': display_path,
},
) from exc
return lock_fd
except BaseException:
try:
os.close(lock_fd)
except OSError:
pass
raise
def normalize_returncode(returncode: int) -> int:
if returncode >= 0:
return returncode
signal_number = -returncode
return min(255, 128 + signal_number)
def run_child(command: list[str], lock_fd: int) -> int:
try:
completed = subprocess.run(
command,
stdin=subprocess.DEVNULL,
check=False,
shell=False,
close_fds=True,
pass_fds=(lock_fd,),
)
except FileNotFoundError as exc:
raise AppError(
EXIT_RUNTIME,
{
'status': 'error',
'reason': 'command_not_found',
'command': command[0],
},
) from exc
except PermissionError as exc:
raise AppError(
EXIT_RUNTIME,
{
'status': 'error',
'reason': 'command_not_executable',
'command': command[0],
},
) from exc
except OSError as exc:
raise AppError(
EXIT_RUNTIME,
{
'status': 'error',
'reason': 'child_start_failed',
'command': command[0],
},
) from exc
return normalize_returncode(completed.returncode)
def close_lock(lock_fd: int) -> None:
# 明示的に解錠せず、複製された記述子がすべて閉じた時点で解放する。
try:
os.close(lock_fd)
except OSError:
pass
def main(argv: list[str] | None = None) -> int:
lock_fd: int | None = None
try:
lock_arg, display_path, command = parse_args(argv)
lock_path = Path(os.path.abspath(os.fspath(lock_arg)))
lock_fd = acquire_lock(lock_path, display_path)
return run_child(command, lock_fd)
except AppError as exc:
emit_json(sys.stderr, exc.payload)
return exc.exit_code
except KeyboardInterrupt:
return 130
except OSError:
emit_json(
sys.stderr,
{'status': 'error', 'reason': 'unexpected_os_error'},
)
return EXIT_RUNTIME
finally:
if lock_fd is not None:
close_lock(lock_fd)
if __name__ == '__main__':
raise SystemExit(main())
pass_fdsでロックのファイル記述子を直接の子プロセスにも渡しています。Linuxのflock()は複製された記述子と同じロックを共有し、最後の記述子が閉じるまで保持されます。このため、ラッパーだけが強制終了しても、子プロセスが動いている間はロックが残ります。
反対に、子プロセスが継承した記述子へ明示的なLOCK_UNを実行すると、ラッパーが持つロックも解除されます。子プロセスからこの記述子を操作しないことが前提です。
ロックファイルにはコマンドや引数を書きません。引数に秘密情報が含まれる場合があるためです。PIDと取得時刻は調査用であり、現在も実行中であることの証明には使いません。
2．動作確認用の子コマンドを作る
次のコードをdemo_job.pyとして、single_run.pyと同じディレクトリへ保存します。
from __future__ import annotations
import argparse
import json
import time
def seconds_arg(text: str) -> int:
try:
value = int(text)
except ValueError as exc:
raise argparse.ArgumentTypeError('整数を指定してください') from exc
if not 1 <= value <= 30:
raise argparse.ArgumentTypeError('1から30までを指定してください')
return value
parser = argparse.ArgumentParser()
parser.add_argument('--seconds', required=True, type=seconds_arg)
args = parser.parse_args()
time.sleep(args.seconds)
print(
json.dumps(
{'status': 'ok', 'slept_seconds': args.seconds},
ensure_ascii=False,
separators=(',', ':'),
)
)
ロックを取得し、子コマンドを8秒間動かします。
python3 single_run.py --lock workflow.lock -- python3 demo_job.py --seconds 8完了すると、子コマンドの標準出力が表示されます。
{"status":"ok","slept_seconds":8}workflow.lockは初回に作成され、処理後も残ります。正常に記録された内容は次のコマンドで確認できます。
python3 -m json.tool workflow.lockメタデータの書き込み中にプロセスを強制終了すると、内容が空または不完全になる可能性があります。次回のロック取得時に上書きされるため、内容ではなくロック取得の成否を実行中の判定に使います。
3．二つのシェルから同時に実行する
一つ目のシェルで8秒の処理を開始し、完了する前に二つ目のシェルで次を実行します。
python3 single_run.py --lock workflow.lock -- python3 demo_job.py --seconds 1二つ目は待機せず、標準エラー出力へ次のJSONを出します。
{"status":"busy","reason":"lock_held","lock":"workflow.lock"}終了コードは75です。
echo $?
75一つ目が完了した後なら、ロックファイルが残っていても正常に開始できます。
4．子コマンドの失敗を引き継ぐ
子コマンドが終了コード7を返す例です。
python3 single_run.py --lock workflow.lock -- python3 -c 'raise SystemExit(7)'
echo $?
7存在しない実行ファイルを指定した場合は、ラッパー自身のエラーとして終了コード3になります。
python3 single_run.py --lock workflow.lock -- ./not-present-command{"status":"error","reason":"command_not_found","command":"./not-present-command"}| 終了コード | 意味 |
|---|---|
| 0 | 子コマンドが正常終了 |
| 2 | ラッパーの引数が不正 |
| 3 | ロックファイル操作または子コマンド起動に失敗 |
| 75 | 指定したロックを別の処理が保持中 |
| 130 | ラッパーがKeyboardInterruptを受けた |
| そのほか | 子コマンドの終了コード、またはシグナルを変換した値 |
子コマンド自身も2、3、75を返す可能性があります。ラッパー側の失敗かを区別する場合は、終了コードだけでなく、標準エラー出力のstatusとreasonも確認します。
通常の端末でCtrl+Cを押すと、同じフォアグラウンドプロセスグループにいるラッパーと子プロセスの両方へSIGINTが送られます。このコードに独自のシグナル転送処理はありません。ただし、subprocess.run()が子プロセスの終了を待っている間にラッパーだけがKeyboardInterruptを受けた場合は、subprocess.run()の例外処理が直接の子プロセスを強制終了してから例外を再送出し、main()が終了コード130を返します。そのため、直接の子プロセスがそのまま動き続けるのが通常の挙動ではありません。子プロセスが別のプロセスを起動し、その子孫にもロック記述子を継承させていた場合は、その子孫が動き続けてロックを保持する可能性があります。
第172回のワークフローへ適用する
第172回の実行コマンド全体を、今回の--より後ろへ置きます。正常時にはラッパーが出力を追加しないため、子CLIの標準出力と終了コードをそのまま受け取れます。
定期実行では、作業ディレクトリの違いによって別のロックファイルを作らないよう、ラッパー、ロックファイル、子コマンド、設定ファイルを絶対パスで指定します。複数のワークフローが同じ出力先を更新するなら、スクリプト名が異なっても同じロックファイルを使います。
運用時の制約
- ロックファイルのパスを固定し、すべての起動経路で同じ値を使う
- ロックファイルを削除、移動、ローテーションしない
- PIDや取得時刻を見て「古いロック」と判断し、自動削除しない
- 子コマンドはフォアグラウンドで動かし、別プロセスを残して先に終了しない
- 子コマンドから継承したロック記述子を操作しない
- NFSやSMBなどのネットワークファイルシステム、複数ホストの排他には使わない
- パイプ、リダイレクト、ワイルドカードは解釈されない。必要な処理は個別の引数またはPython側で組み立てる
実行中にロックファイルを削除すると、既存処理は削除前のinodeを保持したまま、新しい処理が同名の別ファイルを作成できます。その結果、二つの処理が別々のファイルをロックして同時に動くことがあります。
最初に二つのシェルで競合時の動作を確認します。その後、第172回のワークフローを外側から包み、手動実行と定期実行の両方で同じ絶対パスのロックを指定します。