Python Multiprocessing & Queues

The Story That Explains Multiprocessing 🏗️Multiprocessing समझाने वाली कहानी 🏗️一个讲清多进程的故事 🏗️

Threads vs Processes — The Key Difference ⚖️Threads बनाम Processes — मुख्य अंतर ⚖️线程 vs 进程 — 根本区别 ⚖️

Your First Process — The Basics 🚀आपका पहला Process — बुनियाद 🚀你的第一个进程 — 基础 🚀

multiprocessing.Process looks almost exactly like threading.Thread. If you know threads, you already know most of processes: target, args, start(), join().

multiprocessing.Process लगभग बिल्कुल threading.Thread जैसा दिखता है। अगर आप threads जानते हैं, तो processes का ज़्यादातर हिस्सा पहले से जानते हैं: target, args, start(), join()।

multiprocessing.Process 和 threading.Thread 几乎一模一样。如果你会用线程，你就已经掌握了进程的大部分：target、args、start()、join()。

from multiprocessing import Process
import os
import time

def worker(name, duration):
    print(f"[{name}] pid={os.getpid()} starting")
    time.sleep(duration)
    print(f"[{name}] pid={os.getpid()} done")

if __name__ == "__main__":                # REQUIRED on Windows and macOS
    print(f"Main pid={os.getpid()}")
    processes = []
    for i in range(3):
        p = Process(target=worker, args=(f"P{i}", 2))
        p.start()
        processes.append(p)
    for p in processes:
        p.join()
    print("All processes complete.")

Each child has its own process id (pid). All three slept for 2 seconds at the same time, so the whole program took about 2 seconds, not 6.

हर child का अपना process id (pid) है। तीनों ने एक साथ 2 seconds sleep किया, इसलिए पूरा program 6 नहीं, लगभग 2 seconds में चला।

每个子进程都有自己的进程 ID（pid）。三个进程同时 sleep 了 2 秒，所以整个程序大约用了 2 秒，而不是 6 秒。

What really happens without the guardGuard के बिना असल में क्या होता है没有保护时实际会发生什么

import multiprocessing as mp, os

ctx = mp.get_context("spawn")      # the start method Windows and macOS use

def worker():
    print("child is working")

print("top of script, pid", os.getpid())
p = ctx.Process(target=worker)     # NO  if __name__ == "__main__":  guard!
p.start()
p.join()
print("exit code of child:", p.exitcode)

Daemon Processes — Background Workers That End With the Parent 👻Daemon Processes — Parent के साथ ख़त्म होने वाले Background Workers 👻守护进程 — 随父进程结束的后台工作者 👻

Daemon in actionDaemon काम करते हुए守护进程实战

from multiprocessing import Process
import time

def background_monitor():
    while True:
        print("  [daemon] still watching...")
        time.sleep(1)

if __name__ == "__main__":
    p = Process(target=background_monitor)
    p.daemon = True                     # MUST be set before start()
    p.start()
    print("[main] doing work for 3 seconds...")
    time.sleep(3)
    print("[main] exiting - daemon will be killed automatically")
    # No p.join() needed - daemon dies with parent

The same program WITHOUT the daemon flagवही program daemon flag के बिना同一个程序，不设守护标志

from multiprocessing import Process
import time

def background_monitor():
    while True:
        print("  [child] still watching...")
        time.sleep(1)

if __name__ == "__main__":
    p = Process(target=background_monitor)
    # p.daemon = True   <-- omitted!
    p.start()
    time.sleep(3)
    print("[main] want to exit...")
    # PROGRAM HANGS FOREVER - the child loops, the parent waits for it

from multiprocessing import Process

def grandchild():
    print("grandchild running")

def daemon_worker():
    p = Process(target=grandchild)     # a daemon tries to start a child
    p.start()
    p.join()

if __name__ == "__main__":
    d = Process(target=daemon_worker, daemon=True)
    d.start()
    d.join()
    print("daemon exit code:", d.exitcode)

from multiprocessing import Process, Event
import time
import atexit

def graceful_monitor(stop_event):
    try:
        while not stop_event.is_set():
            print("  [worker] doing work...")
            stop_event.wait(timeout=1)     # wakes early if signalled
    finally:
        print("  [worker] shutting down cleanly - flushing state")

