如何用Python subprocess运行管道命令避免死锁？

你的核心问题是Python通过subprocess循环执行带多管道的shell命令时出现死锁挂起，但直接用bash脚本执行完全正常。这大概率是Python子进程的stdout/stderr缓冲区阻塞导致的：当子进程产生大量输出（哪怕你的命令最终把输出写入文件，中间管道的临时输出也可能填满缓冲区），而Python没有及时读取这些数据时，子进程会被卡住等待父进程消费缓冲区，最终导致死锁，表现为CPU占用100%且无限运行。

另外，Singularity容器和Snakemake的环境可能会放大这个问题——容器内的文件描述符限制、Snakemake对输出的捕获机制都可能加剧缓冲区积压。

以下是几个针对性的解决方法，按推荐优先级排序：

1. 用临时bash脚本执行命令 #

把每个长管道命令封装成独立的bash脚本，让Python调用bash执行脚本，而非直接解析管道命令。这种方式和你手动运行bash example.sh的环境完全一致，能避免Python对shell管道的解析问题：

import subprocess
import tempfile
import os
from your_module import info  # 替换为你的日志函数所在模块

for cmd in chr_commands:
    info(f"Run: {cmd}")
    # 创建临时shell脚本
    with tempfile.NamedTemporaryFile(mode='w', suffix='.sh', delete=False) as tmp_sh:
        tmp_sh.write("#!/bin/bash\n")
        tmp_sh.write(cmd + "\n")
    # 给脚本加执行权限
    os.chmod(tmp_sh.name, 0o755)
    try:
        # 调用bash执行脚本
        subprocess.run(
            ["bash", tmp_sh.name],
            check=True,
            close_fds=True  # 确保关闭多余文件描述符，避免泄漏
        )
    finally:
        # 删除临时脚本
        os.unlink(tmp_sh.name)

2. 正确配置subprocess参数，避免缓冲区阻塞 #

如果不想用临时脚本，直接调用subprocess.run时，需要明确让shell解析命令，并确保Python不阻塞子进程的输出缓冲区：

import subprocess
from your_module import info

for cmd in chr_commands:
    info(f"Run: {cmd}")
    subprocess.run(
        cmd,
        shell=True,  # 必须开启，让shell解析管道符
        check=True,
        close_fds=True,
        # 让子进程的stdout/stderr直接继承父进程的输出（交给Snakemake处理）
        stdout=None,
        stderr=None
    )

如果担心输出干扰，也可以把stdout/stderr重定向到文件（和你的命令里的2>>log.txt配合）：

subprocess.run(
    cmd,
    shell=True,
    check=True,
    close_fds=True,
    stdout=open("/dev/null", "w"),  # 丢弃stdout（你的命令已把输出写入文件）
    stderr=subprocess.STDOUT  # 把stderr合并到stdout，或直接指定到log.txt
)

3. 用Popen+communicate()主动消费缓冲区 #

communicate()方法会主动读取子进程的stdout和stderr，避免缓冲区满导致的阻塞，适合需要捕获输出的场景：

import subprocess
from your_module import info

for cmd in chr_commands:
    info(f"Run: {cmd}")
    proc = subprocess.Popen(
        cmd,
        shell=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True  # 直接返回字符串而非字节
    )
    # 读取所有输出，避免缓冲区阻塞
    stdout, stderr = proc.communicate()
    if proc.returncode != 0:
        info(f"Command failed: {cmd}\nError: {stderr}")

• 避免用map_async等并行方式执行：你的命令本身涉及大量IO和CPU操作，并行可能导致资源竞争，反而降低效率，甚至加剧死锁风险。
• 检查Singularity容器的资源限制：确保容器有足够的文件描述符和内存，避免因资源不足导致进程卡住。
• 简化awk脚本：你的awk逻辑比较复杂，尤其是第二个awk里的字符串拼接（fastq_records=fastq_records...）可能会占用大量内存，当处理大文件时也可能导致进程假死，可考虑拆分逻辑或用更高效的方式处理。

内容的提问来源于stack exchange，提问作者Niek de Klein