Airflow 3 中 Python multiprocessing 完成后 Task 卡在 running
结论
问题更可能出在 Python 子进程或 multiprocessing.Manager 没有正常退出。现有信息不足以说明 Airflow 漏掉了大量 process signals。
XML 处理完成，只代表 multiprocessing_xml_handler() 中的业务代码已经执行完毕。Airflow 要等整个 Task Runner 进程退出并上报终态，才会把 Task 从 running 改为 success。当前代码中的日志：
logger.info("Finished processing the XML data")位于 Pool 上下文内部。日志出现时，程序还没有执行 Pool.__exit__() 和 Manager.__exit__()。业务处理可能已经结束，程序却在之后的 Pool 清理、子进程回收、Manager 关闭、锁释放或日志刷新阶段卡住。
代码还存在一个需要优先修正的问题：它先通过默认上下文创建 Manager()，随后又使用 get_context() 创建 Pool，可能混用了不同 multiprocessing context 创建的对象。这个问题在 spawn 模式下更容易暴露。
为什么单独运行正常，放进 Airflow 就会卡住
Airflow Task 的运行环境比独立 Python 脚本复杂。运行 Task 的进程通常已经初始化了日志、数据库连接、线程、信号处理器和通信管道。
在这样的进程中使用 fork，子进程会继承父进程的内存状态，却不会完整复制父进程中的其他线程。如果 fork 时某个锁正被另一个线程持有，子进程可能一直等待该锁。常见对象包括：
- Python logging handler 的锁；
- SQLAlchemy 连接池和数据库 socket；
- OpenSSL、XML 库或其他 C 扩展的内部锁；
- Airflow Task Runner 使用的管道和文件描述符；
multiprocessing.Manager的代理连接。
解析器在普通 Linux 脚本中运行正常，却在 Airflow 中出现问题，原因可能就是两种运行方式的父进程状态不同。
spawn 不会继承这些内存状态，一般更适合从复杂运行时中创建子进程。不过，传给 worker 的函数和参数必须能够被 pickle，相关模块也必须能由子进程安全地重新导入。下面这些情况都可能导致 spawn 启动失败：
- worker 函数不是 module 顶层可导入对象；
- 参数中包含不能 pickle 的对象；
- DAG 文件或被导入模块在顶层直接创建 Pool；
- standalone 入口缺少
if __name__ == "__main__":； - 子进程导入 DAG 模块时触发额外的初始化代码；
- 混用不同 multiprocessing context 创建的 Lock、Queue、Manager 或 Pool。
所以，“fork 卡住”和”spawn 崩溃”通常是两个独立问题，不能只靠调整 Airflow 配置并在两种启动方式之间切换来解决。
先修正 Pool 的生命周期
可以暂时去掉嵌套的上下文管理器，显式调用 close()、join() 和 terminate()，以便确认程序究竟卡在哪一步。
import multiprocessing as mp
import logging
logger = logging.getLogger(__name__)
_worker_lock = None
def init_worker(lock):
global _worker_lock
_worker_lock = lock
def multiprocessing_xml_handler(args):
chunk, table_query, csv_path, threshhold_processes = args
chunk_xml_dataframes = {}
for xml in chunk:
try:
xml_dataframes = xml_reader.process_xml(
xml,
table_query,
csv_path,
_worker_lock,
)
for csv_file, result_dict in xml_dataframes.items():
if csv_file in chunk_xml_dataframes:
current = chunk_xml_dataframes[csv_file]
chunk_xml_dataframes[csv_file] = ToolBox.dict_extend(
result_dict,
current,
)
else:
chunk_xml_dataframes[csv_file] = result_dict
except Exception:
logger.exception("Error while processing XML: %s", xml)
raise
ToolBox.flush_to_csv(chunk_xml_dataframes, _worker_lock)
logger.info("Worker flushed its data")
def parse_xml(
output_folder: str,
input_folder: str,
config_path: str | None = None,
):
ctx = mp.get_context("spawn")
lock = ctx.Lock()
arguments = [
(chunk, table_query, csv_path, threshhold_processes)
for chunk in chunks
]
pool = ctx.Pool(
processes=pool_size,
initializer=init_worker,
initargs=(lock,),
)
try:
logger.info("Submitting %d XML chunks", len(arguments))
for _ in pool.imap_unordered(
multiprocessing_xml_handler,
arguments,
):
pass
logger.info("All worker results received")
pool.close()
logger.info("Waiting for worker processes to exit")
pool.join()
logger.info("All worker processes exited")
except BaseException:
logger.exception("XML multiprocessing failed")
pool.terminate()
pool.join()
raise这个版本做了以下调整：
- Lock 和 Pool 都由同一个
ctx创建。 - worker 放在 module 顶层，方便
spawn导入和 pickle。 - Lock 通过 Pool initializer 注入，不再随每个任务重复传入。
Finished类日志移到pool.join()之后。- 子进程异常会重新抛出，不会只写入日志后继续运行。
- 日志能够区分”结果已收齐”和”子进程已退出”。
如果确实需要 Manager，也应使用同一个 context 创建：
ctx = mp.get_context("spawn")
with ctx.Manager() as manager:
lock = manager.Lock()
# 使用同一个 ctx 创建 Pool从当前代码看，这里可能只需要一个跨进程锁。可以优先使用 ctx.Lock()，不必额外启动 Manager server process。
检查 spawn 的导入和 pickle 问题
独立运行入口应保留 main guard：
def main():
parse_xml(
output_folder="/path/to/output",
input_folder="/path/to/input",
)
if __name__ == "__main__":
main()DAG 文件及其导入的模块不应在 module 顶层调用 parse_xml()、创建 Manager 或启动 Pool。顶层代码只用于定义 DAG、Task 和可导入函数。
可以先单独验证 worker 和参数是否能够被 pickle：
import pickle
pickle.dumps(multiprocessing_xml_handler)
pickle.dumps(arguments[0])这种检查不能覆盖所有 spawn 问题，但能快速发现常见的不可序列化对象。如果 spawn 持续崩溃，应先保留第一次出现的完整 traceback。后续异常可能只是子进程反复启动失败引起的连锁反应。
确认程序卡在哪个退出阶段
在每个资源清理点前后添加日志：
logger.info("Before pool.close()")
pool.close()
logger.info("After pool.close()")
logger.info("Before pool.join()")
pool.join()
logger.info("After pool.join()")如果仍然使用 Manager，也要记录 Manager shutdown 前后的状态：
logger.info("Before manager shutdown")
manager.shutdown()
logger.info("After manager shutdown")根据最后一条日志，通常可以缩小排查范围：
- 卡在
imap_unordered()：至少有一个任务没有返回，或者结果管道未能正常传递数据。 - 卡在
pool.join()：某个 worker、Pool helper thread 或 worker 中的非守护线程没有退出。 - Pool 已退出，但程序在关闭 Manager 前卡住：检查是否仍有 Manager proxy、Lock 或连接被持有。
- 函数已经 return，但 Task 仍是
running：检查 Task Runner 中残留的线程、子进程、文件描述符，以及 Airflow 的终态上报链路。
在 Linux 上查看进程状态
先找到执行该 Task 的父进程，再查看完整的父子进程关系：
ps -eo pid,ppid,pgid,sid,stat,wchan:32,etime,cmd --forest如果系统提供 pstree：
pstree -ap <task-runner-pid>重点查看以下状态：
Z：子进程已经退出，但父进程尚未回收。D：进程正在等待不可中断的 I/O，可能与文件系统或磁盘有关。S且wchan为futex_*：通常表示进程正在等待线程锁或进程锁。wait_*：父进程正在等待子进程退出。pipe_*、poll_*：进程可能正在等待 Pool、Manager 或日志管道。- XML worker 已经消失，但 Manager、resource tracker 或其他子进程仍然存活。
查看进程持有的文件描述符：
ls -l /proc/<pid>/fd如果安装了 lsof：
lsof -p <pid>这些信息可以帮助定位尚未关闭的 pipe、socket、日志文件或临时文件。
在测试环境中，还可以短时间附加 strace：
strace -ff \
-p <pid> \
-e trace=process,signal,futex,wait4,read,write \
-s 128可以按下面的方式判断：
- 持续停在
futex()：很可能是锁或线程清理问题。 - 持续停在
wait4()：父进程仍在等待某个子进程。 - 持续读取某个 pipe：Pool、Manager 或日志通道可能没有关闭。
- 反复收到信号并创建子进程：检查
spawn导入模块时是否重复执行了启动代码。
strace 会影响进程时序，适合在开发环境中使用。
输出 Python 线程栈
如果进程无法退出，又无法从进程状态看出它在等待什么，可以启用 faulthandler：
import faulthandler
import signal
faulthandler.enable()
faulthandler.register(signal.SIGUSR2, all_threads=True)进程卡住后，向 Task Runner 或目标 worker 发送：
kill -USR2 <pid>Python 会把所有线程的堆栈写入标准错误或 Task 日志。通过堆栈通常可以判断主线程卡在 pool.join()、Manager shutdown、logging flush，还是某个业务锁中。
如果环境允许安装并附加 py-spy，也可以执行：
py-spy dump --pid <pid>py-spy 展示 Python 调用栈时通常比系统调用跟踪更直观，但能否附加取决于主机的 ptrace 权限。
检查日志和异常处理
除了 Task 日志，还要检查运行 Task 的组件日志，以及 scheduler 和 executor 的相关日志。重点查找：
- worker 的真实退出码；
BrokenPipeError、EOFError或ConnectionResetError；- pickle 和 import 错误；
daemonic processes are not allowed to have children；resource_tracker关于 semaphore 或 shared memory 的警告；- Task Runner 是否已经退出；
- executor 是否收到 Task 的终态；
- SIGTERM、SIGKILL、OOM 或系统级 abort 记录。
当前 worker 的异常处理会掩盖失败：
except Exception as e:
logger.error(f"Error:{e} while processing: {xml}")这段代码只记录异常文本，没有 traceback，也不会让 Pool 任务失败。至少应改为：
except Exception:
logger.exception("Error while processing XML: %s", xml)
raise否则，某个库的内部异常可能被误判成”XML 已经处理完成”。
还应确认操作系统是否杀死过相关进程：
journalctl -k重点搜索 OOM、segfault、killed process 和 denied 等信息。解析大量 XML 时，多进程会增加内存占用。如果多个 LocalExecutor Task 又分别创建 Pool，实际进程数和内存消耗可能远高于 pool_size。
Airflow 配置不能替代应用自身的 multiprocessing context
Airflow 的 mp_start_method 等配置主要控制 Airflow 自己创建进程时使用的启动方式。应用代码显式调用：
ctx = multiprocessing.get_context("spawn")应用 Pool 的启动方式就由这个 ctx 决定。Airflow 配置与它有关，但修改 Airflow 配置不会自动修复嵌套 Pool。
execute_tasks_new_python_interpreter 会改变 Task 的启动环境，也无法解决 worker 不能被 pickle、Manager context 混用或 Pool 无法 join 等问题。
排查时可以先固定以下条件：
- 保持 LocalExecutor 配置不变。
- 应用统一使用
spawn。 - 所有 Lock、Queue、Manager 和 Pool 都通过同一个
ctx创建。 - 先解决
spawn的第一条完整 traceback。 - 确认
pool.join()和 Manager shutdown 都能返回。 - 再比较其他启动方式。
不要同时修改多个 Airflow 参数和应用代码，否则很难判断是哪项改动影响了结果。
是否可能是 Airflow 没有处理好大量信号
进程管理程序确实可能遇到退出信号、子进程回收或状态上报问题，但 Task 停在 running 本身不能证明 Airflow 被大量 signals 压垮。
要支持这个判断，至少需要以下一项证据：
- worker 和父进程都已退出，但 executor 没有收到终态；
- Airflow 日志明确记录了异常退出码或信号处理错误；
- 进程表中已经没有相关 Task 进程，数据库状态却长期保持
running； - 相同代码在不同进程数量下，能够稳定复现信号或回收异常。
如果 Task Runner 仍然存在，并卡在 futex()、wait4()、pipe read 或 pool.join()，问题还在 Python 进程内部。此时 Airflow 尚未获得把 Task 标记为成功的机会。
更稳妥的处理方式
如果每个 worker 最终都要争用同一个锁来写入 CSV，串行写入可能抵消多进程带来的收益。可以让 worker 只负责解析 XML，并分别写入独立的临时文件，再由父进程合并结果。这样就能去掉共享 Manager 和跨进程写锁。
数据量允许时，也可以让 worker 返回结果，再由父进程统一写入：
results = []
for result in pool.imap_unordered(parse_chunk, arguments):
results.append(result)
pool.close()
pool.join()
for result in results:
write_result(result)返回大对象会增加进程间序列化和内存开销。数据量较大时，使用独立临时文件一般更合适。
还可以为 Airflow Task 设置执行超时，防止故障发生后长期占用 LocalExecutor slot。超时只能用于兜底，不能替代对子进程退出问题的修复。
备注：内容仅供参考。