Windows下线程内使用Multiprocessing Manager脚本异常重启问题

1. 重复创建Genetic实例与RuntimeError解决 #

原因：Windows的multiprocessing采用spawn启动模式，会重新导入主模块（main.py），如果主模块没有用if __name__ == '__main__':包裹执行代码，会导致初始化逻辑重复运行，创建多个Genetic实例，同时触发RuntimeError。之前添加freeze_support()无效是因为位置错误，必须放在主模块的入口判断块内。

修复步骤：

• 将main.py中的执行代码全部放入if __name__ == '__main__':块内，同时在块开头添加multiprocessing.freeze_support()（Linux下无影响，仅针对Windows处理）。

修改后的main.py：

import time
import multiprocessing as mp
from genetic import Genetic

# 假设你的常量和waypointDict定义在这里
LANDING_TIME = ...
TAKEOFF_TIME = ...
FLIGHT_SPEED = ...
MAX_FLIGHT_TIME = ...
waypointDict = ...
msg = ...

if __name__ == '__main__':
    mp.freeze_support()
    print("Creating optimiser")
    optimiser = Genetic(waypointDict,
                        msg,
                        LANDING_TIME,
                        TAKEOFF_TIME,
                        FLIGHT_SPEED,
                        MAX_FLIGHT_TIME,
                        population_size=320,
                        breeding_pool_size_percent=0.4,
                        selection_pool_size=15)

    print("Starting optimiser")
    optimiser.start()

    print("Waiting for optimiser to finish")
    while optimiser.is_alive():
        time.sleep(0.1)

2. AttributeError: 无法pickle本地对象解决 #

原因：mutate_population是定义在Genetic.run()方法内部的局部函数，Windows的spawn模式需要序列化（pickle）传递给子进程的目标函数，而局部函数的依赖关系无法被序列化，因此抛出错误。

修复步骤：

• 将mutate_population从run()内部移出，改为模块级函数或类的实例/静态方法。这里提供两种可行方案：

方案一：改为类的实例方法 #

from threading import Thread, Event
import multiprocessing as mp

class Genetic(Thread):
    def __init__(self, waypoints, msg, landing_time, takeoff_time, flight_speed, max_flight_time, population_size, breeding_pool_size_percent, selection_pool_size):
        Thread.__init__(self)
        self.exit_event = Event()
        # 其他初始化逻辑...

    def stop(self) -> None:
        print("Stopping genetic algorithm")
        if self.exit_event.is_set():
            print("Already stopped")
            return
        self.exit_event.set()
        self.join()

    # 移到类内作为实例方法
    def mutate_population(self, population: list[Solution], new_population) -> None:
        for i in range(len(population)):
            population[i] = self.mutate(population[i])
        new_population.extend(population)

    def run(self) -> None:        
        print("Initializing population")
        self.population = self.initialize_population()
        self.best_solution = self.population[0]
        last_best_solution = None
        stop_count: int = 0
        iteration: int = 1

        while stop_count < 8 and not self.exit_event.is_set():
            print(f"Genetic algorithm iteration {iteration}: ".ljust(40) + f"${self.best_solution.total_revenue} at {self.best_solution.total_time} minutes")

            self.population = self.breed()

            jobs = []
            process_count = 4
            manager = mp.Manager()
            new_population = manager.list()
            chunk_size = len(self.population) // process_count
            for i in range(0, len(self.population), chunk_size):
                chunk = self.population[i:i + chunk_size]
                # 直接传递实例方法作为目标
                p = mp.Process(target=self.mutate_population, args=(chunk, new_population))
                jobs.append(p)
                p.start()
            for job in jobs:
                job.join()
            self.population = list(new_population)

            self.best_solution = max(self.population)
            if self.best_solution != last_best_solution:
                stop_count = 0
            last_best_solution = self.best_solution
            stop_count += 1
            iteration += 1

方案二：改为模块级函数（避免序列化self） #

如果self包含不可序列化的对象（如线程内部状态），可以用此方案：

# 在genetic.py的模块级别定义函数
def mutate_population(population: list[Solution], new_population, mutate_func) -> None:
    for i in range(len(population)):
        population[i] = mutate_func(population[i])
    new_population.extend(population)

class Genetic(Thread):
    # ...其他方法不变...
    def run(self) -> None:
        # ...循环内的代码修改为：
        jobs = []
        process_count = 4
        manager = mp.Manager()
        new_population = manager.list()
        chunk_size = len(self.population) // process_count
        for i in range(0, len(self.population), chunk_size):
            chunk = self.population[i:i + chunk_size]
            # 传递模块级函数，并把mutate方法作为参数传入
            p = mp.Process(target=mutate_population, args=(chunk, new_population, self.mutate))
            jobs.append(p)
            p.start()
        for job in jobs:
            job.join()
        self.population = list(new_population)

关键原理说明 #

• Linux/macOS的multiprocessing默认用fork模式，直接复制父进程内存空间，不需要序列化函数和对象；而Windows用spawn模式，会重新启动Python解释器并导入模块，因此必须保证主模块的执行代码在if __name__ == '__main__':内，且传递给子进程的函数、对象必须可pickle。
• 局部函数、嵌套函数无法被pickle，因为它们的依赖关系无法被序列化，必须改为模块级或类级的可访问函数。

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