带有原子等待的 C++20 互斥锁

Question

从 C++20 std::atomics 开始，有等待和通知操作。使用 is_always_lock_free 我们可以确保实现是无锁的。使用这些积木构建无锁互斥锁并不那么困难。在简单的情况下，锁定将是比较交换操作，或者如果互斥体被锁定则等待。这里最大的问题是这是否值得。如果我可以创建这样的实现，那么 STL 版本很可能会更好、更快。然而，我仍然记得当我在 2016 年看到 QMutex 如何优于 std::mutex QMutex 与 std::mutex时，我是多么惊讶。那么您认为我应该尝试这样的实现还是 std::mutex 的当前实现已经足够成熟，可以进行远远超出这些技巧的优化？

更新 我的措辞不是最好的，我的意思是实现可以在快乐的路径上无锁（从未锁定状态锁定）。当然，如果我们需要等待获取锁，我们应该被阻塞并重新调度。在大多数平台上，atomic::wait 很可能不是通过简单的自旋锁实现的（现在让我们忽略极端情况），所以基本上它实现了 mutex::lock 所做的相同的事情。所以基本上，如果我实现这样一个类，它将执行与 std::mutex 完全相同的操作（同样在大多数流行平台上）。这意味着 STL 可以在支持这些技巧的平台上的互斥实现中使用相同的技巧。就像这个 spinlock，但我会使用原子等待而不是旋转。我应该相信我的 STL 实现是他们这样做的吗？

Answer 1

无锁互斥体是一个矛盾。

您可以用无锁构建块构建一个锁，事实上，无论是用 asm 手写还是用std::atomic.

但根据定义，整体锁定算法并不是无锁的。（https://en.wikipedia.org/wiki/Non-blocking_algorithm）。整个要点是当一个线程位于临界区时阻止其他线程向前推进，即使它不幸地在该区域休眠。

我的意思是，实现可以在快乐的路径上无锁（从未锁定状态锁定）

std::mutex::lock()也是这样：如果不需要，它不会阻塞！ 它可能需要进行系统调用，例如futex(FUTEX_WAIT_PRIVATE)是否有线程正在等待锁。但使用的实现也是如此std::notify。

也许您还没有理解“无锁”的含义：无论其他线程在做什么，它都不会阻塞。就这些。它并不意味着“在简单/容易的情况下更快”。对于复杂的算法（例如队列），如果循环缓冲区已满，则放弃并阻塞通常会更快，而不是在简单情况下增加开销以允许其他线程“协助”或取消卡住的部分操作。（循环缓冲队列中的无锁进度保证）

推出自己的产品并没有固有的优势。std::mutexstd::atomic 无论哪种方式，编译器生成的机器代码都必须执行大致相同的操作，并且我希望快速路径大致相同。唯一的区别在于选择在已经锁定的情况下做什么。（但设计一种方法来避免调用notifyiff 可能很棘手，如果没有服务员，这是std::mutexglibc / pthreads 实现 Linux 中实际管理的东西。）

（我假设与原子 RMW 获取互斥体的成本相比，库函数调用的开销可以忽略不计。将其内联到代码中是一个相当小的优势。）

互斥体实现可以针对某些用例进行调整，包括睡眠前自旋等待的时间（使用操作系统辅助机制，例如futex在释放锁时使其他线程能够唤醒它），以及指数退避自旋等待部分。

如果std::mutex您的应用程序在您关心的硬件上表现不佳，则值得考虑替代方案。尽管我不知道如何衡量它是否有效。也许如果你能弄清楚它正在决定睡觉

是的，您可以考虑自己推出，因为std::atomic现在有一个可移植的机制，希望能够提供一种回退到操作系统辅助的睡眠/唤醒机制（例如futex. 不过，您仍然希望_mm_pause()在自旋等待循环中手动使用系统特定的东西，例如 x86，因为我认为 C++ 没有任何与Rust 相同的东西std::hint::spin_loop()，实现可以用来公开 x86pause指令之类的东西，旨在用于在自旋循环体中使用。（请参阅通过内联汇编围绕内存操作锁定：此类注意事项，以及旋转只读而不是垃圾邮件原子 RMW 尝试。并查看 x86 汇编语言中自旋锁的必要部分，无论您是否获得自旋锁，这些部分都是相同的C++ 编译器是否为您生成机器代码。）

另请参阅https://rigtorp.se/spinlock/回复：使用 std::atomic 在 C++ 中实现互斥体。

Linux / libstdc++ 通知/等待行为

