[C++] 同步原語 (Synchronize Primitive)

本文會講解 C++ 中同步原語 (synchronize primitive) 的基本概念。

Thread

為了提高程式的效能，許多程式語言都支持多執行緒 (multi-thread)，而 C++ 也提供了 std::thread 來支援多執行緒。

#include <iostream>
#include <thread>

int count = 0;

void add_count() { count += 1; }

int main() {
    std::thread t1(add_count);
    std::thread t2(add_count);
    t1.join();
    t2.join();

    std::cout << "Printing count: " << count << std::endl;
    return 0;
}

Mutex

在多執行緒的環境下，為了避免多個執行緒同時修改共享資源 (shared resource)，我們可以使用 std::mutex 來保護共享資源。

#include <iostream>
#include <mutex>
#include <thread>

int count = 0;
std::mutex mtx;

void add_count() {
    mtx.lock();
    count += 1;
    mtx.unlock();
}

int main() {
    std::thread t1(add_count);
    std::thread t2(add_count);
    t1.join();
    t2.join();

    std::cout << "Printing count: " << count << std::endl;
    return 0;
}

std::lock_guard

為了避免沒有 unlock 的意外情況，C++11 中提供了一個基於 RAII 機制的 std::lock_guard 來取代 std::mutex 的 lock 和 unlock。

#include <iostream>
#include <mutex>
#include <thread>

int count = 0;
std::mutex mtx;

void add_count() {
    std::lock_guard<std::mutex> lock(mtx);
    count += 1;
}

int main() {
    std::thread t1(add_count);
    std::thread t2(add_count);
    t1.join();
    t2.join();

    std::cout << "Printing count: " << count << std::endl;
    return 0;
}

std::unique_lock

除此之外，C++11 還提供了另一個更靈活的同步原語 std::unique_lock，可以在需要時手動 lock 和 unlock。

#include <iostream>
#include <mutex>
#include <thread>

int count = 0;
std::mutex mtx;

void add_count() {
    std::unique_lock<std::mutex> lock(mtx);
    count += 1;
    lock.unlock();

    // Do some other work

    lock.lock();
    count += 1;
    lock.unlock();
}

int main() {
    std::thread t1(add_count);
    std::thread t2(add_count);
    t1.join();
    t2.join();

    std::cout << "Printing count: " << count << std::endl;
    return 0;
}

std::scoped_lock

到了 C++17，C++ 提供了一種更新的同步原語 std::scoped_lock，可以同時 lock 多個 mutex，並且可以避免死鎖 (deadlock) 的問題。

#include <iostream>
#include <mutex>
#include <thread>

int count = 0;
std::mutex mtx;

void add_count() {
    std::scoped_lock lock(mtx);
    count += 1;
}

int main() {
    std::thread t1(add_count);
    std::thread t2(add_count);
    t1.join();
    t2.join();

    std::cout << "Printing count: " << count << std::endl;
    return 0;
}

Condition Variable

在多執行緒的環境下，有時候我們需要等待某個條件成立後再繼續執行，這時候我們可以使用 std::condition_variable 來達成。如果要使用 std::condition_variable，一定要搭配 std::unique_lock 來使用。

#include <condition_variable>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>

std::mutex mtx;
std::condition_variable cv;
std::queue<int> dataQueue;
bool done = false;

// 生產者：生成數據並加入佇列
void producer(int n) {
    for (int i = 0; i < n; ++i) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));  // 模擬生產時間
        std::unique_lock<std::mutex> lock(mtx);
        dataQueue.push(i);
        std::cout << "Produced: " << i << std::endl;
        cv.notify_one();
    }
    std::unique_lock<std::mutex> lock(mtx);
    done = true;
    cv.notify_all();
}

// 消費者：消耗佇列中的數據
void consumer() {
    while (true) {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, [] { return !dataQueue.empty() || done; });

        while (!dataQueue.empty()) {
            int value = dataQueue.front();
            dataQueue.pop();
            lock.unlock();
            std::cout << "Consumed: " << value << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(150));
            lock.lock();
        }

        if (done) break;
    }
}

int main() {
    std::thread prod(producer, 10);
    std::thread cons(consumer);
    prod.join();
    cons.join();
    return 0;
}

