multiprocessing trong Python cho tác vụ nặng CPU

1. Tổng quan

Khi code Python phải tính toán nhiều, ví dụ xử lý ảnh, tính checksum hay mô phỏng số, bạn sẽ thấy dùng thread gần như không nhanh hơn. Nguyên nhân là GIL (Global Interpreter Lock). Bài này giải thích ngắn gọn vấn đề đó. Sau đó bài hướng dẫn dùng module multiprocessing và ProcessPoolExecutor để chia việc ra nhiều process. Đọc xong, bạn sẽ biết cách song song hóa một tác vụ CPU-bound, viết test cho nó và tránh các lỗi hay gặp.

Mọi ví dụ dùng Python 3.12+ và chỉ cần thư viện chuẩn, không phải cài thêm gì.

2. Vì sao threading không giúp được

2.1. Một tác vụ nặng CPU

Ta dùng một hàm đếm số nguyên tố viết bằng Python thuần. Đây là việc tốn CPU điển hình:

# cpu_task.py
import math

def count_primes(start: int, end: int) -> int:
    count = 0
    for n in range(max(start, 2), end):
        if all(n % d for d in range(2, math.isqrt(n) + 1)):
            count += 1
    return count

Hàm duyệt từng số trong [start, end) và thử chia cho các số từ 2 tới căn bậc hai của nó. Hàm không đọc file, không gọi mạng. Nó chỉ tính toán.

2.2. Đo thời gian: tuần tự và thread

# bench_threads.py
import time
from concurrent.futures import ThreadPoolExecutor
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

def run_sequential() -> int:
    return sum(count_primes(a, b) for a, b in RANGES)

def run_threads() -> int:
    starts, ends = zip(*RANGES)
    with ThreadPoolExecutor(max_workers=8) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    for fn in (run_sequential, run_threads):
        t0 = time.perf_counter()
        result = fn()
        print(f"{fn.__name__}: {result} primes, {time.perf_counter() - t0:.2f}s")

RANGES chia đoạn [0, 1_000_000) thành 8 phần. run_threads giao mỗi phần cho một thread. Kết quả trên một máy 8 nhân (số liệu trên máy bạn sẽ khác):

run_sequential: 78498 primes, 6.41s
run_threads: 78498 primes, 6.58s

Bản dùng thread còn chậm hơn một chút. Trong CPython bản chuẩn, tại mỗi thời điểm chỉ một thread được chạy bytecode Python, vì nó phải giữ GIL. Tám thread chỉ thay phiên nhau dùng một nhân, và còn tốn thêm chi phí chuyển đổi giữa các thread.

Lưu ý: Từ Python 3.13 có bản build free-threaded (python3.13t) bỏ GIL. Bản này vẫn đang ở giai đoạn thử nghiệm, và nhiều thư viện C extension chưa hỗ trợ. Với hầu hết dự án hiện nay, dùng process vẫn là cách chắc chắn nhất.

3. multiprocessing cơ bản

Mỗi process có trình thông dịch riêng và GIL riêng. Vì vậy N process có thể chạy thật sự song song trên N nhân.

3.1. Tạo Process thủ công

# basic_process.py
from multiprocessing import Process, Queue
from cpu_task import count_primes

def worker(start: int, end: int, results: Queue) -> None:
    results.put(count_primes(start, end))

if __name__ == "__main__":
    results = Queue()
    ranges = [(0, 500_000), (500_000, 1_000_000)]
    processes = [Process(target=worker, args=(a, b, results)) for a, b in ranges]

    for p in processes:
        p.start()
    total = sum(results.get() for _ in processes)
    for p in processes:
        p.join()

    print("Total primes:", total)

Giải thích từng phần:

• Process(target=..., args=...) mô tả một process con. Nó chưa chạy cho tới khi bạn gọi start().
• Process con không dùng chung bộ nhớ với process cha. Muốn gửi kết quả về, bạn phải dùng kênh liên lạc. Ở đây là Queue.
• Chương trình gọi results.get() trước join(). Nếu join() trước trong khi queue còn dữ liệu lớn chưa được đọc, chương trình có thể bị deadlock.

Total primes: 78498

Cách này linh hoạt nhưng dài dòng. Trong thực tế bạn hiếm khi tự quản lý Process.

3.2. Dùng Pool

Pool tạo sẵn một nhóm worker và tự phân phối việc cho chúng:

# pool_example.py
from multiprocessing import Pool
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

if __name__ == "__main__":
    with Pool(processes=8) as pool:
        counts = pool.starmap(count_primes, RANGES)
    print(counts)
    print("Total:", sum(counts))

starmap tách mỗi tuple trong RANGES thành các tham số start, end. Kết quả trả về theo đúng thứ tự đầu vào:

