Metadata-Version: 2.4
Name: procwarden
Version: 0.2.0
Summary: Linux/Docker subprocess supervisor with optional realtime output and process-tree cleanup.
Author-email: haicheng li <lemo.lihaicheng@gmail.com>
License: MIT
Project-URL: Homepage, https://example.com/procwarden
Project-URL: Documentation, https://example.com/procwarden/docs
Project-URL: Repository, https://example.com/procwarden
Keywords: subprocess,docker,linux,process-tree,signal,supervisor
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Systems Administration
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# ProcWarden

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

它的核心目标很窄：

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

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

## 安装

本地开发安装：

```bash
pip install -e .
```

普通安装：

```bash
pip install .
```

构建 wheel / sdist：

```bash
python -m pip install build
python -m build
```

## 快速开始

### 兼容旧接口

旧代码如果是这样：

```python
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)
```

可以继续使用。

### 推荐新名字

```python
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(...)`

```python
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(...)`

```python
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` 也使用同一份内容，因此通常能直接看到末尾的错误栈或报错行。

```python
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 继承环境：

```bash
conda run -n bioenv python main.py
```

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

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

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

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

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

## 并行调用和文件锁

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

```python
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 文件。

```python
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 直接作为容器主进程

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

### 如果使用 shell entrypoint，必须 `exec`

不推荐：

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

推荐：

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

### 更稳：使用 init

```bash
docker run --init your-image
```

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

## 行为说明

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

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

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

```text
SIGTERM
```

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

```text
SIGKILL
```

## 统计信息

```python
from procwarden import ProcWarden

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

返回示例：

```python
{
    "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 并实例化，会抛出：

```text
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，可以使用：

```python
runner = ProcWarden(register_signal_handlers=False)
```

## 版本

当前版本：`0.2.0`
