Python ProcessPoolExecutor future.result()永久阻塞问题

问题描述 #

学习David Beazley并发编程相关内容时，运行编写的斐波那契微服务服务端与配套客户端代码，从客户端提交数字20的计算请求后，服务端调用future.result()获取进程池任务返回值时永久阻塞，服务端直接冻结无响应，无法定位问题成因。

相关代码 #

服务端代码 server.py #

# server.py
# Fib microservice

from socket import *
from fib import fib
from threading import Thread
from concurrent.futures import ProcessPoolExecutor as Pool

pool = Pool(4)

def fib_server(address):
    sock = socket(AF_INET, SOCK_STREAM)
    sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    sock.bind(address)
    sock.listen(5)
    while True:
        client, addr = sock.accept()
        print("Connection", addr)
        Thread(target=fib_handler, args=(client,), daemon=True).start()

def fib_handler(client):
    while True:
        req = client.recv(100)
        if not req:
            break
        n = int(req)
        future = pool.submit(fib, n)
        # 下一行代码会触发阻塞
        result = future.result()
        resp = str(result).encode('ascii') + b'\n'
        client.send(resp)
    print("Closed")

fib_server(('',25000))

客户端代码 client.py #

#client.py
import socket

s = socket.socket()
s.connect(('localhost',25000))
while True:
    num=input("number?")
    s.send(str(num).encode('ascii') + b'\n')
    res = s.recv(1000)
    print('res:',res)

复现步骤 #

• 服务端执行命令：python server.py
• 客户端执行命令：python client.py

复现现象 #

• 服务端打印日志：Connection ('127.0.0.1', 57876)
• 客户端出现输入提示number?时输入20
• 服务端直接冻结无响应

根因分析 #

• ProcessPoolExecutor基于多进程实现，Python多进程默认使用spawn启动模式（Windows、新版macOS默认，Linux可配置），该模式下子进程启动时会重新导入主模块（即server.py所在模块）的全部顶层代码。
• 当前代码把进程池初始化pool = Pool(4)、服务启动调用fib_server(('',25000))直接写在模块顶层，没有加主模块判断保护。子进程启动时会重新执行这些顶层代码，递归创建新的进程池、重复尝试绑定25000端口，最终触发进程启动死锁：所有进程池子进程无法正常执行fib计算任务，主进程调用future.result()永远等不到返回值，就会永久阻塞。

修复方案 #

将进程池初始化、服务启动逻辑放到if __name__ == '__main__':判断块内，保证这部分逻辑仅在主进程中执行，子进程导入主模块时不会重复触发。
修复后的server.py代码如下：

# server.py
# Fib microservice

from socket import *
from fib import fib
from threading import Thread
from concurrent.futures import ProcessPoolExecutor as Pool

pool = None

def fib_server(address):
    sock = socket(AF_INET, SOCK_STREAM)
    sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    sock.bind(address)
    sock.listen(5)
    while True:
        client, addr = sock.accept()
        print("Connection", addr)
        Thread(target=fib_handler, args=(client,), daemon=True).start()

def fib_handler(client):
    while True:
        req = client.recv(100)
        if not req:
            break
        n = int(req)
        future = pool.submit(fib, n)
        result = future.result()
        resp = str(result).encode('ascii') + b'\n'
        client.send(resp)
    print("Closed")

if __name__ == '__main__':
    pool = Pool(4)
    fib_server(('',25000))

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