关于c++:C线程同步

39次阅读

共计 5924 个字符，预计需要花费 15 分钟才能阅读完成。

C++ 规范库提供了如下线程同步机制：

  • 互斥量（反对超时加锁、递归加锁）
  • 读写锁（共享互斥量，也反对超时加锁）
  • 互斥量包装器（基于 RAII 的思维）
  • 条件变量
  • 信号量（二元信号量、计数信号量）
  • 栅栏（反对重用）
  • 调用一次

1. 互斥量

#include <mutex>

  • mutex：提供根底的互斥性能。

    std::mutex mtx;

mtx.lock();                      // locks the mutex, blocks if the mutex is not available
bool ok = mtx.try_lock();        // tries to lock the mutex, returns if the mutex is not available 
mtx.unlock();                    // unlocks the mutex

  • timed_mutex：在 mutex 的根底上减少了超时加锁的性能。

    #include <chrono>

using namespace std::chrono_literals;

std::timed_mutex mtx;

// 以绝对工夫的模式指定超时工夫
if (mtx.try_lock_for(100ms))
{// 已加锁}

// 以相对工夫的模式指定超时工夫
auto now = std::chrono::steady_clock::now();
if (mtx.try_lock_until(now + 10s))
{// 已加锁}

  • recursive_mutex：在 mutex 的根底上减少了递归加锁的性能（此时，lock() 函数能够被同一线程在不开释锁的状况下屡次调用）。

    std::recursive_mutex mtx;

void fun1() {mtx.lock();
  // ...
  mtx.unlock();}

void fun2() {mtx.lock();
  // ...
  fun1(); // recursive lock becomes useful here
  mtx.unlock();};
  • recursive_timed_mutex：在 timed_mutex 的根底上减少了递归加锁的性能。

2. 读写锁

#include <shared_mutex>

  • shared_mutex

    std::shared_mutex mtx;

// 写者（互斥）mtx.lock();
bool ok = mtx.try_lock();
mtx.unlock();

// 读者（共享）mtx.lock_shared();
bool ok = mtx.try_lock_shared();
mtx.unlock_shared();

  • shared_timed_mutex：在 shared_mutex 的根底上减少了超时加锁的性能。

    #include <chrono>

using namespace std::chrono_literals;

std::shared_timed_mutex mtx;

/*********************************** 写者 ***********************************/

// 以绝对工夫的模式指定超时工夫
if (mtx.try_lock_for(100ms))
{// 已加锁}

// 以相对工夫的模式指定超时工夫
auto now = std::chrono::steady_clock::now();
if (mtx.try_lock_until(now + 10s))
{// 已加锁}


/*********************************** 读者 ***********************************/

// 以绝对工夫的模式指定超时工夫
if (mtx.try_lock_shared_for(100ms))
{// 已加锁}

// 以相对工夫的模式指定超时工夫
auto now = std::chrono::steady_clock::now();
if (mtx.try_lock_shared_until(now + 10s))
{// 已加锁}

3. 互斥量包装器

  • lock_guard：应用了 RAII 的机制，结构时加锁，析构时解锁。

    #include <mutex>

std::mutex mtx;

void f()
{const std::lock_guard<std::mutex> lock(mtx);
  // ...
  // mtx is automatically released when lock goes out of scope
}

  • scoped_lock：相似于 lock_guard，但能够治理多个互斥量（能够避免死锁）。

    #include <mutex>

std::mutex mtx1, mtx2;

void f()
{const std::scoped_lock<std::mutex, std::mutex> lock(mtx1, mtx2);
  // ...
  // mtx is automatically released when lock goes out of scope
}

  • unique_lock：相似于 lock_guard，但反对提早加锁、超时加锁、递归加锁。

    #include <mutex>
#include <chrono>

using namespace std::chrono_literals;

std::timed_mutex mtx;

// 结构时加锁
std::unique_lock<std::timed_mutex> lock(mtx);

// 提早加锁：先不加锁
std::unique_lock<std::timed_mutex> lock(mtx, std::defer_lock);

lock.lock();
bool ok = lock.try_lock();
lock.unlock();

bool ok = lock.try_lock_for(100ms);

auto now = std::chrono::steady_clock::now();
bool ok = mtx.try_lock_until(now + 10s);

  • shared_lock：用于治理读写锁中的读者模式（写者模式应用 unique_lock 即可），反对提早加锁、超时加锁。

    #include <mutex>
#include <chrono>

using namespace std::chrono_literals;

std::shared_mutex mtx;

// 结构时加锁
std::shared_lock<std::shared_mutexx> lock(mtx);

// 提早加锁：先不加锁
std::shared_lock<std::shared_mutex> lock(mtx, std::defer_lock);

lock.lock();
bool ok = lock.try_lock();
lock.unlock();

bool ok = lock.try_lock_for(100ms);

auto now = std::chrono::steady_clock::now();
bool ok = mtx.try_lock_until(now + 10s);

4. 条件变量

#include <condition_variable>

  • condition_variable：期待时只能应用 std::unique_lock<std::mutex>。

    std::mutex mtx;
std::condition_variable cv;
bool ready = false;

void worker_thread()
{
  /*
  期待，直到 ready 为 true，等价于
  while (!ready)
  {cv.wait(lock);
  }
  */
  std::unique_lock<std::mutex> lock(mtx);
  cv.wait(lock, []{return ready;});
 
  // after the wait, we own the lock.
  // ...
 
  // 在唤醒之前解锁，免得被唤醒的线程仍阻塞于互斥量
  lock.unlock();
  cv.notify_one();}
    #include <chrono>

using namespace std::chrono_literals;

std::mutex mtx;
std::condition_variable cv;
int i;

std::unique_lock<std::mutex> lock(mtx);

// 超时期待：绝对工夫
if(cv.wait_for(lock, 100ms, []{return i == 1;}))
{// 条件满足 ...}

// 超时期待：相对工夫
auto now = std::chrono::system_clock::now();
if(cv.wait_until(lock, now + 100ms, [](){return i == 1;}))
{// 条件满足 ...}

// 唤醒所有期待线程
cv.notify_all();
  • condition_variable_any：与 condition_variable 相似，但能够联合其余锁应用。

5. 信号量

#include <semaphore>

  • binary_semaphore：二元信号量，其实就是计数信号量模板的特化（计数为 1）。

    std::binary_semaphore sem;

sem.acquire();        // decrements the internal counter or blocks until it can
sem.release();        // increments the internal counter and unblocks acquirers

book ok = sem.try_acquire();    // tries to decrement the internal counter without blocking

// 超时 acquire：绝对工夫
bool ok = sem.try_acquire_for(100ms);

// 超时 acquire：相对工夫
auto now = std::chrono::system_clock::now();
bool ok = sem.try_acquire_until(now + 100ms);

    注：ok 为 true 示意 acquire 胜利。

  • counting_semaphore：计数信号量，反对的操作同上。

    std::counting_semaphore sem(4);

6. 栅栏

  • latch：其外部保护着一个计数器，当计数不为 0 时，所有参与者（线程）都将阻塞在期待操作处，计数为 0 时，解除阻塞。计数器不可重置或减少，故它是一次性的，不可重用。

    #include <latch>

std::latch work_done(4);

work_done.count_down();             // decrements the counter in a non-blocking manner
work_done.wait();                   // blocks until the counter reaches zero
bool ok = work_done.try_wait();     // tests if the internal counter equals zero
work_done.arrive_and_wait();        // decrements the counter and blocks until it reaches zero

  • barrier：相似于 latch，它会阻塞线程直到所有参与者线程都达到一个同步点，但它是可重用的。
    一个 barrier 的生命周期蕴含多个阶段，每个阶段都定义了一个同步点。一个 barrier 阶段蕴含：

    • 冀望计数（设创立时指定的计数为 n），当冀望计数不为 0 时，参与者将阻塞于期待操作处；
    • 当冀望计数为 0 时，会执行创立 barrier 时指定的 阶段实现步骤，而后解除阻塞所有阻塞于同步点的参与者线程。
    • 当阶段实现步骤执行实现后，会重置冀望计数为 n - 调用 arrive_and_drop()的次数，而后开始下一个阶段。
    #include <barrier>

auto on_completion = []() noexcept { 
  // locking not needed here
  // ...
};

std::barrier sync_point(4, on_completion);

sync_point.arrive();            // arrives at barrier and decrements the expected count 
sync_point.wait();              // blocks at the phase synchronization point until its phase completion step is run
sync_point.arrive_and_wait();    // arrives at barrier and decrements the expected count by one, then blocks until current phase completes
sync_point.arrive_and_drop();    // decrements both the initial expected count for subsequent phases and the expected count for current phase by one

7. 调用一次

确保某个操作只被执行一次（胜利执行才算），即便是多线程环境下也确保只执行一次。

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

std::once_flag flag;

void may_throw_function(bool do_throw)
{if (do_throw)
    {
        std::cout << "throw: call_once will retry\n"; // this may appear more than once
        throw std::exception();}
    std::cout << "Didn't throw, call_once will not attempt again\n"; // guaranteed once
}

void do_once(bool do_throw)
{
    try
    {std::call_once(flag, may_throw_function, do_throw);
    }
    catch (...)
    {}}

int main()
{std::thread t1(do_once, true);
    std::thread t2(do_once, true);
    std::thread t3(do_once, false);
    std::thread t4(do_once, true);
    t1.join();
    t2.join();
    t3.join();
    t4.join();}
throw: call_once will retry
throw: call_once will retry
Didn't throw, call_once will not attempt again

正文完
c++
发表至： c#
2021-09-03
 0
关于c++:树与图
关于c++:DeepRoute-Lab-C-内存管理介绍
关于c++:Essentiacal-c课后题
关于c:和C死磕到底