# Python chậm ở đâu: chọn thread, process hay asyncio khi xử lý dữ liệu của khách

> Thêm thread vào một script chậm có khi chẳng nhanh hơn chút nào. Trước khi chọn công cụ, bạn phải biết chương trình đang chờ hay đang tính.

Bản gốc: https://fdetimes.net/vi/bach-khoa/python-gil-thread-process-asyncio-du-lieu-khach/

Khách gửi bạn một file 200.000 dòng giao dịch. Với mỗi dòng, script phải gọi API của khách để lấy thêm thông tin, chuẩn hóa vài trường văn bản, rồi tính tổng theo từng chi nhánh. Chạy thử thì mất cả buổi chiều. Bạn bọc vòng lặp vào `ThreadPoolExecutor` và chạy lại: có lúc nhanh hẳn, có lúc gần như không đổi.

Kết quả thất thường đó không phải ngẫu nhiên. Nó cho thấy bạn đã chọn công cụ trước khi biết chương trình mất thời gian vào việc gì. Ở chỗ khách, sai lầm này có cái giá thật: bạn hứa một con số trong buổi demo, rồi nhận ra mình không bao giờ đạt được nó.

Cách làm đúng đi theo thứ tự ngược lại. Trước hết đo xem chương trình đang chờ hay đang tính, sau đó mới chọn thread, process hay asyncio. Mỗi lựa chọn có một giới hạn, và tài liệu chính thức của Python ghi rất rõ các giới hạn đó.

## GIL chỉ chặn một thứ: chạy mã Python song song

Tài liệu `threading` của Python viết thẳng: với GIL, mỗi lúc chỉ một thread được chạy bytecode Python, nên thread không đem lại bao nhiêu hiệu năng cho việc nặng CPU. Câu này rất hay bị hiểu sai thành "thread trong Python vô dụng".

Thật ra GIL chỉ chặn việc *chạy mã Python* cùng lúc. Cũng tài liệu đó khẳng định thread đặc biệt hữu ích cho tác vụ I/O-bound như thao tác file hay gọi mạng, nơi phần lớn thời gian là chờ tài nguyên bên ngoài. Trong lúc một thread chờ phản hồi HTTP, thread khác vẫn làm việc được.

Với tác vụ CPU-bound, tài liệu khuyên dùng `multiprocessing` hoặc `ProcessPoolExecutor`. Theo tài liệu `multiprocessing`, cách này né GIL bằng tiến trình con thay vì thread.

Có một ngoại lệ quan trọng với người làm dữ liệu. Tài liệu NumPy cho biết NumPy nhả GIL ở nhiều phép toán cấp thấp, nên các thread dành phần lớn thời gian trong mã cấp thấp đó sẽ chạy song song thật. Riêng phép toán trên mảng `dtype=np.object_` thì không nhả GIL.

**Điểm mấu chốt:** Đo xem chương trình đang chờ hay đang tính, rồi mới chọn thread hay process.

## Hai chiếc đồng hồ cho bạn biết nên chọn gì

Cách chẩn đoán đơn giản nhất là so hai đồng hồ. `time.perf_counter()` đo thời gian thực trôi qua (wall time), kể cả lúc chương trình ngủ. `time.process_time()` chỉ đo thời gian CPU, không tính lúc ngủ.

Từ định nghĩa đó có thể đọc kết quả như sau. Nếu hai số gần bằng nhau, gần như suốt thời gian ấy CPU đang làm việc. Nếu CPU time nhỏ hơn wall time nhiều, phần chênh lệch là lúc chương trình không dùng CPU, thường là đang chờ mạng hoặc ổ đĩa.

```python
import time

def measure(label, fn, *args):
    t0, c0 = time.perf_counter(), time.process_time()
    result = fn(*args)
    wall = time.perf_counter() - t0
    cpu = time.process_time() - c0
    print(f"{label}: wall={wall:.1f}s cpu={cpu:.1f}s")
    return result
```

