为何env参数会导致Python Popen出现挂起死锁问题？

补充说明：已确定问题由env参数相关因素导致，移除env参数后运行正常。

遇到典型的Python Popen 问题：用以下代码通过Popen创建异步迭代器，多数情况正常，但偶尔死锁，和uvicorn服务器同进程运行时死锁频繁。

import os
import subprocess
from threading import Thread
from queue import Queue
from typing import IO, AsyncIterator, Dict, Iterable, Optional, Tuple
from enum import Enum, IntEnum
import sys


class StdFd(IntEnum):
    stdout = 0
    stderr = 1
    stdeof = 2

def read_fd(pipe :  IO[bytes], queue : Queue[Tuple[StdFd, bytes]], fd : Optional[StdFd] = StdFd.stdout):
    try:
        with pipe:
            for line in iter(pipe.readline, b""):
                queue.put((fd, line))
    finally:
        queue.put((StdFd.stdeof, b""))

async def call_with_env(cmd : Iterable[str], *, env : Dict[str, str])->AsyncIterator[Tuple[StdFd, bytes]]:
    
    queue : Queue[Tuple[StdFd, bytes]] = Queue()
    process = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE, 
        stderr=subprocess.PIPE, 
        env=env,
    )
   
    Thread(target=read_fd, args=[process.stdout, queue, StdFd.stdout]).start()
    Thread(target=read_fd, args=[process.stderr, queue, StdFd.stderr]).start()
    
    hits = 0
    while hits < 2:
        try:
            next = queue.get(block=False)
            
            if next[0] == StdFd.stdeof:
                hits += 1
                continue
        
            yield next
        except Exception:
            continue
        
    yield (StdFd.stdeof, b"")

尝试过多次刷新操作无效，调用process.communicate()也有相同行为但不符合需求。Python版本为3.10.10，使用asyncio.create_subprocess_exec时同样遇到问题。

接下来可以检查的方向： #

• 检查env参数完整性：手动传入env时容易遗漏系统默认关键变量（如PATH、HOME或目标命令依赖的特定变量），导致子进程启动异常。建议基于os.environ扩展env：

  env = os.environ.copy()
  env.update(你的自定义变量)
  

• 监控子进程退出状态：死锁时子进程可能未正常退出，可在循环中定期调用process.poll()，若子进程已退出但未收到两个EOF，手动触发结束逻辑，避免无限等待。
• 优化队列轮询逻辑：当前block=False+异常捕获的轮询方式效率低且可能丢消息，可改用带超时的阻塞等待，结合子进程状态判断：

  while hits < 2:
      try:
          next_item = queue.get(timeout=1)
          if next_item[0] == StdFd.stdeof:
              hits += 1
              continue
          yield next_item
      except queue.Empty:
          # 子进程已退出则终止循环
          if process.poll() is not None:
              break
  

• 排查管道缓冲问题：部分命令在非终端环境下会缓冲输出，导致数据无法实时写入管道。可尝试给Popen设置bufsize=0（无缓冲），或让目标命令强制刷新输出（如Python脚本用print(..., flush=True)）。
• 适配uvicorn异步模型：uvicorn的异步IO模型可能和手动创建的线程冲突，建议改用asyncio原生方式处理子进程管道，避免手动线程：

  import asyncio
  from typing import AsyncIterator, Dict, Iterable, Tuple
  from enum import IntEnum
  
  class StdFd(IntEnum):
      stdout = 0
      stderr = 1
      stdeof = 2
  
  async def call_with_env(cmd: Iterable[str], *, env: Dict[str, str]) -> AsyncIterator[Tuple[StdFd, bytes]]:
      process = await asyncio.create_subprocess_exec(
          *cmd,
          stdout=asyncio.subprocess.PIPE,
          stderr=asyncio.subprocess.PIPE,
          env=env
      )
  
      async def read_stream(stream, fd):
          while True:
              line = await stream.readline()
              if not line:
                  break
              yield (fd, line)
          yield (StdFd.stdeof, b"")
  
      # 并发处理两个输出流
      tasks = [
          asyncio.create_task(read_stream(process.stdout, StdFd.stdout)),
          asyncio.create_task(read_stream(process.stderr, StdFd.stderr))
      ]
      
      hits = 0
      while hits < 2 and tasks:
          done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
          for task in done:
              async for item in task:
                  if item[0] == StdFd.stdeof:
                      hits += 1
                  else:
                      yield item
              tasks.remove(task)
      
      await process.wait()
      yield (StdFd.stdeof, b"")
  

内容的提问来源于stack exchange，提问作者lmonninger