procwarden 0.2.0

Linux/Docker subprocess supervisor with optional realtime output and process-tree cleanup.

ProcWarden

ProcWarden 是一个 Linux / Docker 专用的 Python 子进程守卫器，用来替代项目里零散的 subprocess.Popen 包装逻辑，避免每次都重新写。

它的核心目标很窄：

1. 控制是否实时输出命令日志；
2. 在收到 Ctrl+C、SIGTERM、docker stop 等停止信号时，杀掉命令启动的整棵进程树；
3. 保留旧脚本的 PipelineManager / run_command(...) 接口，方便平滑迁移。

包名：procwarden
推荐类名：ProcWarden
兼容旧类名：PipelineManager

安装

本地开发安装：

pip install -e .

普通安装：

pip install .

构建 wheel / sdist：

python -m pip install build
python -m build

快速开始

兼容旧接口

旧代码如果是这样：

from procwarden import PipelineManager

with PipelineManager() as pm:
    result = pm.run_command(
        "python worker.py",
        real_time_output=True,
        timeout=3600,
    )

print(result.returncode)
print(result.stdout)

可以继续使用。

推荐新名字

from procwarden import ProcWarden

runner = ProcWarden(log_prefix="job-1")

result = runner.run_command(
    "python worker.py",
    real_time_output=True,
    timeout=3600,
)

API

ProcWarden(...) / PipelineManager(...)

ProcWarden(
    graceful_timeout: float = 2.0,
    final_wait_timeout: float = 1.0,
    kill_process_group: bool = True,
    log_prefix: str = "",
    max_processes: int = 100,
    register_signal_handlers: bool = True,
    max_output_bytes: int | None = 67108864,
    max_output_tail_bytes: int = 4194304,
    output_chunk_size: int = 65536,
    realtime_output_max_bytes_per_sec: int | None = 1048576,
    realtime_flush_interval: float = 0.2,
    output_encoding: str = "utf-8",
    output_errors: str = "replace",
    env: Mapping[str, str | None] | None = None,
    cwd: str | PathLike | None = None,
)

参数说明：

参数	说明
graceful_timeout	发送 SIGTERM 后等待进程组退出的秒数
final_wait_timeout	发送 SIGKILL 后额外等待的秒数
kill_process_group	是否杀进程组；Linux/Docker 场景建议保持 True
log_prefix	日志前缀
max_processes	最多同时跟踪多少个子进程
register_signal_handlers	是否注册 SIGINT / SIGTERM handler
max_output_bytes	最多缓存多少字节开头输出，默认 64 MiB；设为 None 表示不限
max_output_tail_bytes	输出被截断时额外保留多少字节末尾输出，默认 4 MiB，方便保留报错信息
output_chunk_size	从 stdout/stderr 管道读取输出的块大小
realtime_output_max_bytes_per_sec	实时打印每秒最多输出多少字节，默认 1 MiB/s；设为 None 表示不限
realtime_flush_interval	实时输出 flush 间隔，避免每行 flush 导致高频日志卡住
output_encoding	输出解码编码
output_errors	输出解码错误处理策略
env	实例级环境变量覆盖；值为 None 时删除该环境变量
cwd	实例级默认工作目录

run_command(...)

run_command(
    cmd: str | Sequence[str],
    shell: bool = True,
    check: bool = True,
    real_time_output: bool = False,
    timeout: float | None = None,
    max_output_bytes: int | None = <use instance default>,
    max_output_tail_bytes: int = <use instance default>,
    realtime_output_max_bytes_per_sec: int | None = <use instance default>,
    env: Mapping[str, str | None] | None = None,
    cwd: str | PathLike | None = None,
    conda_env: str | None = None,
    conda_prefix: str | PathLike | None = None,
    lock_paths: Sequence[str | PathLike] | None = None,
    lock_timeout: float | None = None,
    lock_shared: bool = False,
) -> subprocess.CompletedProcess

为了兼容旧脚本，shell=True 仍然是默认值。

参数	说明
cmd	要执行的命令字符串；shell=False 时也支持 argv 序列
shell	是否通过 shell 执行；为兼容旧接口默认是 True
check	非 0 返回码是否抛 subprocess.CalledProcessError
real_time_output	是否边执行边打印输出；高频日志会按速率限制打印，但仍持续读取管道
timeout	最大运行秒数，超时后杀进程树
max_output_bytes	本次命令最多缓存多少开头输出字节；None 表示不限
max_output_tail_bytes	本次命令在截断时额外保留多少末尾输出字节
realtime_output_max_bytes_per_sec	本次命令实时打印速率限制；None 表示不限
env	本次命令的环境变量覆盖
cwd	本次命令的工作目录
conda_env	通过 conda run -n <env> 执行命令
conda_prefix	将某个 Conda 环境前缀的 bin 加入 PATH，并设置 CONDA_PREFIX
lock_paths	命令运行期间需要互斥访问的文件或资源路径
lock_timeout	等待文件锁的最长秒数；None 表示一直等
lock_shared	是否使用共享锁，适合只读场景

返回值是 subprocess.CompletedProcess，其中：

• stdout：合并后的 stdout/stderr；
• stderr：固定为 None，因为当前实现使用 stderr=subprocess.STDOUT；
• returncode：子进程返回码。

长时间生信任务和高频日志