Phép đo này phải chạy trên phiên bản tuần tự, chưa có thread hay process. Lý do: `process_time()` chỉ tính CPU của chính tiến trình đang đo, nên việc đã đẩy sang tiến trình con sẽ không hiện ra trong con số đó, còn nhiều thread cùng tính thì cộng dồn CPU lại. Mục đích ở đây là phân loại từng giai đoạn trước khi bạn sửa gì.

Bọc từng giai đoạn của pipeline bằng hàm này. Thử hình dung bạn nhận được kết quả như sau. Giai đoạn gọi API có wall rất lớn nhưng CPU gần bằng không. Giai đoạn chuẩn hóa bằng regex và vòng `for` có wall xấp xỉ CPU, giai đoạn tổng hợp bằng NumPy cũng vậy. Ba giai đoạn này cần ba cách xử lý khác nhau.

| Giai đoạn | Thời gian đi đâu | Công cụ hợp lý |
|---|---|---|
| Gọi API, đọc database, tải file | Chờ mạng hoặc ổ đĩa | asyncio hoặc ThreadPoolExecutor |
| Vòng lặp, regex, dict bằng Python thuần | Tính bằng mã Python, bị GIL giữ | ProcessPoolExecutor |
| Phép toán trên mảng NumPy lớn | Tính trong mã cấp thấp đã nhả GIL | Thread, hoặc để NumPy tự lo |

## Làm mẫu: một pipeline, ba công cụ

Bắt đầu với giai đoạn gọi API và làm phép tính thử. Giả sử mỗi lời gọi mất 200 ms, và bạn cần đúng 200.000 lời gọi, mỗi dòng một lời gọi: chạy tuần tự hết 40.000 giây, hơn 11 giờ.

Nếu API của khách chịu được 20 kết nối cùng lúc, về lý thuyết bạn chỉ còn khoảng 2.000 giây, vì gần như toàn bộ thời gian ấy là chờ.

Bước tiếp theo là chỗ dễ sai. Bạn đã có event loop asyncio, nên rất dễ gọi luôn hàm chuẩn hóa ngay trong coroutine. Tài liệu asyncio cảnh báo: nếu một hàm tính toán nặng chạy 1 giây, mọi Task và thao tác I/O đồng thời khác đều bị trễ 1 giây.

Cách đúng là đẩy phần tính toán sang process pool qua `loop.run_in_executor`, để loop chỉ lo việc chờ. Đoạn code dưới đây là bản phác để minh họa cấu trúc: `load_ids` là chỗ bạn nối với file thật của khách, và trước khi chạy ở môi trường của khách bạn vẫn cần thêm retry, log và xử lý lỗi.

```python
import asyncio
from concurrent.futures import ProcessPoolExecutor
import httpx

API_URL = "https://api.example-customer.com"  # thay bằng địa chỉ thật của khách
CHUNK = 5000

def clean_row(row):            # ví dụ: chuẩn hóa tên chi nhánh
    row["branch"] = row["branch"].strip().upper()
    return row

def normalise(batch):          # hàm cấp module, có thể pickle
    return [clean_row(r) for r in batch]

async def fetch(client, sem, rid):
    async with sem:            # giới hạn tải lên API của khách
        r = await client.get(f"/records/{rid}")
        r.raise_for_status()
        return r.json()

async def main(ids):
    sem = asyncio.Semaphore(20)
    loop = asyncio.get_running_loop()
    pending = []
    with ProcessPoolExecutor() as pool:
        async with httpx.AsyncClient(base_url=API_URL, timeout=30) as client:
            for i in range(0, len(ids), CHUNK):   # mỗi lần chỉ 5.000 coroutine
                chunk = ids[i:i + CHUNK]
                rows = await asyncio.gather(*(fetch(client, sem, rid) for rid in chunk))
                # chuẩn hóa lô này trong process pool, trong lúc lô sau đang gọi API
                pending.append(loop.run_in_executor(pool, normalise, rows))
        parts = await asyncio.gather(*pending)
    return [row for part in parts for row in part]

if __name__ == "__main__":
    ids = load_ids("transactions.csv")   # bạn tự viết: đọc 200.000 id từ file của khách
    rows = asyncio.run(main(ids))
```

