Python多进程返回值异常：监控后程序卡死问题排查

问题根源分析 #

1. 执行成功场景：check进程检测到my_fun进程结束后，仅打印信息并返回True，但未向check_q存入任何值，导致主进程在check_q.get()处无限阻塞。
2. 超时/内存超限场景：主进程执行my_fun_q.get()时，若my_fun已被杀死且未向队列存值，会触发无限阻塞；同时进程join的顺序逻辑未匹配队列获取的时机。

具体修复步骤 #

• 在check函数的执行成功分支中，添加q.put(True)，确保主进程能获取到check的返回状态。
• 主进程对队列get()操作添加超时时间，避免无限阻塞；仅当check返回成功时，才尝试获取my_fun的结果。
• 修正内存判断的数值对齐（将minimum_available_memory调整为用户预期的2GB）。
• 给my_fun添加异常捕获，处理进程被杀死时的潜在异常。

修复后的完整代码 #

import os
import time
import psutil
from multiprocessing import Process, SimpleQueue


def my_fun(a, b, c, q):
    l = []
    i = 2
    try:
        for _ in range(a * b * c):
            l.append(i ** 50)
            i *= 1024

        q.put((a, b, c))
    except Exception:
        # 进程被杀死时捕获异常，避免无处理报错
        pass


def check(pid, available_mem, start_time, allowed_time, q):
    try:
        while True:
            if not psutil.pid_exists(pid):
                print('Function executed successfully')
                q.put(True)
                return

            available_memory = psutil.virtual_memory()[1]
            if available_memory < available_mem * (10 ** 9):
                os.kill(pid, 9)
                q.put(False)
                print('Memory usage violation')
                break

            if time.time() - start_time > allowed_time:
                os.kill(pid, 9)
                q.put(False)
                print('Time out')
                break

            time.sleep(1)
    except Exception as e:
        q.put(False)
        print(e)


if __name__ == "__main__":
    minimum_available_memory = 2  # 调整为用户预期的2GB阈值
    time_allowed = 5  # 允许的执行时长（秒）

    my_fun_q = SimpleQueue()
    check_q = SimpleQueue()
    arg1, arg2, arg3 = 1, 1, 1
    # arg1, arg2, arg3 = 100000,100000,100000  # 可切换测试超时场景

    my_fun_process = Process(target=my_fun, args=(arg1, arg2, arg3, my_fun_q))
    start_time = time.time()
    my_fun_process.start()
    pid = my_fun_process.pid

    check_process = Process(target=check, args=(pid, minimum_available_memory, start_time, time_allowed, check_q))
    check_process.start()

    # 获取监控结果，设置超时避免阻塞
    try:
        check_result = check_q.get(timeout=time_allowed + 2)
    except:
        check_result = False
        print("Check process timeout or failed")

    # 等待子进程完成清理
    my_fun_process.join()
    check_process.join()

    # 输出结果，仅在监控成功时尝试获取函数返回值
    print(check_result)
    if check_result:
        try:
            my_fun_result = my_fun_q.get(timeout=1)
            print(my_fun_result)
        except:
            print("Failed to get my_fun result")

修复说明 #

• check函数补充了成功状态的队列写入，解决主进程阻塞问题。
• 队列get()操作添加超时逻辑，彻底避免无限阻塞风险。
• 仅在确认my_fun正常执行完成时，才尝试获取其返回值，匹配业务逻辑。

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