[10746, 9895, 9622, 9480, 9637, 9554, 9454, 9610]
Total: 78498

Nếu không truyền processes, Pool dùng os.cpu_count() worker. Khối with sẽ dọn dẹp các worker khi kết thúc.

4. ProcessPoolExecutor: API hiện đại hơn

concurrent.futures.ProcessPoolExecutor có cùng giao diện với ThreadPoolExecutor. Nhờ đó bạn chuyển từ thread sang process gần như chỉ bằng cách đổi tên lớp:

# parallel.py
import os
from concurrent.futures import ProcessPoolExecutor
from cpu_task import count_primes

def split_range(start: int, end: int, parts: int) -> list[tuple[int, int]]:
    step = (end - start + parts - 1) // parts
    return [(s, min(s + step, end)) for s in range(start, end, step)]

def parallel_count_primes(start: int, end: int, workers: int | None = None) -> int:
    workers = workers or os.cpu_count() or 1
    chunks = split_range(start, end, workers)
    starts, ends = zip(*chunks)
    with ProcessPoolExecutor(max_workers=workers) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    import time
    t0 = time.perf_counter()
    print(parallel_count_primes(0, 1_000_000, workers=8))
    print(f"{time.perf_counter() - t0:.2f}s")

78498
1.27s

Thời gian giảm từ khoảng 6,4 giây xuống 1,3 giây. Tốc độ không tăng đúng 8 lần vì hai lý do. Thứ nhất, các đoạn có khối lượng việc không đều: đoạn số lớn thì kiểm tra lâu hơn. Thứ hai, việc khởi động process và truyền dữ liệu cũng tốn thời gian.

Nếu muốn xử lý từng kết quả ngay khi nó xong, hãy dùng submit kết hợp với as_completed:

from concurrent.futures import ProcessPoolExecutor, as_completed
from cpu_task import count_primes

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        futures = {executor.submit(count_primes, a, a + 250_000): a
                   for a in range(0, 1_000_000, 250_000)}
        for future in as_completed(futures):
            print(f"Đoạn bắt đầu {futures[future]}: {future.result()} primes")

Nếu worker ném exception, future.result() sẽ ném lại chính exception đó ở process cha. Vì vậy bạn có thể bắt lỗi bằng try/except như bình thường.

5. Nhiều tác vụ nhỏ: chunksize và imap_unordered

Mỗi lần gửi một tác vụ sang worker, Python phải pickle tham số, gửi qua pipe, rồi unpickle kết quả. Nếu bạn có 100.000 tác vụ rất nhỏ, chi phí này có thể lớn hơn chính phần tính toán. Bạn xử lý bằng cách gom tác vụ thành lô với chunksize:

# chunks.py
from multiprocessing import Pool

def is_prime(n: int) -> bool:
    if n < 2:
        return False
    return all(n % d for d in range(2, int(n ** 0.5) + 1))

if __name__ == "__main__":
    numbers = range(1_000_000)
    with Pool() as pool:
        total = sum(pool.imap_unordered(is_prime, numbers, chunksize=10_000))
    print(total)

• chunksize=10_000 khiến mỗi lần gửi chứa 10.000 số thay vì 1 số. Số lần giao tiếp giữa các process giảm đi rất nhiều.
• imap_unordered trả kết quả theo kiểu lazy và không giữ thứ tự. Kiểu này phù hợp khi bạn chỉ cần tổng hợp, ví dụ cộng dồn, và muốn tiết kiệm bộ nhớ.
• ProcessPoolExecutor.map cũng nhận tham số chunksize. Mặc định của nó là 1, nên bạn nhớ chỉnh lại khi có nhiều tác vụ nhỏ.

6. Khởi tạo tài nguyên một lần cho mỗi worker

Giả sử mỗi tác vụ cần một bảng tra cứu tốn thời gian để tạo. Nếu tạo lại bảng trong mỗi lần gọi hàm thì rất lãng phí. Tham số initializer cho phép chạy một hàm đúng một lần khi worker khởi động:

# initializer_example.py
from concurrent.futures import ProcessPoolExecutor

_price_table: dict[str, float] = {}

def load_prices(currency: str) -> None:
    global _price_table
    rate = 25_000 if currency == "VND" else 1
    _price_table = {f"SKU-{i}": i * 1.5 * rate for i in range(100_000)}

def order_total(skus: list[str]) -> float:
    return sum(_price_table[s] for s in skus)

if __name__ == "__main__":
    orders = [["SKU-1", "SKU-2"], ["SKU-10", "SKU-20", "SKU-30"]]
    with ProcessPoolExecutor(max_workers=2, initializer=load_prices,
                             initargs=("VND",)) as executor:
        print(list(executor.map(order_total, orders)))

