• Semaphore sử dụng một biến đếm (counter) để theo dõi số lượng tài nguyên đang có sẵn.
• Nó có hai thao tác cơ bản:
  • wait() (hoặc P): Giảm giá trị đếm đi 1. Nếu giá trị đếm bằng 0, tiến trình/luồng sẽ phải đợi cho đến khi có tài nguyên được giải phóng.
  • signal() (hoặc V): Tăng giá trị đếm lên 1 và đánh thức một tiến trình/luồng đang chờ. [1]

• Semaphore nhị phân (Binary Semaphore): Biến đếm chỉ nhận giá trị 0 hoặc 1 (tương tự như một Mutex), dùng để khóa/mở quyền truy cập độc quyền cho một luồng tại một thời điểm.
• Semaphore đếm (Counting Semaphore): Biến đếm có thể nhận các giá trị nguyên lớn hơn 1, cho phép một số lượng giới hạn các luồng cùng truy cập vào tài nguyên.

threading.Semaphore hoạt động như một chiếc máy đếm vé tự động (Counting Semaphore):

• Khi khởi tạo Semaphore(N), hệ thống cấp sẵn $N$ chiếc vé.

• Mỗi khi một thread vào vùng làm việc (with semaphore: hoặc semaphore.acquire()), bộ đếm giảm đi 1 ($N - 1$).

• Nếu bộ đếm về $0$, bất kỳ thread nào đến sau đều phải đứng chờ.

• Khi một thread hoàn thành và bước ra (semaphore.release()), bộ đếm tăng thêm 1, nhường chỗ ngay cho một thread khác đang đợi.

So sánh nhanh:

• Lock tương đương với Semaphore(1): Chỉ duy nhất 1 thread tại một thời điểm (Mutual Exclusion).

• Semaphore(N) với $N > 1$: Cho phép đồng thời tối đa $N$ thread cùng chia sẻ tài nguyên (ví dụ: giới hạn số kết nối database, giới hạn số luồng tải ảnh/video cùng lúc để tránh nghẽn băng thông hoặc tràn RAM).

Bài thực hành: Mô phỏng hệ thống tải đồng thời tối đa 3 Camera

Giả sử ta có 10 nguồn Camera cần kết nối tải dữ liệu, nhưng hạ tầng mạng/phần cứng chỉ chịu tải được tối đa 3 luồng cùng lúc (MAX_CONCURRENT = 3).

import threading
import time
import random

# Thiết lập giới hạn: Tối đa 3 thread được truy cập đồng thời
MAX_CONCURRENT = 3
connection_pool = threading.Semaphore(MAX_CONCURRENT)

def fetch_camera_stream(cam_id):
    print(f"[{time.strftime('%X')}] ⏳ Camera {cam_id:02d}: Đang xếp hàng chờ cấp kết nối...")
    
    # Xin cấp vé (acquire). Nếu đã đủ 3 kết nối, thread sẽ dừng chờ tại đây.
    with connection_pool:
        print(f"[{time.strftime('%X')}] 🟢 Camera {cam_id:02d}: ĐÃ CÓ KẾT NỐI! Đang tải stream...")
        
        # Mô phỏng thời gian tải stream mất từ 1 đến 2 giây
        work_time = random.uniform(1.0, 2.0)
        time.sleep(work_time)
        
        print(f"[{time.strftime('%X')}] 🔴 Camera {cam_id:02d}: Hoàn tất tải sau {work_time:.2f}s. Nhả kết nối!")
    # Tự động trả vé (release) khi ra khỏi khối with

# Tạo danh sách 10 threads mô phỏng 10 camera
threads = []
print(f"=== KHỞI CHẠY HỆ THỐNG VỚI SEMAPHORE = {MAX_CONCURRENT} ===")

for i in range(1, 11):
    t = threading.Thread(target=fetch_camera_stream, args=(i,))
    threads.append(t)
    t.start()

