《Python高级编程》5.3 多进程、共享内存与 IPC 选型

import multiprocessing as mp
print(mp.get_start_method())        # spawn
print(mp.get_all_start_methods())   # ['spawn', 'fork', 'forkserver']

import multiprocessing as mp, time, os

def child():
    return os.getpid()

def measure(method, n=20):
    ctx = mp.get_context(method)
    t0 = time.perf_counter()
    ps = [ctx.Process(target=child) for _ in range(n)]
    for p in ps: p.start()
    for p in ps: p.join()
    return (time.perf_counter() - t0) / n

for m in ("fork", "spawn", "forkserver"):
    print(f"{m:10s}: {measure(m)*1000:.2f} ms/process")

fork      :  6.84 ms/process (n=20)
spawn     : 42.63 ms/process (n=20)
forkserver: 10.76 ms/process (n=20)

import pickle, time, numpy as np
for n in (10_000, 1_000_000, 10_000_000):
    a = np.arange(n, dtype=np.float64)
    t0 = time.perf_counter(); pickle.loads(pickle.dumps(a, -1)); d1 = time.perf_counter()-t0
    t0 = time.perf_counter(); b = np.empty_like(a); b[:] = a[:]; d2 = time.perf_counter()-t0
    print(f"n={n:>10,} bytes={a.nbytes/1e6:6.1f}MB pickle={d1*1000:8.2f}ms memcpy={d2*1000:7.2f}ms ratio={d1/d2:5.1f}x")

n=    10,000 bytes=   0.1MB pickle=   0.76ms memcpy=  0.02ms ratio=45.4x
n= 1,000,000 bytes=   8.0MB pickle=   8.28ms memcpy=  0.69ms ratio=12.1x
n=10,000,000 bytes=  80.0MB pickle=  22.73ms memcpy=  6.67ms ratio= 3.4x

import multiprocessing as mp
from multiprocessing import shared_memory
import numpy as np

def worker(name, shape, dtype):
    shm = shared_memory.SharedMemory(name=name)
    a = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    a += 1                 # 原地修改，不复制回传
    shm.close()

if __name__ == "__main__":
    arr = np.arange(1_000_000, dtype=np.float64)
    shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
    shared = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
    shared[:] = arr[:]     # 只拷贝一次进去
    p = mp.Process(target=worker, args=(shm.name, arr.shape, arr.dtype))
    p.start(); p.join()
    print(int(shared[0]), int(shared[-1]))   # 父进程看到子进程的修改
    shm.close(); shm.unlink()

fork  : shared_mem_roundtrip=5.2ms   pickle_arg_roundtrip=2.5ms   parent_sees=(1, 1000000)
spawn : shared_mem_roundtrip=237.2ms pickle_arg_roundtrip=294.3ms parent_sees=(1, 1000000)

# 消费者进程循环 recv 直到收到哨兵；生产者发送 n 条 payload
# 结果（fork 上下文）：
# n= 10000 payload=     100B  Queue=  55.1ms  Pipe=  31.8ms  Pipe/Queue=0.58
# n= 10000 payload=  100000B  Queue=  50.5ms  Pipe=  31.5ms  Pipe/Queue=0.62
# n=  1000 payload= 1000000B  Queue=   7.1ms  Pipe=   4.7ms  Pipe/Queue=0.66

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def cpu_task(n):
    s = 0
    for i in range(n):
        s += i * i
    return s

CPU-bound K=8 N=4_000_000:
  serial=1.383s  threadpool=1.354s  processpool=0.803s
  speedup: thread=1.02x  process=1.72x  (cpus=10)

I/O-bound K=8 (每份 sleep 0.1s):
  serial=0.841s  threadpool=0.106s  processpool=0.613s

processpool[fork] : 0.282s
processpool[spawn]: 0.779s

from multiprocessing import shared_memory

shm = shared_memory.SharedMemory(create=True, size=1024)
try:
    # ... 子进程用 name 打开、读写 ...
    pass
finally:
    shm.close()
    shm.unlink()      # 只在创建者进程里做一次