Khi làm việc với các hệ thống kết hợp (như camera/phần cứng chạy trên threading.Thread và server xử lý mạng/WebSocket chạy trên asyncio), việc gọi trực tiếp event.set() từ Thread độc lập sang asyncio.Event sẽ phá vỡ tính toàn vẹn trạng thái nội bộ của Event Loop vì các cấu trúc dữ liệu của asyncio vốn không hỗ trợ đa luồng (non-thread-safe).

Phương thức loop.call_soon_threadsafe() là cầu nối tiêu chuẩn giúp giải quyết triệt để vấn đề này.

1. Cơ chế hoạt động của loop.call_soon_threadsafe()

Khi bạn gọi loop.call_soon_threadsafe(callback, *args) từ một Thread phụ:

1. Nó đẩy callback (hàm cần thực thi) vào hàng đợi công việc của Event Loop một cách an toàn bằng khóa Mutex nội bộ cấp độ C.

2. Nó kích hoạt cơ chế ngắt tự đánh thức của hệ điều hành (thông qua cơ chế ghi 1 byte vào một self-pipe hoặc eventfd) để đánh thức Event Loop dậy ngay lập tức nếu Event Loop đang ở trạng thái ngủ (đang bị chặn bởi epoll/kqueue/select).

3. Khi tỉnh dậy, chính luồng chính (Main Thread chứa Event Loop) sẽ lấy hàm đó ra và thực thi trực tiếp, đảm bảo không có tranh chấp dữ liệu (Race Condition).

2. Kịch bản thực hành: Thread Camera kích hoạt Asyncio Event

Kịch bản mô phỏng:

• hardware_camera_thread (Chạy trên OS Thread riêng): Mô phỏng thiết bị camera quét hình ảnh vật lý. Khi phát hiện một sự kiện (ví dụ: phát hiện vật thể), nó cần thông báo cho hệ thống Web.

• websocket_broadcaster (Coroutine chạy trên Event Loop): Đang chờ (await event.wait()) để phát gói tin thông báo ra ngoài mạng.

import asyncio
import threading
import time

def hardware_camera_thread(loop: asyncio.AbstractEventLoop, motion_event: asyncio.Event):
    """Luồng phần cứng độc lập (OS Thread)."""
    print(f"[{time.strftime('%X')}] 📷 [Camera Thread]: Cảm biến phần cứng đã khởi động.")
    
    # Mô phỏng camera quét mất 2 giây mới phát hiện có chuyển động
    time.sleep(2)
    print(f"\n[{time.strftime('%X')}] 🚨 [Camera Thread]: PHÁT HIỆN CHUYỂN ĐỘNG!")
    
    # SAI LẦM NẾU GỌI TRỰC TIẾP:
    # motion_event.set()  <-- Nguy cơ làm hỏng trạng thái hoặc bỏ sót sự kiện!
    
    # CÁCH CHUẨN XÁC:
    # Ủy thác cho Event Loop tự gọi hàm motion_event.set trong luồng của chính nó
    loop.call_soon_threadsafe(motion_event.set)
    print(f"[{time.strftime('%X')}] 📨 [Camera Thread]: Đã gửi tín hiệu thread-safe tới Event Loop.")


async def websocket_broadcaster(motion_event: asyncio.Event):
    """Coroutine tiêu thụ chạy trên Event Loop chính."""
    print(f"[{time.strftime('%X')}] 🌐 [WebSocket Server]: Đang chờ tín hiệu từ Camera...")
    
    # Coroutine tạm dừng tại đây, hoàn toàn không chiếm dụng CPU
    await motion_event.wait()
    
    print(f"[{time.strftime('%X')}] 🚀 [WebSocket Server]: Đã nhận được cờ hiệu! Phát thông báo tới Client: 'MOTION_ALERT'")
    
    # Xóa cờ để sẵn sàng đón nhận sự kiện tiếp theo
    motion_event.clear()


async def main():
    # 1. Lấy tham chiếu đến Event Loop đang vận hành
    loop = asyncio.get_running_loop()
    
    # 2. Khởi tạo asyncio.Event gắn liền với Event Loop này
    motion_event = asyncio.Event()
    
    # 3. Chạy coroutine lắng nghe trên Event Loop
    listener_task = asyncio.create_task(websocket_broadcaster(motion_event))
    
    # 4. Khởi chạy Thread phần cứng bên ngoài, truyền loop và event vào làm tham số
    cam_thread = threading.Thread(
        target=hardware_camera_thread,
        args=(loop, motion_event),
        daemon=True
    )
    cam_thread.start()
    
    # Chờ coroutine xử lý xong
    await listener_task
    cam_thread.join()
    print(f"[{time.strftime('%X')}] ✅ Hệ thống hoàn thành chu trình an toàn.")

if __name__ == "__main__":
    asyncio.run(main())

3. Kết quả thực thi trên Terminal

[14:20:00] 🌐 [WebSocket Server]: Đang chờ tín hiệu từ Camera...
[14:20:00] 📷 [Camera Thread]: Cảm biến phần cứng đã khởi động.

[14:20:02] 🚨 [Camera Thread]: PHÁT HIỆN CHUYỂN ĐỘNG!
[14:20:02] 📨 [Camera Thread]: Đã gửi tín hiệu thread-safe tới Event Loop.
[14:20:02] 🚀 [WebSocket Server]: Đã nhận được cờ hiệu! Phát thông báo tới Client: 'MOTION_ALERT'
[14:20:02] ✅ Hệ thống hoàn thành chu trình an toàn.

4. Ba nguyên tắc sống còn khi làm việc xuyên ranh giới Thread - Asyncio

1. Không truyền nhầm tham số:

   • Cú pháp: loop.call_soon_threadsafe(motion_event.set) (chỉ truyền con trỏ hàm, không gọi hàm với dấu ngoặc ()).

   • Nếu hàm cần truyền tham số: loop.call_soon_threadsafe(my_func, arg1, arg2).

2. Không cố truyền một Coroutine:

   • call_soon_threadsafe chỉ nhận hàm thông thường (def).

   • Nếu từ Thread phụ bạn muốn kích hoạt một Coroutine (async def) chạy ngầm trong Event Loop, bạn phải sử dụng:

     Python

     import asyncio
     asyncio.run_coroutine_threadsafe(my_async_func(), loop)
     

3. Phân biệt phạm vi truyền dữ liệu:

   • Dùng loop.call_soon_threadsafe(event.set) khi chỉ cần bật/tắt cờ hiệu nhị phân (Flag).

   • Nếu cần truyền liên tục dữ liệu lớn (như mảng frame ảnh NumPy từ Camera sang Asyncio Server), hãy dùng loop.call_soon_threadsafe(async_queue.put_nowait, frame) với asyncio.Queue.