Multiprocessing و Threading

پایتون سه راه برای concurrency دارد: asyncio (که در فصل قبل دیدیم)، threading و multiprocessing. در این فصل با GIL، انتخاب درست بین آن‌ها، ThreadPoolExecutor، ProcessPoolExecutor، Lock و Queue آشنا می‌شویم.

GIL – Global Interpreter Lock

پایتون CPython در هر لحظه فقط یک thread را اجرا می‌کند (به دلیل GIL). این یعنی threading در پایتون برای کارهای CPU-bound موازی نیست!

Threading

import threading
import time

def task(name, duration):
    print(f"{name} شروع")
    time.sleep(duration)
    print(f"{name} پایان")

# ساخت thread‌ها
t1 = threading.Thread(target=task, args=("Thread 1", 2))
t2 = threading.Thread(target=task, args=("Thread 2", 1))

# شروع
t1.start()
t2.start()

# انتظار برای اتمام
t1.join()
t2.join()

print("همه threads تمام شدند")

کلاس Thread سفارشی

class WorkerThread(threading.Thread):
    def __init__(self, name, work):
        super().__init__()
        self.name = name
        self.work = work
        self.result = None
    
    def run(self):
        """متد اصلی thread"""
        print(f"{self.name} شروع")
        self.result = self.work() * 2
        print(f"{self.name}: {self.result}")

w = WorkerThread("کارگر-1", lambda: 5)
w.start()
w.join()
print(w.result)  # 10

ThreadPoolExecutor

راه مدرن و راحت‌تر برای استفاده از thread‌ها:

from concurrent.futures import ThreadPoolExecutor
import requests

def fetch_url(url):
    response = requests.get(url)
    return url, response.status_code

urls = [
    "https://example.com",
    "https://python.org",
    "https://github.com",
]

# اجرای موازی
with ThreadPoolExecutor(max_workers=5) as executor:
    results = executor.map(fetch_url, urls)
    
    for url, status in results:
        print(url, status)

submit و future

from concurrent.futures import ThreadPoolExecutor, as_completed

def slow_task(n):
    time.sleep(n)
    return n * 2

with ThreadPoolExecutor(max_workers=3) as executor:
    # ارسال job‌ها
    futures = {
        executor.submit(slow_task, i): i
        for i in [1, 3, 2, 5, 4]
    }
    
    # دریافت نتایج به ترتیب اتمام
    for future in as_completed(futures):
        n = futures[future]
        try:
            result = future.result()
            print(f"task {n}: {result}")
        except Exception as e:
            print(f"task {n} خطا: {e}")

Lock – دسترسی همزمان

وقتی چند thread به یک متغیر مشترک دسترسی دارند، باید race condition جلوگیری شود:

import threading

# مشکل race condition
counter = 0

def increment():
    global counter
    for _ in range(100_000):
        counter += 1  # غیر اتمیک!