[112500.0, 2250000.0]

Biến _price_table là biến global trong từng worker. Process cha không thấy những thay đổi đó. Đây là cách an toàn để mỗi worker có tài nguyên riêng mà không phải gửi bảng tra cứu đi kèm mỗi tác vụ.

7. Viết test cho code song song

Bạn có thể test code dùng process bằng pytest như code thường. Điều kiện là hàm worker phải được định nghĩa ở cấp module:

# test_parallel.py
from cpu_task import count_primes
from parallel import parallel_count_primes, split_range

def test_count_primes_known_value():
    assert count_primes(0, 100) == 25

def test_split_range_covers_whole_interval():
    chunks = split_range(0, 10, 3)
    assert chunks == [(0, 4), (4, 8), (8, 10)]

def test_parallel_matches_sequential():
    expected = count_primes(0, 50_000)
    assert parallel_count_primes(0, 50_000, workers=4) == expected

pip install pytest
pytest -q test_parallel.py

...                                                    [100%]
3 passed in 0.62s

Test quan trọng nhất là test_parallel_matches_sequential. Nó đảm bảo việc chia đoạn không làm sót hay đếm trùng phần tử nào. Đây là lỗi rất hay gặp khi tự viết split_range.

8. Lỗi hay gặp và khi nào không nên dùng

Quên if __name__ == "__main__":. Với start method spawn (mặc định trên Windows và macOS) hoặc forkserver (mặc định trên Linux từ Python 3.14), process con sẽ import lại module chính. Nếu code tạo pool nằm ở cấp module, mỗi process con lại tạo pool mới. Khi đó bạn sẽ gặp RuntimeError hoặc process sinh ra liên tục.

Truyền lambda hoặc hàm lồng nhau. Mọi thứ gửi sang worker đều phải pickle được:

from concurrent.futures import ProcessPoolExecutor

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        print(list(executor.map(lambda x: x * x, range(5))))

_pickle.PicklingError: Can't pickle <function <lambda> at 0x...>: attribute lookup <lambda> on __main__ failed

Bạn hãy định nghĩa hàm bằng def ở cấp module. Nếu cần cố định một số tham số, dùng functools.partial.

Truyền dữ liệu quá lớn. Một list hàng triệu phần tử truyền vào mỗi tác vụ sẽ bị pickle nhiều lần. Thời gian copy có thể lấn át phần tính toán. Có vài cách khắc phục: chỉ gửi chỉ số hoặc đường dẫn file, dùng initializer, hoặc dùng multiprocessing.shared_memory với dữ liệu dạng mảng.

Phụ thuộc vào hành vi của fork. Code chạy được trên Linux cũ nhờ fork (process con kế thừa biến global) có thể lỗi trên macOS, Windows hoặc Python 3.14. Muốn có hành vi nhất quán, hãy chọn rõ start method:

import multiprocessing as mp

if __name__ == "__main__":
    mp.set_start_method("spawn")

Lưu ý: Không nên dùng multiprocessing cho tác vụ I/O-bound như gọi HTTP hay truy vấn database. Khi chờ I/O, thread vẫn nhả GIL, nên ThreadPoolExecutor hoặc asyncio nhẹ hơn nhiều. Cũng không nên dùng multiprocessing khi mỗi tác vụ chỉ mất vài micro giây mà không gom lô, hoặc khi việc tính toán đã nằm trong thư viện C nhả GIL (NumPy, hàm băm trên dữ liệu lớn…). Trong những trường hợp đó, thread hoặc vectorization thường đủ dùng.

Tạo quá nhiều worker. Đặt số worker lớn hơn số nhân CPU không làm code CPU-bound nhanh hơn. Nó chỉ làm tốn RAM, vì mỗi process là một trình thông dịch đầy đủ. Bắt đầu với os.cpu_count() là hợp lý.

9. Kết luận

GIL khiến thread không tăng tốc được code Python thuần nặng CPU. Mỗi process có GIL riêng, nên chia việc ra nhiều process là cách đơn giản để dùng hết các nhân CPU. Với phần lớn trường hợp, ProcessPoolExecutor là lựa chọn gọn nhất. Pool hữu ích khi bạn cần imap_unordered hoặc starmap. Còn Process thủ công chỉ dành cho những tình huống đặc biệt.

Khi dùng multiprocessing, bạn cần nhớ mấy điều sau:

• Luôn có if __name__ == "__main__":.
• Chỉ truyền hàm và dữ liệu pickle được.
• Tăng chunksize khi có nhiều tác vụ nhỏ.
• Dùng initializer cho tài nguyên nặng.
• Viết một test so sánh kết quả song song với kết quả tuần tự để chắc chắn việc chia đoạn đúng.