Python体素光线追踪器性能优化：多进程直接修改Pygame像素

核心结论：子进程无法直接修改主线程的Pygame屏幕 #

Python多进程基于内存隔离机制，每个子进程拥有独立的地址空间。主线程的screen显示表面在子进程中是独立副本，子进程调用screen.set_at只会修改自身副本，无法同步到主线程的显示界面，因此这条路走不通。

最优解决方案方向 #

1. 优化数据传输与屏幕更新方式，替代逐个set_at #

逐个调用screen.set_at是主线程性能瓶颈的核心原因之一，Pygame的批量像素操作效率远高于单像素更新：

• 避免字符串序列化开销：让trace函数直接返回RGB元组（如(x, y, r, g, b)）或字节数据，不要返回十六进制字符串，减少主线程的类型转换耗时。
• 分块计算+批量Blit：将屏幕划分为若干矩形块（比如按行划分，每个块包含10-20行像素），子进程负责计算整块的像素数据，返回后主线程直接将整块数据作为子Surfaceblit到屏幕对应位置。

示例代码片段：

# 子进程任务：计算指定行范围的像素数据
def trace_block(start_row, end_row, width):
    block_data = []
    for y in range(start_row, end_row):
        row = []
        for x in range(width):
            # 替换为你的光线追踪计算逻辑，返回RGB值
            r, g, b = calculate_voxel_color(x, y)
            row.append((r, g, b))
        block_data.append(row)
    # 返回块的起始坐标和像素数据
    return (start_row, 0), block_data

# 主线程处理逻辑
import pygame
from concurrent.futures import ProcessPoolExecutor

screen = pygame.display.set_mode((240, 120))
executor = ProcessPoolExecutor()
# 划分任务：每10行一个计算块
tasks = []
for start_row in range(0, 120, 10):
    end_row = min(start_row + 10, 120)
    tasks.append(executor.submit(trace_block, start_row, end_row, 240))

running = True
while running:
    for event in pygame.event.get():
        if event.type == pygame.QUIT:
            running = False
    # 处理已完成的异步任务
    for future in concurrent.futures.as_completed(tasks):
        pos, block_data = future.result()
        block_height = pos[0] + len(block_data)
        # 创建子Surface并填充像素
        block_surf = pygame.Surface((width, block_height - pos[0]))
        block_surf.set_pixels(block_data)
        # 批量Blit到屏幕对应位置
        screen.blit(block_surf, pos)
    pygame.display.flip()

2. 替换pool.map为异步任务处理，充分利用子进程算力 #

pool.map是阻塞式调用，必须等待所有任务完成才返回结果，导致主线程长时间等待，无法及时更新屏幕。改用异步方式：

• 使用concurrent.futures.ProcessPoolExecutor的as_completed方法，一旦有子进程完成一块计算，立即获取结果并更新屏幕，无需等待所有任务结束。
• 或者用multiprocessing.Queue，子进程计算完一块就把结果放入队列，主线程循环从队列取数据并更新，实现流水线式的计算与渲染。

3. 用共享内存减少进程间数据拷贝开销 #

如果分块传输的数据量仍然很大，可以用共享内存让子进程直接写入像素数据，避免进程间的数据拷贝：

• 使用multiprocessing.shared_memory创建与屏幕像素格式匹配的共享内存区域（比如240×120×3字节，对应RGB888格式）。
• 子进程映射该共享内存，直接写入对应区域的像素值。
• 主线程定期用pygame.surfarray.blit_array将共享内存的数据刷到屏幕上，这是Pygame中最快的像素批量更新方式之一。

示例代码片段：

import multiprocessing
import pygame
import numpy as np
from multiprocessing import shared_memory

# 创建共享内存：240*120像素，每个像素3字节RGB
shm = shared_memory.SharedMemory(create=True, size=240*120*3)
# 用numpy数组映射共享内存，方便操作
pixel_array = np.ndarray((120, 240, 3), dtype=np.uint8, buffer=shm.buf)

def trace_shared(start_row, end_row, shm_name):
    # 子进程连接共享内存
    existing_shm = shared_memory.SharedMemory(name=shm_name)
    local_array = np.ndarray((120, 240, 3), dtype=np.uint8, buffer=existing_shm.buf)
    for y in range(start_row, end_row):
        for x in range(240):
            r, g, b = calculate_voxel_color(x, y)
            local_array[y][x] = (r, g, b)
    existing_shm.close()

# 启动子进程
processes = []
for start_row in range(0, 120, 10):
    end_row = min(start_row + 10, 120)
    p = multiprocessing.Process(target=trace_shared, args=(start_row, end_row, shm.name))
    processes.append(p)
    p.start()

# 主线程更新屏幕
screen = pygame.display.set_mode((240, 120))
running = True
while running:
    for event in pygame.event.get():
        if event.type == pygame.QUIT:
            running = False
    # 将共享内存的像素数据刷到屏幕
    pygame.surfarray.blit_array(screen, pixel_array)
    pygame.display.flip()

# 资源清理
for p in processes:
    p.join()
shm.close()
shm.unlink()

总结优先级 #

1. 优先尝试分块计算+异步任务+批量Blit，实现简单，能大幅降低主线程开销，充分利用子进程算力。
2. 如果数据传输开销仍然过大，再引入共享内存，进一步提升性能。
3. 彻底抛弃pool.map的阻塞式调用，改用异步任务处理模式，避免主线程等待。

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