# Chờ tất cả camera hoàn thành
for t in threads:
    t.join()

print(f"[{time.strftime('%X')}] 🎉 Toàn bộ 10 camera đã được xử lý an toàn!")

Phân tích kết quả chạy

Khi chạy chương trình, bạn sẽ quan sát thấy nhịp điệu phân phối tài nguyên:

1. Ngay giây đầu tiên: Chỉ có đúng 3 Camera (ví dụ Cam 01, Cam 02, Cam 03) được in dòng 🟢 ĐÃ CÓ KẾT NỐI!. Bảy camera còn lại ở trạng thái xếp hàng ⏳.

2. Khi một camera bất kỳ xong việc: (ví dụ Cam 02 nhả kết nối), lập tức một camera đang chờ kế tiếp (ví dụ Cam 04) được cấp phép nhảy vào thế chỗ.

3. Số lượng luồng chạy đồng thời bên trong khối with connection_pool không bao giờ vượt quá 3.

Hai lưu ý kỹ thuật quan trọng khi dùng Semaphore

1. Khác biệt giữa Semaphore và BoundedSemaphore:

   • Với threading.Semaphore, nếu bạn vô tình gọi semaphore.release() nhiều lần hơn số lần acquire(), bộ đếm sẽ tăng vượt quá giá trị ban đầu $N$ (gây rò rỉ cấu hình).

   • threading.BoundedSemaphore(N) có cơ chế an toàn: nếu phát hiện gọi release() vượt quá $N$, nó sẽ lập tức ném ra ngoại lệ ValueError. Trong các dự án thực tế, nên ưu tiên dùng BoundedSemaphore.

2. Luôn dùng Context Manager (with semaphore:):

   • Nếu tự gọi semaphore.acquire() và semaphore.release() thủ công bằng tay, nếu đoạn code ở giữa phát sinh lỗi (Exception), lệnh release() sẽ bị bỏ qua $\rightarrow$ Hệ thống bị mất vĩnh viễn 1 suất kết nối (gọi là Resource Leak). Cú pháp with đảm bảo vé luôn được trả lại trong mọi tình huống.

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

threading.Event

threading.Event là cơ chế truyền tín hiệu đơn giản và an toàn nhất giữa các thread trong Python. Nó hoạt động như một chiếc cờ hiệu (Flag) nội bộ với hai trạng thái logic: True (cờ giương lên) hoặc False (cờ hạ xuống).

Các phương thức cốt lõi của Event:

• event.wait(timeout=None): Chặn (block) thread gọi nó và đưa vào trạng thái ngủ cho đến khi cờ được dựng lên (True). Nếu truyền timeout, thread sẽ chỉ chờ tối đa số giây đó rồi đi tiếp.

• event.set(): Giương cờ lên (True). Lập tức đánh thức toàn bộ các thread đang bị chặn ở lệnh event.wait().

• event.clear(): Hạ cờ xuống (False).

• event.is_set(): Kiểm tra trạng thái cờ hiện tại đang là True hay False (không làm dừng thread).

Kịch bản thực hành: Bắt đầu đồng loạt và Dừng khẩn cấp (Emergency Stop)

Bài toán mô phỏng hệ thống 3 camera giám sát:

1. Tín hiệu 1 (start_event): Cả 3 camera khởi động nhưng phải đứng yên tại vạch xuất phát chờ hiệu lệnh từ luồng điều khiển chính để bắt đầu thu hình cùng lúc.

2. Tín hiệu 2 (stop_event): Luồng chính phát hiện sự cố (hoặc người dùng bấm Stop sau 3 giây), giương cờ dừng khẩn cấp để cả 3 camera tự động dọn dẹp và thoát an toàn cùng lúc.

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.

Bạn có muốn tìm hiểu cách bắt sự kiện bấm tổ hợp phím Ctrl + C từ bàn phím (tín hiệu SIGINT) để kích hoạt stop_event.set() giúp tắt chương trình sạch sẽ không?