import threading
import time

# 1. Khởi tạo 2 cờ hiệu
start_event = threading.Event()  # Mặc định ban đầu cờ là False (chưa cho chạy)
stop_event = threading.Event()   # Mặc định ban đầu cờ là False (chưa có lệnh dừng)

def camera_worker(cam_id):
    print(f"[{time.strftime('%X')}] 📷 Cam {cam_id}: Đã sẵn sàng. Đang chờ hiệu lệnh START...")
    
    # BƯỚC A: Chặn thread lại, chờ cờ start_event được set()
    start_event.wait()
    
    print(f"[{time.strftime('%X')}] 🟢 Cam {cam_id}: Nhận cờ START! Bắt đầu thu phát luồng video...")
    
    frame_count = 0
    # BƯỚC B: Vòng lặp kiểm tra liên tục xem có cờ dừng chưa
    while not stop_event.is_set():
        frame_count += 1
        # wait(0.5) vừa mô phỏng độ trễ đọc frame, vừa lắng nghe cờ stop_event
        # Nếu stop_event.set() được gọi, wait() thoát ngay lập tức mà không cần đợi hết 0.5s
        stop_event.wait(0.5)
        
    # BƯỚC C: Nhận tín hiệu dừng -> Dọn dẹp tài nguyên
    print(f"[{time.strftime('%X')}] 🛑 Cam {cam_id}: Nhận tín hiệu DỪNG KHẨN CẤP! Đã thu {frame_count} frames. Đóng kết nối.")

# 2. Khởi chạy 3 worker threads
threads = [threading.Thread(target=camera_worker, args=(i,)) for i in range(1, 4)]
for t in threads:
    t.start()

# Luồng chính chuẩn bị hệ thống (mất 2 giây)
time.sleep(2)
print(f"\n[{time.strftime('%X')}] 📢 [SYSTEM]: Hệ thống nạp xong! PHÁT HIỆU LỆNH BẮT ĐẦU!")
start_event.set()  # Đánh thức đồng loạt cả 3 cameras

# Cho hệ thống chạy thu hình trong 3 giây
time.sleep(3)
print(f"\n[{time.strftime('%X')}] ⚠️ [SYSTEM]: PHÁT HIỆN SỰ CỐ / BẤM NÚT STOP! PHÁT LỆNH DỪNG KHẨN CẤP!")
stop_event.set()   # Báo hiệu đồng loạt cho cả 3 cameras dừng lại

# Chờ tất cả camera hoàn tất việc dọn dẹp
for t in threads:
    t.join()

print(f"[{time.strftime('%X')}] ✅ Toàn bộ hệ thống đã tắt an toàn (Graceful Shutdown).")

Kết quả in ra trên Terminal

[14:03:00] 📷 Cam 1: Đã sẵn sàng. Đang chờ hiệu lệnh START...
[14:03:00] 📷 Cam 2: Đã sẵn sàng. Đang chờ hiệu lệnh START...
[14:03:00] 📷 Cam 3: Đã sẵn sàng. Đang chờ hiệu lệnh START...

[14:03:02] 📢 [SYSTEM]: Hệ thống nạp xong! PHÁT HIỆU LỆNH BẮT ĐẦU!
[14:03:02] 🟢 Cam 1: Nhận cờ START! Bắt đầu thu phát luồng video...
[14:03:02] 🟢 Cam 2: Nhận cờ START! Bắt đầu thu phát luồng video...
[14:03:02] 🟢 Cam 3: Nhận cờ START! Bắt đầu thu phát luồng video...

[14:03:05] ⚠️ [SYSTEM]: PHÁT HIỆN SỰ CỐ / BẤM NÚT STOP! PHÁT LỆNH DỪNG KHẨN CẤP!
[14:03:05] 🛑 Cam 1: Nhận tín hiệu DỪNG KHẨN CẤP! Đã thu 6 frames. Đóng kết nối.
[14:03:05] 🛑 Cam 2: Nhận tín hiệu DỪNG KHẨN CẤP! Đã thu 6 frames. Đóng kết nối.
[14:03:05] 🛑 Cam 3: Nhận tín hiệu DỪNG KHẨN CẤP! Đã thu 6 frames. Đóng kết nối.
[14:03:05] ✅ Toàn bộ hệ thống đã tắt an toàn (Graceful Shutdown).

Điểm kỹ thuật đắt giá: Thay thế time.sleep() bằng event.wait(timeout)

Trong một vòng lặp chạy ngầm thông thường, người ta hay viết:

while not stop_flag:
    do_something()
    time.sleep(5)  # NGUY HIỂM!