if __name__ == "__main__":
    stop = Event()
    p = Process(target=graceful_monitor, args=(stop,))
    p.start()
    atexit.register(lambda: (stop.set(), p.join(timeout=5)))
    time.sleep(3)
    print("[main] done")

Look at the last line: it printed after "[main] done". The atexit handler set the event when the parent finished, and the worker's finally block ran. A daemon would have been cut off instead.

आख़िरी line देखिए: यह "[main] done" के बाद print हुई। Parent के ख़त्म होने पर atexit handler ने event set किया, और worker का finally block चला। Daemon होता तो वह बीच में कट जाता।

看最后一行：它在"[main] done"之后 打印。父进程结束时 atexit 处理函数设置了事件，工作进程的 finally 块得以运行。如果是守护进程，它就会被直接切断。

Fork vs Spawn — How Processes Are Created 🌱Fork बनाम Spawn — Processes कैसे बनते हैं 🌱Fork vs Spawn — 进程是如何创建的 🌱

import multiprocessing as mp

# Set the start method explicitly for the same behaviour on every platform
if __name__ == "__main__":
    print("Default here:", mp.get_start_method(allow_none=True) or "(not chosen yet)")
    mp.set_start_method("spawn")     # or "fork", "forkserver" - call once only
    print("Available:   ", mp.get_all_start_methods())
    print("Current:     ", mp.get_start_method())

🌱 Simulator: Start-Method Lab🌱 Simulator: Start-Method Lab🌱 模拟器：启动方式实验室

The same script was run with each start method. The parent changes a global CONFIG before starting a child. Does the child see the change? Is the module loaded again? How long does one start + join take? All real.

वही script हर start method से चलाई गई। Parent child शुरू करने से पहले एक global CONFIG बदलता है। क्या child को बदलाव दिखता है? क्या module फिर load होता है? एक start + join में कितना समय लगता है? सब असली।

同一个脚本分别用三种启动方式运行。父进程在启动子进程之前修改了全局变量 CONFIG。子进程能看到修改吗？模块会被再次加载吗？一次启动 + join 要多久？全部是真实数据。

Pool — The Workhorse of Parallel Python 🐎Pool — Parallel Python का मेहनती घोड़ा 🐎Pool — 并行 Python 的主力 🐎

Creating and joining processes by hand gets tedious. multiprocessing.Pool starts a fixed group of worker processes once and gives you high-level methods to spread work across them.

हाथ से processes बनाना और join करना थकाऊ हो जाता है। multiprocessing.Pool worker processes का एक तय समूह एक बार शुरू करता है और उनमें काम बाँटने के ऊँचे-स्तर के methods देता है।

手动创建和等待进程很快就会变得繁琐。multiprocessing.Pool 一次性启动一组固定的工作进程，并提供高级方法把工作分配给它们。

from multiprocessing import Pool
import time

def heavy_compute(n):
    # Simulated CPU work
    total = 0
    for i in range(10_000_000):
        total += i * n
    return n, total

if __name__ == "__main__":
    numbers = list(range(1, 9))

    # --- Sequential ---
    start = time.perf_counter()
    results = [heavy_compute(n) for n in numbers]
    print(f"Sequential: {time.perf_counter() - start:.2f}s")

    # --- Pool with 4 workers ---
    start = time.perf_counter()
    with Pool(processes=4) as pool:
        results = pool.map(heavy_compute, numbers)
    print(f"Pool(4):    {time.perf_counter() - start:.2f}s")

    for n, total in results:
        print(f"  {n}: {total}")

Pool Methods — Choosing the Right One 🧰Pool Methods — सही चुनना 🧰Pool 方法 — 选对那一个 🧰

Practical examples of eachहर एक के व्यावहारिक उदाहरण每种方法的实际示例

from multiprocessing import Pool

def square(x): return x * x
def add(a, b): return a + b

def on_complete(results):
    print(f"Done! {len(results)} results")

if __name__ == "__main__":
    with Pool(4) as pool:
        # 1. map - simplest, blocks until done, results in order
        print(pool.map(square, [1, 2, 3, 4, 5]))

        # 2. starmap - for multi-argument functions
        print(pool.starmap(add, [(1, 2), (3, 4), (5, 6)]))

        # 3. imap_unordered - stream results as they finish
        for result in pool.imap_unordered(square, range(10)):
            print(result, end=" ")
        print()                       # finish the line

        # 4. apply_async - fire off a single task without blocking
        r = pool.apply_async(square, (10,))
        # ... do other work ...
        print(r.get(timeout=5))

        # 5. map_async with a callback
        r = pool.map_async(square, range(100), callback=on_complete)
        r.wait()

