使用多进程加速Python象棋程序反而变慢的问题咨询

问题背景 #

我正在用Python开发一款象棋程序，已接近完成。原本采用2进程架构：主进程运行main负责游戏流程监控与响应，另一个进程运行走法选择算法。为加速最优走法查找，我将合法走法拆分为两个列表，交由两个新进程处理，形成4进程架构（我的PC是4核8线程CPU），但结果耗时反而比2进程架构更高。

耗时数据对比 #

2进程架构核心代码 #

def find_best_move(cre_gs: object(), valid_moves: list, queue) -> object():
    # random.shuffle(valid_moves)
    global next_move
    find_nega_max_move_alpha_beta(cre_gs, DEPTH, valid_moves, 1 if cre_gs.white_to_move else -1, -checkmate, checkmate)
    queue.put(next_move)

def find_nega_max_move_alpha_beta(cre_gs: object(), depth: int, valid_moves: list, turn_multiplier: int, alpha: int, beta: int) -> int:
    global next_move
    if depth == 0:
        return turn_multiplier * score_material(cre_gs)

    max_score = -checkmate
    for move in valid_moves:
        cre_gs.make_move(move)
        next_moves = cre_gs.get_valid_moves()
        score = -find_nega_max_move_alpha_beta(cre_gs, depth - 1, next_moves, -turn_multiplier, -beta, -alpha)
        cre_gs.undo_move()

        if score > max_score:
            max_score = score
            if depth == DEPTH:
                next_move = move

        if max_score > alpha:
            alpha = max_score
        if alpha >= beta:
            break
    return max_score

def score_material(cre_gs: object()) -> int:
    if cre_gs.checkmate:
        if cre_gs.white_to_move:
            return checkmate
        else:
            return -checkmate
    score = 0
    for row in cre_gs.board:
        for square in row:
            if square[0] == 'w':
                score += piece_score[square[1]]
            elif square[0] == 'b':
                score -= piece_score[square[1]]
    return score

4进程架构核心代码 #

def find_best_move(cre_gs: object(), valid_moves: list, queue) -> object():
    # random.shuffle(valid_moves)
    start_time = time.time()
    global next_move
    process_1_moves = valid_moves[0: len(valid_moves) // 2]
    process_2_moves = valid_moves[len(valid_moves) // 2::]
    return_queue_1 = Queue()
    return_queue_2 = Queue()

    process_1 = Process(target=find_best_move_process, args=(cre_gs, process_1_moves, return_queue_1))
    process_2 = Process(target=find_best_move_process, args=(cre_gs, process_2_moves, return_queue_2))
    process_1.start()
    process_2.start()
    print(cpu_count())
    process_1.join()
    print('1.joined', time.time() - start_time)
    process_2.join()
    print('2.joined', time.time() - start_time)

    move_1 = return_queue_1.get()
    move_2 = return_queue_2.get()

    if not process_1.is_alive() and not process_2.is_alive():
        find_nega_max_move_alpha_beta(cre_gs, DEPTH, [move_1, move_2], 1 if cre_gs.white_to_move else -1, -checkmate, checkmate)
        queue.put(next_move)
        print('all.joined', time.time() - start_time)

def find_best_move_process(cre_gs: object(), valid_moves: list, queue) -> object():
    global next_move
    find_nega_max_move_alpha_beta(cre_gs, DEPTH, valid_moves, 1 if cre_gs.white_to_move else -1, -checkmate, checkmate)
    queue.put(next_move)

def find_nega_max_move_alpha_beta(cre_gs: object(), depth: int, valid_moves: list, turn_multiplier: int, alpha: int, beta: int) -> int:
    global next_move
    if depth == 0:
        return turn_multiplier * score_material(cre_gs)

    max_score = -checkmate
    for move in valid_moves:
        cre_gs.make_move(move)
        next_moves = cre_gs.get_valid_moves()
        score = -find_nega_max_move_alpha_beta(cre_gs, depth - 1, next_moves, -turn_multiplier, -beta, -alpha)
        cre_gs.undo_move()

        if score > max_score:
            max_score = score
            if depth == DEPTH:
                next_move = move

        if max_score > alpha:
            alpha = max_score
        if alpha >= beta:
            break
    return max_score

def score_material(cre_gs: object()) -> int:
    if cre_gs.checkmate:
        if cre_gs.white_to_move:
            return checkmate
        else:
            return -checkmate
    score = 0
    for row in cre_gs.board:
        for square in row:
            if square[0] == 'w':
                score += piece_score[square[1]]
            elif square[0] == 'b':
                score -= piece_score[square[1]]
    return score

耗时增加的原因 #

1. 进程创建与通信开销：Python的multiprocessing.Process创建进程、传递cre_gs对象及队列通信都有额外开销，如果拆分的任务量不够大，这些开销会超过并行计算的收益。
2. 任务负载不均衡：简单将合法走法对半拆分，可能导致两个进程的计算量差异极大——比如一半走法的搜索树深度远大于另一半，导致一个进程长时间运行，另一个早早结束，无法有效利用并行资源。
3. 重复计算：子进程结束后重新搜索两个候选走法，相当于重复计算了部分内容，额外增加了耗时。
4. 进程调度开销：4核CPU运行4个进程时，操作系统需频繁进行进程上下文切换，带来额外性能损耗。

优化方案 #

针对多进程架构的优化 #

1. 避免重复计算：让子进程返回每个走法的评分，直接在主进程中比较评分选出最优走法，省去最后一次搜索的开销。
2. 优化任务拆分策略：不要简单对半拆分，而是根据走法的预估复杂度（比如吃子、将军等优先级高的走法）分配任务，或使用进程池进行动态任务分配，让负载更均衡。
3. 减少进程通信数据量：传递不可变的游戏状态副本，或让每个进程独立复制一份游戏状态，避免对象序列化/反序列化的开销。
4. 复用进程：使用multiprocessing.Pool创建进程池，复用已有进程，减少进程创建销毁的开销。

算法本身的优化 #

1. 走法排序：对合法走法进行启发式排序（优先搜索吃子、将军等走法），让α-β剪枝更早生效，减少搜索节点数。
2. 置换表：缓存已计算过的游戏状态评分，避免重复搜索相同局面，这是棋类AI提升效率的核心手段。
3. 剪枝优化：使用MTD(f)等更高效的搜索算法代替普通极小极大搜索，提升剪枝效率。

替代方案 #

如果multiprocessing.Process不适用，可考虑以下方案：

1. multiprocessing.Pool：复用进程减少创建开销，适合批量任务处理，将每个走法的搜索任务提交到进程池后收集结果比较。
2. JIT编译/C扩展：用numba对核心搜索函数进行JIT编译，或用Cython将核心代码编译为C，或用PyPy运行程序，大幅提升单线程效率，收益往往比并行优化更显著。
3. 多线程：仅当搜索算法包含IO等待或使用释放GIL的库时适用，纯CPU密集型任务下收益有限。

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