我在 Arch Linux (glibc 2.33) 上测试了当系统调用std::wait需要等待很长时间时会执行什么操作。

• std::mutex锁定无争用快速路径，解锁无需等待：零系统调用，纯粹的用户空间原子操作。值得注意的是，解锁时能够检测到没有服务员，因此它不会进行FUTEX_WAKE系统调用（否则这可能比获取和释放该核心 L1d 缓存中仍然很热的无竞争互斥量多一百倍。）

• std::mutex lock()on 已锁定：仅futex(0x55be9461b0c0, FUTEX_WAIT_PRIVATE, 2, NULL)系统调用。在此之前可能会在用户空间中发生一些旋转；我没有使用 GDB 单步执行它，但如果是这样，可能会使用pause指令。

• std::mutex unlock()配服务员：用futex(0x55ef4af060c0, FUTEX_WAKE_PRIVATE, 1) = 1。（在原子 RMW 之后，IIRC；不知道为什么它不只使用发布存储。）

• std::notify_one：futex(address, FUTEX_WAKE, 1)即使没有服务员，也总是a，所以当你解锁锁时，如果没有服务员，你就可以避免这种情况。

• std::wait：在用户空间中旋转几次，包括sched_yield()在futex(addr, FUTEX_WAIT, old_val, NULL).

请注意使用/函数FUTEX_WAIT而不是/函数：这些函数应该在共享内存上跨进程工作。 手册页说版本（仅适用于单个进程的线程）允许一些额外的优化。FUTEX_WAIT_PRIVATEwaitnotifyfutex(2)_PRIVATE

我不了解其他系统，尽管我听说Windows / MSVC 上的某些类型的锁必须很糟糕（即使在快速路径上也总是系统调用）才能向后兼容某些 ABI 选择或其他东西。也许std::lock_guard在 MSVC 上速度很慢，但std::unique_lock不是吗？

测试代码：

#include <atomic>
#include <thread>
#include <unistd.h>   // for quick & dirty sleep and usleep.  TODO: port to non-POSIX.
#include <mutex>

#if 1        // test std::atomic
//std::atomic_unsigned_lock_free gvar;
//std::atomic_uint_fast_wait_t gvar;
std::atomic<unsigned> gvar;

void waiter(){
    volatile unsigned sink;
    while (1) {
        sink = gvar;
        gvar.wait(sink);   // block until it's not the old value.
        // on Linux/glibc, 4x sched_yield(), then futex(0x562c3c4881c0, FUTEX_WAIT, 46, NULL )  or whatever the current counter value is
    }
}

void notifier(){
    while(1){
        sleep(1);
        gvar.store(gvar.load(std::memory_order_relaxed)+1, std::memory_order_relaxed);
        gvar.notify_one();
    }
}
#else
std::mutex mut;

void waiter(){
    unsigned sink = 0;
    while (1) {
        mut.lock();     // Linux/glibc2.33 - just a futex system call, maybe some spinning in user-space first. But no sched_yield
        mut.unlock();
        sink++;
        usleep(sink);  // give the other thread plenty of time to take the lock, so we don't get it twice.
    }
}

void notifier(){
    while(1){
        mut.lock();
        sleep(1);
        mut.unlock();
    }
}
#endif

int main(){
    std::thread t (waiter);   // comment this to test the no-contention case
    notifier();
    // loops forever: kill it with control-C
}

编译g++ -Og -std=gnu++20 notifywait.cpp -pthread，运行strace -f ./a.out以查看系统调用。（每秒几次或几次，因为我睡了一个很好的长觉。）

如果用户空间中有任何旋转等待，与 1 秒的睡眠间隔相比，它可以忽略不计，因此它使用大约一毫秒的 CPU 时间（包括启动）来运行 19 次迭代。( perf stat ./a.out)

通常，您最好将时间花在尝试减少所涉及的锁定量或争用量上，而不是尝试优化锁本身。 锁定是一件极其重要的事情，并且针对大多数用例进行了大量的工程调整。

如果您正在滚动自己的锁，您可能想亲自动手处理特定于系统的东西，因为这都是调整选择的问题。std::mutex不同的系统不太可能为waitLinux/glibc做出相同的调整选择。除非std::wait您关心的唯一系统上的重试策略恰好适合您的用例。

如果不首先准确调查系统上的行为，就滚动自己的互斥体是没有意义std::mutex的，例如，对已经锁定的情况单步执行汇编程序以查看它会进行哪些重试。然后你就会更好地了解自己是否可以做得更好。