Có ba chi tiết cần để ý. `normalise` và `clean_row` nằm ở cấp module chứ không phải lambda, vì `ProcessPoolExecutor` chỉ làm việc với đối tượng có thể pickle, cả hàm gửi đi lẫn kết quả trả về.

Dữ liệu được chia thành 40 lô, mỗi lô 5.000 dòng, thay vì gửi từng dòng: Itamar Turner-Trauring ở pythonspeed.com chỉ ra rằng chuyển dữ liệu giữa các tiến trình có thể rất chậm, có khi chậm hơn cả phần tính toán.

Chi tiết thứ ba là vòng lặp theo lô: chương trình không tạo 200.000 coroutine cùng một lúc mà đi từng khúc 5.000 id. Bộ nhớ vì thế được giữ trong tầm kiểm soát, và nếu đến giữa chừng API của khách trả lỗi, bạn biết mình đang ở lô nào.

Giai đoạn cuối, khi dữ liệu đã thành mảng số, là lúc thread có ích. Chẳng hạn bạn cần quy đổi số tiền theo tỷ giá rồi cộng theo chi nhánh. Phép nhân trên mảng chạy trong mã cấp thấp của NumPy, nên có thể chia mảng thành vài khúc và cho mỗi khúc một thread:

```python
import numpy as np
from concurrent.futures import ThreadPoolExecutor

def convert(amounts, rates):
    return amounts * rates     # phép toán trên mảng, không có vòng for Python

def convert_parallel(amounts, rates, n=4):
    a_parts = np.array_split(amounts, n)
    r_parts = np.array_split(rates, n)
    with ThreadPoolExecutor(max_workers=n) as pool:
        parts = list(pool.map(convert, a_parts, r_parts))
    return np.concatenate(parts)

# amounts, rates, branch_ids: các mảng NumPy bạn dựng từ rows ở bước trước
converted = convert_parallel(amounts, rates)
totals = np.bincount(branch_ids, weights=converted)
```

Với 200.000 dòng, phần này có thể đã đủ nhanh mà không cần thread. Kỹ thuật này đáng dùng khi mảng lớn hơn nhiều. Cách kiểm tra đơn giản: đo wall time của `convert` chạy một lần trên cả mảng, rồi đo `convert_parallel`. Nếu con số không giảm, thread không mang lại gì và bạn nên bỏ nó đi.

Ngược lại, nếu "phép toán NumPy" của bạn thực chất là vòng `for` đi qua từng phần tử, thread sẽ không giúp được gì. Vòng lặp đó vẫn là mã Python và vẫn phải giữ GIL.

## Năm bước tự làm ở dự án tiếp theo

1. Đo wall time và CPU time cho từng giai đoạn trên bản tuần tự, ghi lại kết quả trước khi sửa gì.
2. Với giai đoạn chờ, dùng asyncio nếu thư viện client hỗ trợ async, nếu không thì dùng `ThreadPoolExecutor`. Luôn đặt giới hạn số kết nối, vì hệ thống của khách không được thiết kế để chịu bài kiểm tra tải của bạn.
3.

Với giai đoạn tính bằng Python thuần, dùng `ProcessPoolExecutor` và gửi dữ liệu theo lô.
4. Với giai đoạn NumPy, kiểm tra xem code có thực sự là phép toán trên mảng không, và mảng có phải `dtype=object` không, rồi mới thử thread.
5. Đo lại wall time bằng đúng hàm đã dùng ở bước một, và báo cho khách con số trước và sau.

Con số trước và sau đó cũng là thứ bạn nên đưa vào CV. Một dòng như "Rút thời gian xử lý file của khách từ X xuống Y bằng cách tách phần chờ API sang asyncio và phần chuẩn hóa sang process pool" cho người tuyển dụng thấy bạn biết chẩn đoán, chứ không chỉ biết thêm thread.

Nếu buổi phỏng vấn có hỏi về GIL, một ví dụ thật như vậy bao giờ cũng thuyết phục hơn một định nghĩa học thuộc.

