Python contextlib.chdir 为什么不适合并发任务共享

如果一个协程需要临时进入某个目录，contextlib.chdir 写起来很顺手；但它改变的是整个 Python 进程的当前工作目录，不是某个 asyncio.Task 的私有变量。多个任务共享它时，真正危险的不是“能不能进入和退出”，而是任务在目录已经改变后执行了 await，另一个任务就有机会改写同一个全局状态。

结论很明确：并发任务不要把 contextlib.chdir 当作局部上下文共享。优先把目录变成 Path 参数，直接构造目标文件；如果旧接口只能读取当前目录，就把切换目录限制在短小的同步区段，并用同一个锁保护所有调用方。

官方文档：https://docs.python.org/3/library/contextlib.html#contextlib.chdir

相关文档：https://docs.python.org/3/library/asyncio-task.html

问题不在 with，而在 cwd 是进程级状态

with chdir(path) 的进入动作会调用目录切换，退出动作再恢复进入前的目录。这个保存和恢复过程可以嵌套，所以官方文档称它是可重入的；但“可重入”只说明同一条控制路径能够嵌套使用，不等于多个并发任务可以交错使用。

相对路径也不会记住“是谁创建了它”。当代码执行 open("result.json") 时，解释器会根据当时的进程 cwd 解析目标。目录一旦被别的任务改写，调用本身仍然合法，却可能读写另一个目录下的文件。

一个 await 就能把隐式目录暴露给别的任务

下面的例子模拟两个任务分别处理两个目录。代码没有线程，也没有显式改动共享变量，但 asyncio 任务会在 await 处把执行机会交给事件循环。官方文档对这种协作式调度的描述，正好解释了为什么两个任务能够在同一个线程里交错。

import asyncio
from contextlib import chdir
from pathlib import Path

async def read_result(root: Path, pause: float) -> Path:
    with chdir(root):
        # 目录已经是进程级状态，暂停前不要假设它仍归当前任务。
        await asyncio.sleep(pause)
        # 这个相对路径会读取“当前时刻”的 cwd，而不是 root。
        return Path.cwd() / "result.json"

async def main() -> None:
    # 两个任务共享同一个事件循环，但目标目录不同。
    first, second = await asyncio.gather(
        read_result(Path("/srv/job-a"), 0.02),
        read_result(Path("/srv/job-b"), 0.00),
    )
    print(first, second)

asyncio.run(main())

这段代码的具体输出会受到调度时机影响，但风险结构不变：任务 A 进入目录 A 后暂停，任务 B 进入目录 B；任务 A 恢复时看到的 cwd 可能已经是 B。任务 A 退出时还会按照自己的保存值恢复目录，因而任务 B 后续的相对路径也可能再次漂移。

这里不应该把某一次输出当成稳定复现条件。排查时应关注代码结构：chdir 的作用域内是否存在 await、生成器让出执行权、回调或第三方库调用。如果存在，就应该把共享 cwd 视为竞态入口。

可重入不等于并发安全

把目录切换想象成一个栈，可以理解“可重入”和“并发安全”的差别。嵌套调用时，外层目录 A、内层目录 B 能按相反顺序恢复；但两个任务交错时，栈上的进入和退出并不再属于一条连续的控制路径。每个任务都以为自己拥有 cwd，实际却在读写同一个进程状态。

因此，以下几种想法都不够：

• 只要用了 with 就会自动隔离任务：with 只能负责退出时恢复，不能复制进程状态。
• 当前只有一个线程所以安全：asyncio 的任务即使串行运行，也会在 await 处交错。
• 把 chdir 放到工具函数里就安全：只要工具函数仍然依赖相对路径，它仍然依赖全局 cwd。

把目录改成显式 Path，先解决主要架构问题

更稳的设计是让任务入口接收一个目录对象，并在调用点拼出完整目标。这样每个任务携带自己的输入，文件 API 不需要读取隐式 cwd；任务之间可以并发，目录也可以来自不同挂载点或临时目录。

import asyncio
from pathlib import Path

async def read_result(root: Path, pause: float) -> Path:
    # root 是当前任务自己的数据，不写入进程全局状态。
    await asyncio.sleep(pause)
    result_path = root / "result.json"
    # Path.open 直接打开明确目标，避免相对路径随 cwd 漂移。
    with result_path.open("rb") as stream:
        stream.read(1)
    return result_path

async def main() -> None:
    # 两个任务可以交错等待，但各自的 Path 不会互相覆盖。
    paths = await asyncio.gather(
        read_result(Path("/srv/job-a"), 0.02),
        read_result(Path("/srv/job-b"), 0.00),
    )
    print(paths)

asyncio.run(main())

文件写入、配置查找、模板读取和临时产物都可以沿用这个原则。若需要临时目录，可在任务边界创建自己的目录并把它继续作为参数传给下层函数；不要在深层函数里偷偷调用 chdir，否则调用者很难判断它是否修改了整个进程。

旧接口必须依赖 cwd 时，怎样做有限兼容

有些遗留库只接受文件名，内部还会调用相对路径。这时可以使用同一个 asyncio.Lock，把切换目录和旧接口调用包在一个很短的同步区段内。关键点是：锁必须由所有访问该全局 cwd 的调用方共享，而且持锁期间不要 await。

import asyncio
from contextlib import chdir
from pathlib import Path

cwd_lock = asyncio.Lock()

def legacy_export() -> None:
    # 旧接口只接受相对文件名，并在当前目录生成 export.bin。
    Path("export.bin").write_bytes(b"data")

async def run_legacy(root: Path) -> Path:
    async with cwd_lock:
        # 锁只覆盖同步旧接口，避免持锁等待导致其他任务长期饥饿。
        with chdir(root):
            legacy_export()
            return root / "export.bin"

async def main() -> None:
    # 所有 run_legacy 调用共享同一把锁，避免 cwd 重叠。
    results = await asyncio.gather(
        run_legacy(Path("/srv/job-a")),
        run_legacy(Path("/srv/job-b")),
    )
    print(results)

asyncio.run(main())

这个方案只是兼容层，不是把 chdir 变成并发安全 API。它只在同一个事件循环内、所有调用方都遵守同一把锁时成立；如果还有线程、插件或未接入锁的代码修改 cwd，保护就会失效。若旧接口还需要长时间计算，优先改成显式路径，或把整个任务拆到独立进程，并通过进程启动参数指定 cwd。

上线前用这份清单排查 cwd 风险

最后可以把判断压缩成一句话：contextlib.chdir 适合短小的脚本式同步代码，不适合让多个并发任务共享隐式工作目录。目录是输入时就显式传递，旧接口无法修改时才用最小兼容层，并把它当作迁移债务而不是架构能力。

常见问题

contextlib.chdir 是不是线程安全的？

不是。它改变进程的当前工作目录，官方文档明确标注为 non parallel-safe，并提醒不适合大多数线程或异步上下文。

只在 chdir 里面调用同步函数，还需要担心吗？

如果同步函数期间没有其他线程或任务观察、修改 cwd，风险会小很多；但共享库和插件边界仍需确认。更稳的长期方案仍是显式路径。

asyncio.Lock 能让 contextlib.chdir 彻底安全么？

不能。它只能约束同一事件循环中、主动使用同一把锁的调用方；未接入锁的线程、插件或第三方代码仍可能改写 cwd。

为什么 Path 参数比先 chdir 再 open 更适合并发？

Path 参数属于调用数据，跟随当前任务传递；open("file") 则依赖共享的隐式进程状态。显式数据流更容易测试，也更容易在任务之间隔离。