Semaphore và cách sử dụng trong lập trình đa luồng

Semaphore là một cơ chế đồng bộ được sử dụng để điều phối quyền truy cập của nhiều luồng/tiến trình vào một số lượng tài nguyên hữu hạn hoặc để đồng bộ luồng trong môi trường đa luồng.

1. Cách thức hoạt động của semaphore

Semaphore là một tập hợp các giấy phép (permits).

Để truy cập tài nguyên thi luồng phải yêu cầu một giấy phép. Nếu còn phép thì semaphore sẽ cho luồng đi tiếp và giảm số lượng giấy phép đi 1.

Nếu hết giấy phép thì luồng sẽ phải chờ cho đến khi có môt luồng khác trả lại giấy phép.

Sau khi luồng dùng xong sẽ trả lại giấy phép, khi đó số lượng giấy phép của semaphore sẽ tăng lên 1.

Một luồng sẽ thao tác với semaphore ở 2 trạng thái dưới:

• Wait: luồng kiểm tra semaphre, nếu còn giấy phép thì giảm giá số giá trị xuống 1, nếu không còn giấy phép thì đưa thread vào wait queue (trạng thái ngủ (sleep) không tiêu tốn CPU)

• Signal: luồng xử lý xong nếu có luồng đang wait thì đánh thức luồng đó, nếu không có luồng nào đang wait ở trong wait queue thì tăng semaphore lên 1.

2. Các loại semaphore

Binary Semaphore

Binary semaphore là semaphore chỉ có giá trị {0, 1} và chủ yếu được sử dụng để đồng bộ luồng (synchronization).

Chúng ta xem xét bài toán dùng binary semaphore để đồng bộ giữa ngắt của UART và task xử lý dữ liệu nhận trong hệ điều hành thời gian thực RTOS. Ban đầu, semaphore có giá trị khởi tạo là 0, Task này sẽ wait semaphore vô thời hạn trong vòng lặp. UART nhận dữ liệu xong sẽ phát sinh một ngắt nhận. Ngắt này sẽ phát tín hiệu signal cho task nhận dữ liệu.

Nhờ đó, task xử lý dữ liệu chỉ được đánh thức khi có ngắt nhận từ UART, sau khi xử lý xong task lại tiếp tục gọi wait và quay lại trạng thái sleep (blocked/waiting) để không gây tiêu tốn CPU trong thời gian chờ.

Counting Semaphore

Sử dụng khi có một tập hợp các tài nguyên hữu hạn.

Ví dụ về counting semaphore như sau:

Giả sử chúng ta có 2 phòng tắm (tài nguyên) khi đó semaphore ban đầu bằng 2.

Khi có người thứ nhất vào phòng, semaphore giảm xuống 1.

Khi có người thứ hai vào phòng, semaphore giảm xuống 0.

Khi người thứ 2 muốn vào phòng phải chờ cho 1 trong 2 người ra khỏi phòng mới được vào.

Như vậy counting semaphore giống như người điều tiết giao thông, đảm bảo không có quá nhiều luồng cùng vào một khu vực gây nghẽn hệ thống (concurrency limit).

Khi các luồng đã vượt qua cổng và vào được bên trong bằng giấy phép được cấp bởi semaphore nếu cùng tác động lên 1 biến, cấu trúc dữ liệu hay tài nguyên dùng chung thì semaphore không còn bảo vệ được nữa. Khi đó chúng ta cần dùng khóa Mutex hoặc các cơ chế tương đương để đảm bảo không gây ra Race Condition làm sai lệnh dữ liệu.

3. Ví dụ triển khai semaphore trong C++

Binary semaphore chủ yếu được ứng dụng trong lập trình nhúng và hệ thống thời gian thực (RTOS), trong phạm vi lập trình ứng dụng ở tầng user space thì counting semaphore được sử dụng phổ biến hơn để giới hạn mức độ song song nhiều luồng.

C++ 20 trở đi đã cung cấp counting_semaphore trong thư viện tiêu chuẩn STL.

Chú ý: để biên dịch trên vscode thì cần thêm cờ biên dịch “-std=c++20” và cấu hình biên dịch là c++20 trong cấu hình json c/c++ (nhấn tổ hợp phím Ctrl + Shift + P, sau đó chọn C/C++ Edit Configulations (JSON), cấu hình cppStandard là “c++20“).

{
    "configurations": [
        {
            "name": "Win32",
            "includePath": [
                "${workspaceFolder}/**"
            ],
            "defines": [
                "_DEBUG",
                "UNICODE",
                "_UNICODE"
            ],
            "windowsSdkVersion": "10.0.26100.0",
            "compilerPath": "cl.exe",
            "cStandard": "c17",
            "cppStandard": "c++20",
            "intelliSenseMode": "windows-msvc-x64"
        }
    ],
    "version": 4
}

#include <semaphore>            // C++ 20
#include <mutex>
#include <thread>
#include <vector>
#include <iostream>
#include <string>
#include <atomic>

constexpr int MAX_RESOURCE  =   2;
constexpr int MAX_THREAD    =   4;

// Khởi tạo semaphore với giá trị ban đầu là 2
std::counting_semaphore<MAX_RESOURCE> sem(MAX_RESOURCE);

// Khai báo biến đếm mô phỏng số permit của semaphore
// Atomic là thao tác không thể chia nhỏ, đảm bảo không xảy ra race condition trong đa luồng, và được hỗ trợ bởi phần cứng CPU
// Lưu ý inside chỉ dùng để quan sát demo, không phải giá trị nội bộ của semaphore.
std::atomic<int> inside{0};

// Định nghĩa hàm safe log để in tuần tự ra console mà không bị dính chữ do tranh chấp
std::mutex log_mutex;
void safe_log(const std::string& msg) {
    std::lock_guard<std::mutex> lock(log_mutex);
    std::cout << msg << std::endl;
}

// Định nghĩa hàm xử lý cho các worker thread
void worker(int id) {
    std::string msg;
    msg = "Worker " + std::to_string(id) + " is waiting ...";
    safe_log(msg);

    // wait (lấy permit, nếu không có sẽ vào trạng thái blocked/wait không tiêu tốn CPU)
    sem.acquire();

    // Lấy permit thành công
    int current = ++inside;

    // do something
    msg = "Worker " + std::to_string(id) + " acquired resource" + "| inside: " + std::to_string(current);
    safe_log(msg);
    std::this_thread::sleep_for(std::chrono::seconds(1));  // sleep 1 giây

    current = --inside;
    msg = "Worker " + std::to_string(id) + " releasing resource" + "| inside: " + std::to_string(current);
    safe_log(msg);

    // signal (trả permit, đánh thức 1 thread đang wait nếu có)
    sem.release();
}

int main() {
    // định nghĩa vector để lưu handle của các thread
    std::vector<std::thread> threads;

    // khởi tạo các thread và nạp các handle của thread vào vector
    // lưu ý không dùng push_back trong trường hợp này vì handle của thread là đối tượng không thể sao chép
    for (int i = 0; i < MAX_THREAD; i++)
    {
        threads.emplace_back(worker, i);
    }

    // chờ cho các thread hoàn thành
    for(auto& t: threads) {
        t.join();
    }

    return 0;
}

Ở ví dụ trên, có 4 luồng chạy song song và có chia sẻ 2 nguồn tài nguyên.
Từ kết quả chúng ta có thể thấy tại một thời điểm chỉ có tối đa 2 luồng được sử dụng tài nguyên, điều này thể hiện ở biến inside.

The End.