Mô hình Producer-Consumer bất đồng bộ với asyncio.Queue và Sentinel Shutdown

Mục tiêu kiến thức

• Triển khai mô hình đa tiến trình con bất đồng bộ (Async Producer-Consumer Architecture).
• Sử dụng asyncio.Queue có giới hạn kích thước (bounded queue) để điều tiết lưu lượng (Backpressure).
• Sử dụng tín hiệu kết thúc (Sentinel Token None) để dừng workers một cách có trật tự (Graceful Shutdown).

Mô tả bài toán

Viết hàm coroutine bất đồng bộ: run_async_pipeline(items: list[dict], num_workers: int, transform_delay: float = 0.01) -> list[dict]:

• items: Danh sách các dictionary chứa {"id": int, "value": int}.
• Thiết lập một asyncio.Queue(maxsize=max(num_workers * 2, 10)).
• Producer coroutine:
• Đẩy từng item trong items vào queue.
• Sau khi đẩy hết, đẩy num_workers giá trị None (sentinel tokens) vào queue để báo hiệu cho từng worker dừng lại.
• Worker coroutine:
• Liên tục lấy item từ queue bằng await queue.get().
• Nếu item is None, gọi queue.task_done() rồi thoát vòng lặp.
• Nếu có dữ liệu:
• Giả lập độ trễ: await asyncio.sleep(transform_delay).
• Tính toán: tạo dict mới {"id": item["id"], "transformed": item["value"] ** 2, "worker_id": worker_id}.
• Lưu kết quả vào danh sách kết quả chung.
• Gọi queue.task_done().
• Thu thập toàn bộ kết quả, sắp xếp lại theo id tăng dần và trả về.

Nếu num_workers <= 0, raise ValueError("num_workers phai lon hon 0").

Ràng buộc

• Thời gian chạy tối đa: 2000ms.
• Giới hạn bộ nhớ: 256MB.
• Dữ liệu đầu vào tuân thủ đúng kiểu dữ liệu và miền giá trị được mô tả.