🐎 Simulator: Pool Lab (real traces)🐎 Simulator: Pool Lab (असली traces)🐎 模拟器：Pool 实验室（真实追踪）

8 CPU tasks of different sizes (task #0 is the biggest). Choose the number of workers, the method and the chunksize. The bars show which worker process ran which task; the row underneath shows when each result reached your code. 12 real runs.

अलग-अलग size के 8 CPU tasks (task #0 सबसे बड़ा)। Workers की संख्या, method और chunksize चुनिए। Bars दिखाती हैं कि कौन-सा worker process कौन-सा task चला; नीचे की row दिखाती है कि हर result आपके code तक कब पहुँचा। 12 असली runs।

8 个大小不同的 CPU 任务（任务 #0 最大）。选择工作进程数、方法和 chunksize。条形图显示哪个工作进程执行了哪个任务；下面一行显示 每个结果何时到达你的代码。12 次真实运行。

multiprocessing.Queue — Inter-Process Communication 📫multiprocessing.Queue — Processes के बीच बातचीत 📫multiprocessing.Queue — 进程间通信 📫

Producer-consumer between processesProcesses के बीच producer-consumer进程之间的生产者-消费者

from multiprocessing import Process, Queue
import time
import os

def producer(q, count):
    for i in range(count):
        item = {"id": i, "pid": os.getpid(), "payload": i * 2}
        q.put(item)
        print(f"[P] produced {item}")
        time.sleep(0.1)
    q.put(None)                              # poison pill

def consumer(q):
    while True:
        item = q.get()
        if item is None: break
        print(f"    [C pid={os.getpid()}] got {item}")
        time.sleep(0.15)

if __name__ == "__main__":
    q = Queue(maxsize=10)
    p1 = Process(target=producer, args=(q, 5))
    p2 = Process(target=consumer, args=(q,))
    p1.start(); p2.start()
    p1.join();  p2.join()
    print("Done.")

from multiprocessing import Pool

if __name__ == "__main__":
    with Pool(2) as pool:
        print(pool.map(lambda x: x * 2, [1, 2, 3]))   # a lambda cannot be pickled

🥒 Simulator: Pickle Checker🥒 Simulator: Pickle Checker🥒 模拟器：Pickle 检查器

Can this object be sent to another process? Each answer is the real result of pickle.dumps() on Python 3.13.16.

क्या यह object दूसरे process को भेजा जा सकता है? हर जवाब Python 3.13.16 पर pickle.dumps() का असली नतीजा है।

这个对象能发送给另一个进程吗？每个答案都是在 Python 3.13.16 上运行 pickle.dumps() 的真实结果。

Pipes — The Two-Endpoint Alternative 🔌Pipes — दो सिरों वाला विकल्प 🔌管道 — 两端点的替代方案 🔌

multiprocessing.Pipe() returns two connected ends for two-way communication between exactly two processes. It is faster than a Queue for point-to-point traffic, but it is not safe for many producers or many consumers on the same end.

multiprocessing.Pipe() ठीक दो processes के बीच दो-तरफ़ा बातचीत के लिए दो जुड़े सिरे लौटाता है। सीधी बातचीत के लिए यह Queue से तेज़ है, पर एक ही सिरे पर कई producers या कई consumers के लिए सुरक्षित नहीं।

multiprocessing.Pipe() 返回两个相连的端点，用于 恰好两个 进程之间的双向通信。对于点对点通信，它比 Queue 快，但同一端有多个生产者或消费者时并不安全。

from multiprocessing import Process, Pipe

def worker(conn):
    conn.send({"status": "ready", "pid": 42})
    reply = conn.recv()                       # wait for the parent's message
    print(f"Worker got reply: {reply}")
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p = Process(target=worker, args=(child_conn,))
    p.start()
    print("Parent got:", parent_conn.recv())
    parent_conn.send("go ahead")
    p.join()

Shared Memory — When Pickling Is Too Slow 💾Shared Memory — जब Pickling बहुत धीमा हो 💾共享内存 — 当 pickle 太慢时 💾

Pickling a big NumPy array to send it through a Queue is slow. For large data, Python offers real shared memory: one block of RAM that several processes read and write directly, with no copying.

बड़े NumPy array को Queue से भेजने के लिए pickle करना धीमा है। बड़े data के लिए Python असली shared memory देता है: RAM का एक block जिसे कई processes सीधे पढ़ते और लिखते हैं, बिना copy किए।

为了通过 Queue 发送一个大的 NumPy 数组而对它进行 pickle 是很慢的。对于大数据，Python 提供了 真正的共享内存：一块多个进程可以直接读写的内存，无需复制。

📦 Simulator: IPC Cost Lab📦 Simulator: IPC Cost Lab📦 模拟器：进程间通信开销实验室

How long does it take to hand a NumPy array to another process and get a tiny answer back? Measured on our machine (best of 2 runs). For shared_memory only the block's name travels; the child attaches and reads the data in place.

एक NumPy array दूसरे process को देकर छोटा-सा जवाब वापस पाने में कितना समय लगता है? हमारी machine पर मापा गया (2 runs में सबसे अच्छा)। shared_memory में सिर्फ़ block का नाम जाता है; child जुड़कर data वहीं पढ़ता है।

把一个 NumPy 数组交给另一个进程并拿回一个很小的答案需要多久？在我们的机器上测得（2 次中取最好）。使用 shared_memory 时只传递内存块的 名字；子进程连接后直接在原处读取数据。

Shared Memory — Value and Array (Simple C Types) 🔢Shared Memory — Value और Array (सरल C Types) 🔢共享内存 — Value 与 Array（简单 C 类型）🔢

For a shared counter or a small fixed-size array of numbers, use Value and Array. They hold C-level numbers in shared memory and come with an optional lock.

साझा counter या numbers की छोटी, तय-size array के लिए Value और Array लीजिए। ये shared memory में C-level numbers रखते हैं और एक वैकल्पिक lock के साथ आते हैं।

对于共享计数器或固定大小的小型数值数组，使用 Value 和 Array。它们在共享内存中保存 C 级别的数值，并自带一个可选的锁。

from multiprocessing import Process, Value, Array, Lock

def increment(counter, lock, n):
    for _ in range(n):
        with lock:                              # atomic increment
            counter.value += 1

def square_in_place(arr, start, end):
    for i in range(start, end):
        arr[i] = arr[i] * arr[i]

if __name__ == "__main__":
    # --- Shared counter ---
    counter = Value('i', 0)                    # 'i' = signed int
    lock = Lock()
    procs = [Process(target=increment, args=(counter, lock, 10_000))
             for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()
    print(f"Final counter: {counter.value}")

    # --- Shared array ---
    arr = Array('d', range(20))               # 'd' = double
    p1 = Process(target=square_in_place, args=(arr, 0, 10))
    p2 = Process(target=square_in_place, args=(arr, 10, 20))
    p1.start(); p2.start()
    p1.join();  p2.join()
    print(list(arr[:]))

🔀 Simulator: Shared-State Lab🔀 Simulator: Shared-State Lab🔀 模拟器：共享状态实验室

4 processes each add 1 to a counter 5,000 times (expected total: 20,000). Four ways to hold the counter, each run 3 times for real. Compare the totals and the times.

4 processes हर एक counter में 5,000 बार 1 जोड़ते हैं (अपेक्षित कुल: 20,000)। Counter रखने के चार तरीक़े, हर एक 3 बार असल में चलाया गया। कुल और समय की तुलना कीजिए।

4 个进程各自给计数器加 1，共 5,000 次（预期总数：20,000）。计数器有四种保存方式，每种都真实运行 3 次。比较总数和用时。

Shared Memory — Manager (Any Python Object) 🗂️Shared Memory — Manager (कोई भी Python Object) 🗂️共享内存 — Manager（任意 Python 对象）🗂️

To share dicts, lists or other Python objects, use a Manager. It starts a separate server process that owns the real object; every access from a worker is a small request to that server (a remote procedure call).

Dicts, lists या दूसरे Python objects बाँटने के लिए Manager लीजिए। यह एक अलग server process शुरू करता है जो असली object रखता है; worker की हर पहुँच उस server को एक छोटी request है (remote procedure call)।

要共享字典、列表或其他 Python 对象，使用 Manager。它会启动一个单独的 服务器进程 来持有真正的对象；工作进程的每次访问都是向该服务器发出的一个小请求（远程过程调用）。

from multiprocessing import Process, Manager

def worker(shared_dict, shared_list, worker_id):
    shared_dict[worker_id] = f"done by worker {worker_id}"
    shared_list.append(worker_id ** 2)

if __name__ == "__main__":
    with Manager() as manager:
        shared_dict = manager.dict()
        shared_list = manager.list()
        procs = [Process(target=worker, args=(shared_dict, shared_list, i))
                 for i in range(5)]
        for p in procs: p.start()
        for p in procs: p.join()
        print("Dict:", dict(shared_dict))
        print("List:", list(shared_list))

Shared Memory — The Modern shared_memory Module 🚄Shared Memory — आधुनिक shared_memory Module 🚄共享内存 — 现代的 shared_memory 模块 🚄

Added in Python 3.8, multiprocessing.shared_memory lets processes share a raw block of memory with no pickling. Together with NumPy, it is the fastest way to share large numeric arrays.

Python 3.8 में जुड़ा multiprocessing.shared_memory processes को बिना pickling के memory का एक कच्चा block बाँटने देता है। NumPy के साथ, बड़े numeric arrays बाँटने का यह सबसे तेज़ तरीक़ा है।

Python 3.8 新增的 multiprocessing.shared_memory 让进程无需 pickle 就能共享一块原始内存。配合 NumPy，它是共享大型数值数组最快的方式。

from multiprocessing import Process
from multiprocessing.shared_memory import SharedMemory
import numpy as np

def worker(shm_name, shape, dtype, start, end):
    # Attach to the existing shared memory block by name
    existing = SharedMemory(name=shm_name)
    arr = np.ndarray(shape, dtype=dtype, buffer=existing.buf)
    arr[start:end] *= 2                      # change in place - no copy, no pickle
    del arr                                  # drop the view before closing
    existing.close()

if __name__ == "__main__":
    # Allocate a 1-million-element float64 array in shared memory
    shape = (1_000_000,)
    dtype = np.float64
    nbytes = np.zeros(shape, dtype=dtype).nbytes
    shm = SharedMemory(create=True, size=nbytes)
    arr = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    arr[:] = np.arange(1_000_000)

    # Start 4 workers, each changing one slice
    chunk = 250_000
    procs = [Process(target=worker,
                     args=(shm.name, shape, dtype, i * chunk, (i + 1) * chunk))
             for i in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()

    print("First 5 :", arr[:5])
    print("Last 5  :", arr[-5:])
    del arr
    shm.close()
    shm.unlink()                             # free the shared memory block

ShareableList — a simpler alternativeShareableList — एक सरल विकल्पShareableList — 更简单的替代方案

Here is the original example, run for real:

यह मूल example है, असल में चलाया गया:

这是原始示例的真实运行结果：

from multiprocessing import Process
from multiprocessing.shared_memory import ShareableList

def worker(name):
    sl = ShareableList(name=name)
    sl[0] = "modified by child"
    sl[1] += 100
    sl.shm.close()

if __name__ == "__main__":
    sl = ShareableList(["hello", 42, 3.14, True])
    print("Before:", list(sl))
    p = Process(target=worker, args=(sl.shm.name,))
    p.start(); p.join()
    print("After :", list(sl))
    sl.shm.close(); sl.shm.unlink()

from multiprocessing import Process
from multiprocessing.shared_memory import ShareableList

def worker(name):
    sl = ShareableList(name=name)
    sl[0] = "modified by child"        # fits: 32 characters were reserved
    sl[1] += 100
    sl.shm.close()

if __name__ == "__main__":
    sl = ShareableList([" " * 32, 42, 3.14, True])   # reserve room for the string
    sl[0] = "hello"
    print("Before:", list(sl))
    p = Process(target=worker, args=(sl.shm.name,))
    p.start(); p.join()
    print("After :", list(sl))
    sl.shm.close(); sl.shm.unlink()

Synchronisation Primitives 🚦Synchronisation Primitives 🚦同步原语 🚦

Just like threads, processes sometimes need to coordinate. multiprocessing provides process-safe versions of the usual tools.

Threads की तरह processes को भी कभी-कभी तालमेल चाहिए। multiprocessing आम औज़ारों के process-safe versions देता है।

和线程一样，进程有时也需要协调。multiprocessing 提供了常用工具的进程安全版本。

from multiprocessing import Process, Event, Barrier
import time
import random

def worker(worker_id, start_event, barrier):
    start_event.wait()                          # wait for the green light
    print(f"[W{worker_id}] phase 1 starting")
    time.sleep(random.uniform(0.5, 2))
    print(f"[W{worker_id}] phase 1 done, waiting at barrier")
    barrier.wait()                              # wait for all workers
    print(f"[W{worker_id}] phase 2 starting")

if __name__ == "__main__":
    N = 3
    start_event = Event()
    barrier     = Barrier(N)
    procs = [Process(target=worker, args=(i, start_event, barrier))
             for i in range(N)]
    for p in procs: p.start()
    time.sleep(1)                               # let workers set up
    print("All workers ready. GO!")
    start_event.set()                           # release all at once
    for p in procs: p.join()

No worker started phase 1 before "GO!", and no worker started phase 2 until all three had finished phase 1 — even though their phase-1 times were random.

"GO!" से पहले किसी worker ने phase 1 शुरू नहीं किया, और तीनों के phase 1 पूरा करने तक किसी ने phase 2 शुरू नहीं किया — भले ही उनके phase-1 समय random थे।

在"GO!"之前没有任何工作进程开始第一阶段；在三个都完成第一阶段之前，也没有任何一个开始第二阶段 — 即使它们第一阶段的耗时是随机的。

Real-World Example — Parallel Image Processing 🖼️असली उदाहरण — Parallel Image Processing 🖼️实战示例 — 并行图像处理 🖼️

A classic use of multiprocessing: apply a heavy filter to many images. We generated 24 noisy 1024×768 JPEG test images first.

Multiprocessing का एक classic इस्तेमाल: कई images पर भारी filter लगाना। हमने पहले 24 noisy 1024×768 JPEG test images बनाईं।

多进程的经典用法：对大量图片应用繁重的滤镜。我们先生成了 24 张 1024×768 的带噪点 JPEG 测试图片。

from multiprocessing import Pool, cpu_count
from PIL import Image, ImageFilter
from pathlib import Path
import time

INPUT_DIR  = Path("input_images")
OUTPUT_DIR = Path("output_images")
OUTPUT_DIR.mkdir(exist_ok=True)

def process_image(input_path):
    """Load, apply an expensive filter, save."""
    img = Image.open(input_path)
    img = img.filter(ImageFilter.GaussianBlur(radius=5))   # CPU-heavy filter chain
    img = img.filter(ImageFilter.EDGE_ENHANCE_MORE)
    img = img.convert("RGB")
    output_path = OUTPUT_DIR / input_path.name
    img.save(output_path, quality=85)
    return input_path.name

if __name__ == "__main__":
    images = sorted(INPUT_DIR.glob("*.jpg"))
    print(f"Processing {len(images)} images on {cpu_count()} cores")

    start = time.perf_counter()
    for path in images:                      # baseline: one after another
        process_image(path)
    print(f"Sequential: {time.perf_counter() - start:.2f}s")

    start = time.perf_counter()
    with Pool(processes=cpu_count()) as pool:
        # imap_unordered streams results, chunksize batches for efficiency
        for i, name in enumerate(pool.imap_unordered(process_image, images, chunksize=4), 1):
            print(f"  [{i}/{len(images)}] {name}")
    print(f"Pool({cpu_count()}):   {time.perf_counter() - start:.2f}s")

Checking the claim: the same job with threadsदावे की जाँच: वही काम threads से验证说法：用线程做同样的工作

from concurrent.futures import ThreadPoolExecutor
from PIL import Image, ImageFilter
from pathlib import Path
import time

INPUT_DIR, OUTPUT_DIR = Path("input_images"), Path("output_images")
OUTPUT_DIR.mkdir(exist_ok=True)

def process_image(input_path):                # the same function as before
    img = Image.open(input_path)
    img = img.filter(ImageFilter.GaussianBlur(radius=5))
    img = img.filter(ImageFilter.EDGE_ENHANCE_MORE)
    img.convert("RGB").save(OUTPUT_DIR / input_path.name, quality=85)

images = sorted(INPUT_DIR.glob("*.jpg"))
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=2) as ex:     # THREADS, not processes
    list(ex.map(process_image, images))
print(f"2 threads:  {time.perf_counter() - start:.2f}s")

Real-World Example — Parallel Aggregation With Shared Memory 📊असली उदाहरण — Shared Memory के साथ Parallel Aggregation 📊实战示例 — 用共享内存做并行聚合 📊

Compute the mean of every row of a big NumPy array using all cores, without copying the data to each worker. The input lives in shared_memory; the output is a small shared Array. We added one line at the end: the same job as a single NumPy call.

एक बड़े NumPy array की हर row का mean सभी cores से निकालना, हर worker को data copy किए बिना। Input shared_memory में है; output एक छोटी साझा Array है। हमने आख़िर में एक line जोड़ी: वही काम एक अकेली NumPy call से।

用所有核心计算一个大型 NumPy 数组每一行的平均值，而不把数据复制给每个工作进程。输入放在 shared_memory 中；输出是一个小的共享 Array。我们在末尾加了一行：用一次 NumPy 调用完成同样的工作。

from multiprocessing import Process, Array, cpu_count
from multiprocessing.shared_memory import SharedMemory
import numpy as np
import time

def worker(shm_name, shape, dtype, out_arr, start_row, end_row):
    shm = SharedMemory(name=shm_name)
    data = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    for i in range(start_row, end_row):        # row means for our slice
        out_arr[i] = data[i].mean()
    del data
    shm.close()

if __name__ == "__main__":
    # A 20,000 x 1,000 dataset (~160 MB of float64)
    N_ROWS, N_COLS = 20_000, 1_000
    data = np.random.rand(N_ROWS, N_COLS)

    # Move it into shared memory
    shm = SharedMemory(create=True, size=data.nbytes)
    shared = np.ndarray(data.shape, dtype=data.dtype, buffer=shm.buf)
    shared[:] = data

    results = Array('d', N_ROWS)               # shared output - one mean per row

    n_workers = cpu_count()                    # split the rows across cores
    chunk = N_ROWS // n_workers
    start = time.perf_counter()
    procs = []
    for i in range(n_workers):
        s = i * chunk
        e = N_ROWS if i == n_workers - 1 else (i + 1) * chunk
        p = Process(target=worker,
                    args=(shm.name, data.shape, data.dtype, results, s, e))
        procs.append(p); p.start()
    for p in procs: p.join()
    print(f"Parallel:        {time.perf_counter() - start:.3f}s using {n_workers} workers")
    print(f"First 5 row means: {[round(x, 4) for x in results[:5]]}")

    start = time.perf_counter()                # baseline: ONE vectorised NumPy call
    means = data.mean(axis=1)
    print(f"NumPy, 1 process: {time.perf_counter() - start:.3f}s  same answer? {np.allclose(means, results[:])}")
    del shared
    shm.close(); shm.unlink()

Common Pitfalls and Fixes 🩺आम ग़लतियाँ और उनके Fix 🩺常见陷阱与修复 🩺

from multiprocessing import Pool

def process(item):                       # your real work
    if item % 4 == 0:
        raise ValueError(f"bad item {item}")
    return item * 10

def safe_task(item):
    try:
        return {"ok": True, "result": process(item)}
    except Exception as e:
        return {"ok": False, "error": repr(e), "item": item}

if __name__ == "__main__":
    items = range(1, 11)
    with Pool(processes=2) as pool:
        results = pool.map(safe_task, items)
    ok   = [r for r in results if r["ok"]]
    fail = [r for r in results if not r["ok"]]
    print(f"{len(ok)} succeeded, {len(fail)} failed")
    print("failures:", [(r["item"], r["error"]) for r in fail])

🩺 Simulator: Bug Doctor🩺 Simulator: Bug Doctor🩺 模拟器：Bug 医生

Six real multiprocessing mistakes. Buggy code on the left, fixed code on the right, changed lines marked. Every output is from a real run.

छह असली multiprocessing ग़लतियाँ। बाईं ओर buggy code, दाईं ओर ठीक किया हुआ, बदली lines चिह्नित। हर output असली run से है।

六个真实的多进程错误。左边是有 bug 的代码，右边是修正后的代码，改动的行已标出。每个输出都来自真实运行。

Choosing the Right Tool — Decision Table 🧭सही औज़ार चुनना — फ़ैसले की Table 🧭选择合适的工具 — 决策表 🧭

Golden Rules and Quiz 🏆सुनहरे नियम और Quiz 🏆黄金法则与小测验 🏆

Quick Quiz — 8 Questionsछोटा Quiz — 8 सवाल小测验 — 8 道题