Nếu viết như trên, khi người dùng bấm Stop, thread phải đợi ngủ hết 5 giây của time.sleep mới kiểm tra lại điều kiện while để dừng. Ứng dụng sẽ có cảm giác bị treo (unresponsive).

Khi thay bằng stop_event.wait(5):

• Nếu không có biến cố: Thread tạm dừng đúng 5 giây như bình thường.

• Nếu luồng chính gọi stop_event.set() ở giây thứ 1: Lệnh stop_event.wait(5) lập tức mở khóa và trả về ngay, giúp thread dừng trong tích tắc mà không có bất kỳ độ trễ nào.

-----------------------------------------------------------------------------------------

threading.Event và asyncio.Event đều có chung giao diện lập trình (API) và mục đích sử dụng là quản lý một cờ hiệu nội bộ (Internal Flag) nhằm truyền tín hiệu giữa các đơn vị thực thi. Tuy nhiên, cơ chế kích hoạt và bản chất vật lý bên dưới của chúng hoàn toàn khác nhau.

1. Bảng đối chiếu cơ chế hoạt động

2. Chi tiết cơ chế đánh thức bên dưới

A. threading.Event: Đánh thức qua biến điều kiện của OS (Condition Variable)

Khi bạn gọi threading.Event().wait():

1. Thread hiện tại yêu cầu lấy một Lock nội bộ của biến điều kiện (Condition).

2. Nó đăng ký Thread ID của mình vào danh sách chờ của Kernel hệ điều hành.

3. Thread tự giải phóng khóa và đi vào trạng thái ngủ sâu (pthread_cond_wait trong C). CPU hoàn toàn bỏ qua thread này cho đến khi có một luồng khác gọi event.set().

4. Khi event.set() được gọi từ Thread B, hệ điều hành phát tín hiệu đánh thức tất cả các thread đang ngủ để tiếp tục chạy tiếp.

[Thread 1] ──> event.wait() ──> [OS Kernel: Đưa Thread 1 vào trạng thái Sleep]
                                                    ▲
[Thread 2] ──> event.set()  ──> [OS Kernel: Gửi tín hiệu đánh thức Thread 1]

B. asyncio.Event: Đánh thức qua danh sách Future trong Event Loop

Bên trong mã nguồn của asyncio.Event không hề có Mutex Lock hay System Call nào của OS. Nó chỉ là một cấu trúc dữ liệu thuần Python:

# Mô phỏng logic nội bộ của asyncio.Event
class AsyncioEventMock:
    def __init__(self):
        self._value = False
        self._waiters = []  # Danh sách chứa các asyncio.Future

    async def wait(self):
        if self._value:
            return True
        fut = loop.create_future()
        self._waiters.append(fut)
        await fut  # Coroutine nhường quyền tại đây
        return True

    def set(self):
        self._value = True
        for fut in self._waiters:
            if not fut.done():
                fut.set_result(True) # Đánh thức Coroutine thông qua Future
        self._waiters.clear()

Khi bạn gõ await event.wait():

1. Nó tạo một chiếc hộp rỗng Future và cất vào mảng self._waiters.

2. Coroutine của bạn dừng lại tại điểm await fut, nhường CPU cho Event Loop chạy các việc khác.

3. Khi ai đó gọi event.set(), mã nguồn chỉ đơn giản duyệt qua mảng _waiters và gọi fut.set_result(True).

4. Event Loop thấy các Future này đã xong liền đưa các Coroutine tương ứng trở lại hàng đợi sẵn sàng chạy.

3. Cạm bẫy sống còn khi phối hợp giữa Threading và Asyncio

Một lỗi kinh điển xảy ra khi một Thread phụ (ví dụ: Thread kéo RTSP Camera) muốn phát tín hiệu báo cho Event Loop chính (Asyncio Web Server) biết có frame mới bằng cách gọi trực tiếp asyncio_event.set():

# NGUY HIỂM: Gây hỏng trạng thái (Race Condition nội bộ) hoặc Crash
def camera_thread_worker(async_event):
    # Đang chạy trong OS Thread riêng biệt
    async_event.set()  # SAI! asyncio.Event không thread-safe

Cách khắc phục chuẩn: Vì asyncio.Event không an toàn khi gọi xuyên luồng, bạn phải dùng loop.call_soon_threadsafe() để ủy thác việc giương cờ cho chính Event Loop thực thi:

def camera_thread_worker(loop, async_event):
    # Đẩy lệnh set() an toàn xuyên luồng vào Event Loop
    loop.call_soon_threadsafe(async_event.set)