GATK HaplotypeCaller、fastp 等工具通常会运行很久并持续打印进度。ProcWarden 会持续读取 stdout/stderr 管道，避免子进程因为管道写满而阻塞；同时默认缓存前 64 MiB 输出和最后 4 MiB 输出，并把实时打印限制在 1 MiB/s，避免日志过多拖垮终端或容器日志驱动。

如果输出超过缓存上限，stdout 中会包含截断提示，然后追加最后一段 tail 输出。命令失败且 check=True 时，subprocess.CalledProcessError.output 也使用同一份内容，因此通常能直接看到末尾的错误栈或报错行。

from procwarden import ProcWarden

runner = ProcWarden(log_prefix="gatk")
result = runner.run_command(
    [
        "gatk",
        "HaplotypeCaller",
        "-R", "ref.fa",
        "-I", "sample.bam",
        "-O", "sample.g.vcf.gz",
    ],
    shell=False,
    real_time_output=True,
    timeout=None,
)

Conda / 环境管理

生信工具通常由 Conda/Mamba 管理，并依赖环境里的动态库、插件或同环境中的其他命令。推荐优先让容器或任务入口先激活环境，然后 ProcWarden 继承环境：

conda run -n bioenv python main.py

如果需要在 Python 里指定环境，可以使用 conda_env：

runner.run_command(
    ["gatk", "--version"],
    shell=False,
    conda_env="bioenv",
)

如果已经知道环境路径，conda_prefix 更轻量：它只设置 CONDA_PREFIX 并把 <prefix>/bin 放到 PATH 最前面，不额外启动 conda run：

runner.run_command(
    ["fastp", "--version"],
    shell=False,
    conda_prefix="/opt/conda/envs/bioenv",
)

实例级和命令级 env 可用于补充 PATH、LD_LIBRARY_PATH、TMPDIR 等变量。值设为 None 会从子进程环境中删除该变量。

并行调用和文件锁

run_commands_parallel(...) 会在线程池中并行启动多个子进程，并保持返回结果顺序与输入顺序一致：

results = runner.run_commands_parallel(
    [
        {"cmd": ["fastp", "-i", "a.fq.gz", "-o", "a.clean.fq.gz"], "shell": False},
        {"cmd": ["fastp", "-i", "b.fq.gz", "-o", "b.clean.fq.gz"], "shell": False},
    ],
    max_workers=2,
    real_time_output=True,
)

并行任务如果会读写同一个文件、索引、临时目录或数据库，给命令加 lock_paths。ProcWarden 会在同一进程内用线程锁、在多进程间用 fcntl.flock 锁住对应资源的 .procwarden.lock sidecar 文件。

runner.run_commands_parallel(
    [
        {"cmd": ["tool", "--db", "shared.sqlite", "--sample", "A"], "shell": False},
        {"cmd": ["tool", "--db", "shared.sqlite", "--sample", "B"], "shell": False},
    ],
    max_workers=2,
    lock_paths=["shared.sqlite"],
)

只读任务可使用 lock_shared=True。写任务应使用默认独占锁，并尽量把每个样本的输出、临时目录和日志文件拆开，减少不必要的串行等待。

Docker 使用建议

ProcWarden 只能保证：Python 进程收到停止信号后，会清理它启动的子进程树。

因此 Docker 里需要确保信号能到达 Python 进程。

推荐：Python 直接作为容器主进程

CMD ["python", "main.py"]

如果使用 shell entrypoint，必须 exec

不推荐：

#!/usr/bin/env bash
python main.py

推荐：

#!/usr/bin/env bash
exec python main.py

更稳：使用 init

docker run --init your-image

或在镜像里使用 tini / dumb-init。

行为说明

每个命令启动时，ProcWarden 会使用：

subprocess.Popen(..., start_new_session=True)

这会让子进程成为新的 session / process group leader。停止时，ProcWarden 会先对该进程组发送：

SIGTERM

如果进程组没有在 graceful_timeout 内退出，再发送：

SIGKILL

统计信息

from procwarden import ProcWarden

runner = ProcWarden()
runner.run_command("echo hello")
print(runner.get_process_stats())

返回示例：

{
    "total_processes": 0,
    "active_processes": 0,
    "is_shutting_down": False,
    "oldest_process_age": 0,
    "total_started": 1,
    "total_finished": 1,
    "total_terminated": 0,
    "total_killed": 0,
    "total_timeout": 0,
    "max_concurrent": 1,
    "recent_history": [
        {
            "pid": 12345,
            "cmd": "echo hello",
            "rc": 0,
            "state": "completed",
            "duration_sec": 0.003,
        }
    ],
}

注意事项

1. 这是 Linux/POSIX 专用包

如果在 Windows 上 import 并实例化，会抛出：

RuntimeError: ProcWarden/PipelineManager is Linux/POSIX only

2. 默认 shell=True 是为了兼容旧接口

如果命令里包含外部输入，请避免字符串拼接，或者显式使用更安全的命令构造方式。当前版本为了兼容旧脚本，没有把默认值改成 shell=False。

3. stdout/stderr 会合并

当前版本使用 stderr=subprocess.STDOUT，因此 CompletedProcess.stderr 为 None。

4. 多个 ProcWarden 实例会覆盖进程级 signal handler

一般建议一个进程里只创建一个长期使用的 ProcWarden。如果你不希望它注册 signal handler，可以使用：

runner = ProcWarden(register_signal_handlers=False)

版本

当前版本：0.2.0