Python多进程调用Dialogflow出现非确定性Segmentation fault问题求助

本次非确定性段错误由gRPC客户端与fork模式多进程的兼容性缺陷导致：

• Google Cloud官方SDK的底层通信基于gRPC实现，gRPC核心逻辑为C语言编写的原生扩展模块
• Linux/Unix环境下Python multiprocessing默认使用fork模式创建子进程，会完整拷贝父进程的内存空间，包括父进程中已经完成初始化的gRPC连接通道对象
• 拷贝到子进程中的gRPC通道处于非法不可用状态，发起调用时就会触发随机段错误；线程模式不存在独立内存空间拷贝，因此不会触发该问题

你提供的复现代码中，在主进程提前初始化了dialogflow.SessionsClient实例，是触发故障的直接诱因。

以下方案可独立使用，也可组合使用提升稳定性：

• 调整客户端初始化位置
  所有Dialogflow相关客户端（包括SessionsClient、AgentsClient）的实例化逻辑全部移动到子进程的执行函数内部，禁止在主进程预创建任何gRPC相关的客户端实例。你复现代码里主进程创建的未使用的session_client需要删除或移入子进程逻辑。
• 修改多进程启动模式为spawn
  spawn模式会启动完全独立的全新Python解释器进程，不会继承父进程的内存状态，从根源避免非法gRPC通道问题。仅需在代码导入multiprocessing后添加一行配置即可：

import multiprocessing
multiprocessing.set_start_method('spawn')

• 升级依赖版本
  你当前使用的google-cloud-dialogflow==2.7.1、google-api-core==2.0.1版本较旧，可升级到对应最新稳定版本，修复官方已知的兼容性bug。

import faulthandler
import os
import shutil
import multiprocessing
from multiprocessing import Process
from time import sleep

import google.cloud.dialogflow as dialogflow

# 强制使用spawn模式启动多进程
multiprocessing.set_start_method('spawn')

os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = 'credentials.json'  # USE YOUR OWN
SESSION_ID = '0' 
PROJECT_ID = '####'    # USE YOUR OWN
AGENT_NAME = '####'    # USE YOUR OWN
DIALOGFLOW_PROJECT_ID = '###'    # USE YOUR OWN

faulthandler.enable(all_threads=True)

os.makedirs('dir_foo', exist_ok=True)
shutil.make_archive(f'agent_foo', 'zip', f'dir_foo')  # 生成空zip文件模拟agent


def create_agent(language):
    print('Creating agent', AGENT_NAME)

    parent = "projects/" + PROJECT_ID
    agents_client = dialogflow.AgentsClient()

    agent = dialogflow.Agent(
        parent=parent, display_name=AGENT_NAME, default_language_code=language, time_zone="GMT"
    )

    print('Creating new agent')
    agents_client.set_agent(request={"agent": agent})

    return agents_client


def upload_bot(language):
    for i in range(5):
        agent = create_agent(language)
        encoded_string = open(f"agent_foo.zip", "rb").read()

    request = {
        "parent": f'projects/{PROJECT_ID}',
        "agent_content": encoded_string
    }
    print('Uploading...')
    for i in range(5):
        agent.restore_agent(request)
    print('Successfully updated bot!')


def execute():
    # 所有客户端初始化移到子进程内部
    session_client = dialogflow.SessionsClient()
    session = session_client.session_path(DIALOGFLOW_PROJECT_ID, SESSION_ID)
    try:
        upload_bot('en')
    except Exception as e:
        print(e)


processes_list = list()
for i in range(16):
    t = Process(target=execute)
    t.start()
    processes_list.append(t)
    sleep(1)

for c in range(len(processes_list)):
    processes_list[c].join()

print('DONE')

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