Python多进程Reduce函数在任务队列超1200元素时挂起问题排查

当to_sum列表元素数量约在1150以内时程序运行正常；超过该值后，进程会在首次执行task_queue = result_queue时挂起。进程能成功填充结果队列并终止，但随后陷入挂起状态，数组规模低于1150则无此问题。重启电脑后可支持的队列上限会在1100-1300区间波动，但仍会触发该问题。

import multiprocessing

class CustomProcess(multiprocessing.Process):

    def __init__(self, name, task_queue, result_queue, lock, chunks=2, *args, **kwargs):
        super().__init__(name=name, *args, **kwargs)
        self.name = name
        self.task_queue = task_queue
        self.result_queue = result_queue
        self.chunks = chunks
        self.lock = lock

    def run(self):
        while True:
            """
            Using a lock to avoid a race condition where 2 threads both get a number, then
            both to get another, but it is None so they both put the number back resulting in a result queue
            with bigger size. For example:
            Expected result_queue_size = 500, current_queue_size 499, after summation of 1 and 2 we will add 3 to result queue
            achieving 500.
            With race condition result is 501.
            [1, 2, None, None]
            Thread 1 gets 1.
            Thread 2 gets 2.
            Thread 1 gets None and puts 1 in result queue instead of getting 2 and summing.
            Thread 2 gets None and puts 2 in result queue instead of just getting the first None and returning.
            A lock on both gets removes the race condition.
            """
            with self.lock:
                if not self.task_queue.empty():
                    number_1 = self.task_queue.get()
                    self.task_queue.task_done()
                    if number_1 is None:
                        #Poison pill - terminate.
                        print(f"Terminated {self.name}")
                        return
                else:
                    #Queue empty - terminate.
                    return
                if not self.task_queue.empty():
                    number_2 = self.task_queue.get()
                    self.task_queue.task_done()
                    if number_2 is None:
                        #Cannot compute sum of 1 number so just add number_1 to result_queue and terminate since poison pill
                        #acquired.
                        self.result_queue.put(number_1)
                        print(f"Terminated {self.name}")
                        return
                else:
                    self.result_queue.put(number_1)
                    #Queue empty, put the 1 number in result queue and terminate.
                    return
            self.result_queue.put(number_1 + number_2)


def multiprocess_sum(array):
    if len(array) == 1:
        return array[0]
    lock = multiprocessing.Lock()
    task_queue = multiprocessing.JoinableQueue()
    [task_queue.put(element) for element in to_sum]
    task_queue_size = len(array)

    while task_queue_size > 1:
        print(task_queue.qsize(), task_queue_size)
        result_queue = multiprocessing.JoinableQueue()
        processes = [CustomProcess(name=str(i), task_queue=task_queue, result_queue=result_queue, lock=lock)
                    for i in range(8)]
        [task_queue.put(None) for process in processes]
        [process.start() for process in processes]
        #[process.join() for process in processes]
        task_queue.join()
        task_queue = result_queue
        task_queue_size = task_queue_size // 2 + task_queue_size % 2
    return result_queue.get()

if __name__ == "__main__":
    to_sum = [i for i in range(1350)]
    """
    If range is below 1200, the program will run and compute everything correctly.
    If it is above it, it will hang at the first halving, the moment the first task_queue is empty and the
    result_queue becomes the new task_queue. 
    Computer restart will make the range values fluctuate, yesterday it would hang at 1177 but run fine up to 1776.
    Queue pipe full???
    """
    print(sum(to_sum))
    for i in range(5):
        print(multiprocess_sum(to_sum))

核心原因 #

1. 队列管道缓冲区阻塞：JoinableQueue基于系统管道实现，缓冲区大小有限。当子进程向result_queue写入大量数据的速度超过主进程处理速度时，管道会被填满，子进程会阻塞在put()操作上。数据量较小时，缓冲区能容纳所有数据，子进程可快速完成写入；数据量过大时，缓冲区溢出导致子进程阻塞，主进程却已开始切换队列，最终引发死锁。
2. 未等待子进程完全退出：代码中注释掉了process.join()，虽然task_queue.join()会等待所有任务标记为完成，但子进程可能还在执行向result_queue写入的操作。主进程直接将result_queue赋值为新的task_queue，此时子进程可能仍阻塞在put()，导致整个程序挂起。
3. 函数依赖全局变量：multiprocess_sum函数中使用了全局变量to_sum而非参数array，这不仅破坏了封装性，还可能在多轮调用时引发意外的状态问题。

修复方案 #

1. 恢复子进程等待逻辑 #

在task_queue.join()之后，添加process.join()等待所有子进程完全退出，确保它们完成所有result_queue的写入操作：

[process.start() for process in processes]
task_queue.join()
# 等待所有子进程退出
[process.join() for process in processes]
task_queue = result_queue

2. 修正全局变量依赖 #

将multiprocess_sum中的to_sum替换为参数array：

[task_queue.put(element) for element in array]

3. 可选：增大队列缓冲区 #

创建JoinableQueue时指定较大的maxsize，减少缓冲区溢出的概率：

task_queue = multiprocessing.JoinableQueue(maxsize=10000)
result_queue = multiprocessing.JoinableQueue(maxsize=10000)

修复后的完整代码 #

import multiprocessing

class CustomProcess(multiprocessing.Process):

    def __init__(self, name, task_queue, result_queue, lock, chunks=2, *args, **kwargs):
        super().__init__(name=name, *args, **kwargs)
        self.name = name
        self.task_queue = task_queue
        self.result_queue = result_queue
        self.chunks = chunks
        self.lock = lock

    def run(self):
        while True:
            with self.lock:
                if not self.task_queue.empty():
                    number_1 = self.task_queue.get()
                    self.task_queue.task_done()
                    if number_1 is None:
                        print(f"Terminated {self.name}")
                        return
                else:
                    return
                if not self.task_queue.empty():
                    number_2 = self.task_queue.get()
                    self.task_queue.task_done()
                    if number_2 is None:
                        self.result_queue.put(number_1)
                        print(f"Terminated {self.name}")
                        return
                else:
                    self.result_queue.put(number_1)
                    return
            self.result_queue.put(number_1 + number_2)


def multiprocess_sum(array):
    if len(array) == 1:
        return array[0]
    lock = multiprocessing.Lock()
    task_queue = multiprocessing.JoinableQueue(maxsize=10000)
    [task_queue.put(element) for element in array]
    task_queue_size = len(array)

    while task_queue_size > 1:
        print(task_queue.qsize(), task_queue_size)
        result_queue = multiprocessing.JoinableQueue(maxsize=10000)
        processes = [CustomProcess(name=str(i), task_queue=task_queue, result_queue=result_queue, lock=lock)
                    for i in range(8)]
        [task_queue.put(None) for process in processes]
        [process.start() for process in processes]
        task_queue.join()
        # 等待所有子进程完成写入操作
        [process.join() for process in processes]
        task_queue = result_queue
        task_queue_size = task_queue_size // 2 + task_queue_size % 2
    return task_queue.get()

if __name__ == "__main__":
    to_sum = [i for i in range(1350)]
    print(sum(to_sum))
    for i in range(5):
        print(multiprocess_sum(to_sum))

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