使用Ape框架+ProcessPoolExecutor遇PicklingError：无法序列化函数

错误详情 #

_pickle.PicklingError: Can't pickle <function fetchTokensWithIndex at 0x7f217c9b1a20>: it's not the same object as scripts.getAllTokensList.fetchTokensWithIndex

问题背景 #

• 测试过multiprocessing、pathos.multiprocessing、concurrent.futures.ProcessPoolExecutor，均触发相同错误
• 多进程代码在全局作用域执行正常，但放到函数内部调用就报错
• ThreadPoolExecutor可正常运行，排除代码逻辑本身问题
• 基于Ape框架开发，通过命令ape run scripts/getAllTokensList.py启动，从main函数开始执行

相关代码 #

import csv
from concurrent.futures import ThreadPoolExecutor, as_completed, ProcessPoolExecutor
from re import findall
from typing import Dict, Optional

from ape_ethereum import multicall
from loguru import logger

from scripts.Classes import Token
from scripts.colours import *
from scripts.contracts import *
from scripts.utils import *

def fetchTokensWithIndex(uniswap_v1_factory, start_id: int, end_id: int) -> list:
    call = multicall.Call()
    # print(f"Proceesing {end_id + 1 - start_id} number of tokens")
    for i in range(start_id, end_id + 1):
        call.add(uniswap_v1_factory.getTokenWithId, i)
    return list(call())

def batchGetUniswapAllTokens():
    [...]
    chunk_size = 200
    multicall_results = []
    num_threads = 5

    with ProcessPoolExecutor(max_workers=num_threads) as executor:
        for start_id in range(1, token_count + 1, chunk_size):
            end_id = min(start_id + chunk_size - 1, token_count)
            futures = executor.submit(
                fetchTokensWithIndex, uniswap_v1_factory, start_id, end_id
            )
            multicall_results.append(futures)

    final_results = []
    for future in as_completed(multicall_results):
        final_results.extend(future.result())

def main():
    batchGetUniswapAllTokens()

原因分析 #

Pickle序列化多进程任务时，要求被传递的函数在子进程中能通过模块路径被识别为同一个对象。当函数处于非全局作用域（或框架启动后函数的模块引用路径发生变更）时，子进程无法匹配到完全一致的函数对象，从而触发该错误。

结合Ape框架开发者的官方回复：

该框架可能对多进程支持不佳，我们尚未对此进行测试。另外，由于Ape脚本涉及大量网络I/O，使用异步处理通常是加速脚本的更好方案。

解决方案 #

• 优先改用异步处理：由于脚本以网络I/O为主，异步处理（如Ape框架自带的异步能力、asyncio）能更高效利用资源，同时规避多进程的序列化问题
• 调整多进程代码作用域：若必须使用多进程，将ProcessPoolExecutor的调用逻辑移至全局作用域，确保子进程能正确识别fetchTokensWithIndex函数
• 避免传递框架复杂对象：检查uniswap_v1_factory等框架对象是否可安全序列化，若不可行，在子进程内重新初始化该对象，而非从主进程传递

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