如上所示，我們使用 std::condition_variable 來實現 producer-consumer 模式。

Shared-Mutex & Shared-Lock

在較新版本的 C++ 中，提供了兩個新的同步原語 std::shared_mutex (C++17) 和 std::shared_lock (C++14)，可以用來實現讀寫鎖 (read-write lock)。

#include <iostream>
#include <mutex>
#include <shared_mutex>
#include <thread>

int count = 0;
std::shared_mutex m;

void read_value() {
    std::shared_lock lk(m);
    std::cout << "Reading value " + std::to_string(count) + "\n" << std::flush;
}

void write_value() {
    std::unique_lock lk(m);
    count += 5;
}

int main() {
    std::thread t1(read_value);
    std::thread t2(write_value);
    std::thread t3(read_value);
    std::thread t4(read_value);
    std::thread t5(write_value);
    std::thread t6(read_value);

    t1.join();
    t2.join();
    t3.join();
    t4.join();
    t5.join();
    t6.join();

    return 0;
}

std::shared_mutex 允許以兩種方式鎖定：共享鎖 (shared lock) 和獨佔鎖 (exclusive lock)。共享鎖允許多個執行緒同時讀取資源，而獨佔鎖則只允許一個執行緒寫入資源。

Future

std::future 是在 C++11 中引入的同步工具，主要用於異步操作 (asynchronous operation) 和多執行緒之間的通信，可以讓一個執行緒等待另一個執行緒的結果。

std::async

#include <chrono>
#include <future>
#include <iostream>

int longComputation(int x) {
    std::this_thread::sleep_for(std::chrono::seconds(2));
    return x * x;
}

int main() {
    std::future<int> result = std::async(std::launch::async, longComputation, 5);

    // do something else
    std::cout << "Performing other work...\n";

    int value = result.get();
    std::cout << "Result of computation: " << value << '\n';

    return 0;
}

如上方所示，我們使用 std::async 來啟動一個異步操作，並使用 std::future 來等待結果。

std::promise

如果想要傳遞資訊給其他執行緒，可以使用 std::promise 和 std::future 來實現。

#include <future>
#include <iostream>
#include <thread>

void setPromiseValue(std::promise<int> &&promise, int value) {
    std::this_thread::sleep_for(std::chrono::seconds(2));
    promise.set_value(value * 2);
}

int main() {
    std::promise<int> promise;
    std::future<int> result = promise.get_future();

    std::thread t(setPromiseValue, std::move(promise), 10);

    std::cout << "Waiting for the result...\n";
    int value = result.get();
    std::cout << "Result from promise: " << value << '\n';

    t.join();
    return 0;
}

std::packaged_task

也可以使用 std::packaged_task 來將函數包裝成一個 std::future。

#include <chrono>
#include <future>
#include <iostream>
#include <thread>

int square(int x) {
    std::this_thread::sleep_for(std::chrono::seconds(2));
    return x * x;
}

int main() {
    std::packaged_task<int(int)> task(square);

    std::future<int> result = task.get_future();

    std::thread t(std::move(task), 5);

    std::cout << "Calculating...\n";
    std::cout << "Result of square computation: " << result.get() << '\n';

    t.join();
    return 0;
}

std::shared_future

如果要在多個執行緒之間共享資料，可以使用 std::shared_future。

#include <future>
#include <iostream>
#include <thread>

int longComputation(int x) {
    std::this_thread::sleep_for(std::chrono::seconds(2));
    return x * x;
}

int main() {
    std::future<int> fut = std::async(std::launch::async, longComputation, 5);
    std::shared_future<int> shared_result = fut.share();

    auto print_result = [](std::shared_future<int> shared_result) {
        std::cout << "Computed result: " << shared_result.get() << '\n';
    };

    std::thread t1(print_result, shared_result);
    std::thread t2(print_result, shared_result);

    t1.join();
    t2.join();

    return 0;
}

Thread​

Mutex​

std::lock_guard​

std::unique_lock​

std::scoped_lock​

Condition Variable​

Shared-Mutex & Shared-Lock​

Future​

std::async​

std::promise​

std::packaged_task​

std::shared_future​