如何在独立进程中恢复Python生成器的执行？

一、常规方案：手动管理状态（无需修改解释器） #

原生Python生成器的状态（栈帧、局部变量等）是进程私有内存的一部分，无法直接跨进程序列化。我们可以通过显式提取并传递状态的方式，手动实现跨进程恢复。

实现思路 #

1. 在生成器中加入状态参数，支持从传入的状态中恢复局部变量和执行位置。
2. 第一个进程执行到yield时，将当前状态（如局部变量、执行步骤）打包为可序列化对象（如字典）传递出去。
3. 第二个进程通过传入该状态，初始化生成器并跳过已执行的逻辑，直接从yield后的位置继续执行。

示例代码 #

import multiprocessing

def simple_generator(state=None):
    # 从传入状态恢复上下文
    current_step = state.get('step', 0) if state else 0

    if current_step == 0:
        # 进程一中执行的逻辑
        print("Running in process one")
        # 保存当前状态
        state = {'step': 1}
        yield state
    
    # 进程二中执行的逻辑
    print("Running in process two")

def process_one():
    gen = simple_generator()
    # 执行到yield，获取状态
    state = next(gen)
    return state

def process_two(state):
    # 传入状态，恢复生成器执行
    gen = simple_generator(state)
    # 跳过已执行的步骤，直接进入后续逻辑
    next(gen)

if __name__ == "__main__":
    state = process_one()
    # 在新进程中恢复执行
    p = multiprocessing.Process(target=process_two, args=(state,))
    p.start()
    p.join()

二、底层方案：修改Python解释器支持生成器序列化 #

如果需要原生生成器直接跨进程恢复，可以通过修改Python解释器的底层实现，让生成器状态可被序列化/反序列化。

核心修改方向 #

1. 扩展生成器对象的__reduce__方法，使其能导出关键状态：包括栈帧的f_locals（局部变量）、f_lasti（指令指针位置）、f_code（代码对象）等。
2. 实现对应的反序列化逻辑，在目标进程中重新构建生成器的栈帧结构，恢复执行位置。
3. 注意处理引用类型对象的跨进程传递，确保对象在两个进程中可访问或已被正确序列化。

概念性使用示例（修改解释器后） #

import multiprocessing
import pickle

def simple_generator():
    print("Running in process one")
    yield
    print("Running in process two")

def process_one():
    gen = simple_generator()
    next(gen)  # 执行到yield处暂停
    # 序列化生成器状态
    return pickle.dumps(gen)

def process_two(serialized_gen):
    # 反序列化并恢复生成器
    gen = pickle.loads(serialized_gen)
    next(gen)  # 从yield后继续执行

if __name__ == "__main__":
    serialized = process_one()
    p = multiprocessing.Process(target=process_two, args=(serialized,))
    p.start()
    p.join()

这种方案复杂度极高，需要深入理解Python解释器的生成器、栈帧底层实现，且需考虑不同版本的兼容性。

三、替代方案：用类实现状态机模拟生成器 #

如果不想修改解释器，也可以用类实例保存状态，模拟生成器的行为，天然支持跨进程传递。

示例代码 #

import multiprocessing

class SimpleGenerator:
    def __init__(self, state=None):
        self.state = state or {}
        self.current_step = self.state.get('step', 0)
    
    def __next__(self):
        if self.current_step == 0:
            print("Running in process one")
            self.current_step = 1
            self.state['step'] = 1
            return self.state
        elif self.current_step == 1:
            print("Running in process two")
            self.current_step = 2
            raise StopIteration

def process_one():
    gen = SimpleGenerator()
    state = next(gen)
    return state

def process_two(state):
    gen = SimpleGenerator(state)
    next(gen)

if __name__ == "__main__":
    state = process_one()
    p = multiprocessing.Process(target=process_two, args=(state,))
    p.start()
    p.join()

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