## Những lỗi gặp đi gặp lại

Lỗi phổ biến nhất là đưa lambda hoặc hàm viết vội trong notebook vào process pool. Vì `ProcessPoolExecutor` chỉ làm việc với đối tượng có thể pickle, cách an toàn là chuyển hàm ra một file `.py` ở cấp module rồi import vào.

Lỗi thứ hai là để code tính toán lọt vào event loop mà không ai hay. Mỗi giây tính toán ở đó là một giây mọi Task khác phải đứng chờ. Khi service asyncio chậm bất thường, hãy đo CPU time quanh các coroutine trước khi đổ lỗi cho mạng.

Lỗi thứ ba là trộn process với thư viện đã đa luồng sẵn. Tài liệu Polars nhắc rằng Polars tự chạy đa luồng, và fork một tiến trình đa luồng có thể gây deadlock, nên khuyên dùng `spawn` vì đây là cách an toàn nhất trên mọi nền tảng.

Cái giá của `spawn` là khởi động tiến trình khá chậm. Đó là thêm một lý do để tạo pool một lần, dùng lại cho nhiều lô, thay vì mở tiến trình mới cho từng việc nhỏ.

## Còn bản Python không có GIL thì sao?

Từ Python 3.13, CPython có bản build free-threaded tắt GIL. PEP 779 đặt tiêu chí cho giai đoạn II, khi bản build này được hỗ trợ chính thức nhưng vẫn là tùy chọn; việc biến nó thành mặc định để dành cho một PEP sau. Nghe hấp dẫn, nhưng chưa nên là lựa chọn đầu tiên cho hệ thống của khách.

Tài liệu Python cho biết bản free-threaded có thêm chi phí. Trên bộ benchmark pyperformance, mức overhead trung bình dao động từ khoảng 1% trên macOS aarch64 đến 8% trên Linux x86-64.

Nguy hiểm hơn, một số package bên thứ ba có extension module chưa sẵn sàng sẽ bật GIL trở lại.

Nếu muốn thử, hãy liệt kê mọi thư viện có extension trong môi trường của khách và kiểm tra từng cái trước. Sau đó đo lại bằng đúng cặp đồng hồ ở trên. Công cụ có thể thay đổi, nhưng câu hỏi đầu tiên vẫn vậy: chương trình của bạn đang chờ hay đang tính?

**Thử ngay tuần này:**

- Lấy một script xử lý dữ liệu đang chạy chậm, in time.perf_counter() và time.process_time() quanh từng giai đoạn để biết phần nào chờ, phần nào tính.
- Rà các mảng NumPy trong pipeline và tìm những mảng có dtype=object: đó là chỗ thread sẽ không giúp được gì.
- Trong ProcessPoolExecutor, viết lại một đoạn đang dùng lambda thành hàm định nghĩa ở cấp module, gửi dữ liệu theo lô, rồi đo thời gian trước và sau.

## Nguồn

- [threading — Thread-based parallelism — Python documentation](https://docs.python.org/3/library/threading.html)

- [multiprocessing — Process-based parallelism — Python documentation](https://docs.python.org/3/library/multiprocessing.html)

- [Developing with asyncio — Python documentation](https://docs.python.org/3/library/asyncio-dev.html)

- [concurrent.futures — Launching parallel tasks — Python documentation](https://docs.python.org/3/library/concurrent.futures.html)

- [Python support for free threading — Python documentation](https://docs.python.org/3/howto/free-threading-python.html)

- [PEP 779 – Criteria for supported status for free-threaded Python](https://peps.python.org/pep-0779/)

- [Thread Safety — NumPy Manual](https://numpy.org/doc/stable/reference/thread_safety.html)

- [time — Time access and conversions — Python documentation](https://docs.python.org/3/library/time.html)

- [Python's multiprocessing performance problem (Itamar Turner-Trauring, pythonspeed.com)](https://pythonspeed.com/articles/faster-multiprocessing-pickle/)

- [Multiprocessing — Polars user guide](https://docs.pola.rs/user-guide/misc/multiprocessing/)