threads = [threading.Thread(target=increment) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(counter)  # کمتر از 500_000 - اشتباه!

# راه‌حل: Lock
counter = 0
lock = threading.Lock()

def safe_increment():
    global counter
    for _ in range(100_000):
        with lock:  # context manager
            counter += 1

threads = [threading.Thread(target=safe_increment) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(counter)  # 500_000 ✓

RLock – Reentrant Lock

rlock = threading.RLock()

def outer():
    with rlock:
        inner()  # همان thread می‌تواند دوباره lock بگیرد

def inner():
    with rlock:
        print("داخل")

Lock‌های دیگر

• Semaphore – محدود کردن تعداد همزمان
• Event – سیگنال بین thread‌ها
• Condition – wait/notify
• Barrier – synchronization point

Thread-Safe Queue

import queue
import threading
import time

q = queue.Queue()

def producer():
    for i in range(10):
        item = f"آیتم-{i}"
        q.put(item)
        print(f"تولید: {item}")
        time.sleep(0.5)
    q.put(None)  # سیگنال پایان

def consumer():
    while True:
        item = q.get()
        if item is None:
            break
        print(f"مصرف: {item}")
        time.sleep(1)
        q.task_done()

prod = threading.Thread(target=producer)
cons = threading.Thread(target=consumer)
prod.start()
cons.start()
prod.join()
cons.join()

Multiprocessing

برای کارهای CPU-bound، چند پروسه حقیقی (هر کدام GIL خودش):

from multiprocessing import Process
import os

def worker(name):
    print(f"{name} - PID: {os.getpid()}")
    # محاسبه سنگین
    result = sum(i**2 for i in range(10_000_000))
    print(f"{name}: {result}")

if __name__ == "__main__":  # ضروری در Windows
    processes = []
    for i in range(4):
        p = Process(target=worker, args=(f"کارگر-{i}",))
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()

ProcessPoolExecutor

from concurrent.futures import ProcessPoolExecutor
import time

def cpu_intensive(n):
    return sum(i**2 for i in range(n))

if __name__ == "__main__":
    numbers = [10_000_000] * 8
    
    # سنکرون
    start = time.time()
    results = [cpu_intensive(n) for n in numbers]
    print(f"سنکرون: {time.time() - start:.2f}s")  # ~16s
    
    # چندپروسه‌ای
    start = time.time()
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(cpu_intensive, numbers))
    print(f"موازی: {time.time() - start:.2f}s")  # ~4s

اشتراک داده بین پروسه‌ها

from multiprocessing import Process, Value, Array, Lock

# مقدار اشتراکی
counter = Value("i", 0)  # i = integer
lock = Lock()

def increment(counter, lock):
    for _ in range(100_000):
        with lock:
            counter.value += 1

if __name__ == "__main__":
    processes = [
        Process(target=increment, args=(counter, lock))
        for _ in range(4)
    ]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    
    print(counter.value)  # 400_000

# آرایه اشتراکی
shared_array = Array("i", [0, 0, 0, 0, 0])

Manager – دیکشنری/لیست اشتراکی

from multiprocessing import Process, Manager

def worker(shared_dict, shared_list, key, value):
    shared_dict[key] = value
    shared_list.append(value)

if __name__ == "__main__":
    with Manager() as manager:
        shared_dict = manager.dict()
        shared_list = manager.list()
        
        processes = [
            Process(target=worker, args=(shared_dict, shared_list, f"key{i}", i))
            for i in range(5)
        ]
        for p in processes:
            p.start()
        for p in processes:
            p.join()
        
        print(dict(shared_dict))  # {'key0': 0, 'key1': 1, ...}
        print(list(shared_list))

Queue و Pipe بین پروسه‌ها

from multiprocessing import Process, Queue, Pipe

# Queue
def producer(q):
    for i in range(5):
        q.put(f"data-{i}")
    q.put(None)

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"دریافت: {item}")

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

# Pipe (دو طرفه)
def child(conn):
    conn.send("سلام از child")
    print(conn.recv())
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p = Process(target=child, args=(child_conn,))
    p.start()
    print(parent_conn.recv())     # سلام از child
    parent_conn.send("جواب از parent")
    p.join()

مقایسه عملی

import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def io_bound(n):
    time.sleep(1)
    return n

def cpu_bound(n):
    return sum(i*i for i in range(n))

# I/O-bound: threading بهتر
with ThreadPoolExecutor(max_workers=10) as ex:
    list(ex.map(io_bound, range(10)))  # ~1s ✓

with ProcessPoolExecutor(max_workers=10) as ex:
    list(ex.map(io_bound, range(10)))  # ~1s اما overhead بیشتر

# CPU-bound: multiprocessing بهتر
data = [10_000_000] * 4

start = time.time()
with ThreadPoolExecutor() as ex:
    list(ex.map(cpu_bound, data))
print(f"Thread: {time.time()-start:.1f}s")  # ~8s (بدون موازی واقعی)

start = time.time()
with ProcessPoolExecutor() as ex:
    list(ex.map(cpu_bound, data))
print(f"Process: {time.time()-start:.1f}s")  # ~2s ✓

بهترین شیوه‌ها

• I/O-bound کم → threading؛ زیاد → asyncio
• CPU-bound → multiprocessing
• همیشه از concurrent.futures به‌جای Thread/Process دستی
• در multiprocessing روی Windows، if __name__ == "__main__" ضروری است
• shared state را به حداقل برسانید – message passing با Queue ساده‌تر
• Lock به‌اندازه‌ای که لازم است باشد، نه بیشتر

جمع‌بندی

• GIL: threading برای CPU-bound مفید نیست
• ThreadPoolExecutor برای I/O-bound
• ProcessPoolExecutor برای CPU-bound
• Lock برای race condition
• Queue برای ارتباط امن بین thread‌ها/پروسه‌ها
• Manager برای اشتراک داده‌های پیچیده