2

我正在考虑为项目实施队列的选项，其要求是生产者至少必须尽可能低延迟。为此，我一直在研究“无锁”队列，std::atomic用于控制生产者和消费者线程对数据结构的访问。我希望这将避免代码当前使用的开销std::mutex，特别是避免开销。std::unique_lock

为此，我编写了一个简单的测试程序来评估std::mutex（加上std::unique_lock）和std::atomic. 该程序还进行检查以确保原子对象是无锁的，它确实是。

#include <mutex>
#include <atomic>
#include <thread>
#include <chrono>
#include <iostream>

#define CYCLES 100000000

void testAtomic()
{
   bool var(true);
   std::atomic_bool _value(true);

   std::cout << "atomic bool is ";

   if(!_value.is_lock_free())
      std::cout << "not ";
   std::cout << "lock free" << std::endl;
   const auto _start_time = std::chrono::high_resolution_clock::now();

   for(size_t counter = 0; counter < CYCLES; counter++)
   {
      var = _value.load();
      var = !var;
      _value.store(var);
   }

   const auto _end_time = std::chrono::high_resolution_clock::now();

   std::cout << 1e-6 * std::chrono::duration_cast<std::chrono::microseconds>
      (_end_time - _start_time).count() << " s" << std::endl;
}

void testMutex()
{
   bool var(true);
   std::mutex _mutex;
   std::chrono::high_resolution_clock _clock;

   const auto _start_time = std::chrono::high_resolution_clock::now();

   for(size_t counter = 0; counter < CYCLES; counter++)
   {
      std::unique_lock<std::mutex> lock(_mutex);
      var = !var;
   }

   const auto _end_time = std::chrono::high_resolution_clock::now();

   std::cout << 1e-6 * std::chrono::duration_cast<std::chrono::microseconds>
      (_end_time - _start_time).count() << " s" << std::endl;
}

int main()
{
   std::thread t1(testAtomic);
   t1.join();
   std::thread t2(testMutex);
   t2.join();
   return 0;
}

运行此程序时，我得到以下输出：

atomic bool is lock free
3.49434 s 
2.31755 s

这将向我表明std::mutex(and std::unique_lock) 明显更快，这与我阅读有关原子与互斥锁的期望相反。我的发现正确吗？我的测试程序有问题吗？我对两者之间差异的理解不正确吗？

代码在 CentOS7 上使用 GCC 4.8.5 编译

4

2 回答 2

2

你在什么硬件上测试？

由于您使用的是 GCC，std::atomicseq_cst 存储将使用mov+ 慢mfence而不是稍慢一些的xchg-with-mem （这也是一个完整的障碍，就像所有其他 x86 原子 RMW 操作一样）。

采用互斥体需要原子 RMW（例如xchg，而不是 mov + mfence）。如果幸运的话，互斥锁可能只是一个普通的存储（如mo_release）。竞争为零，因此获取锁总是成功的。

显然，这些互斥锁/解锁库函数背后的代码比 更便宜mfence，尤其是在带有更新微码的 Skylake CPU 上，这mfence是乱序执行和内存的完全障碍。（请参阅此答案的底部，以及lock xchg 是否具有与 mfence 相同的行为？）

另外，请注意，您的互斥循环将本地优化bool var为寄存器，实际上并没有在循环内的内存中更新它。（您在 Godbolt 编译器资源管理器中使用 gcc4.8.5 的代码）。

# the main loop from testMutex
.L80:                                                       # do {
        mov     rdi, rsp                                      # pointer to _mutex on the stack
        call    __gthrw_pthread_mutex_lock(pthread_mutex_t*)
        test    eax, eax
        jne     .L91                                          # mutex error handling
        mov     rdi, rsp                                      # pointer to _mutex again
        call    __gthrw_pthread_mutex_unlock(pthread_mutex_t*)
        sub     rbx, 1
        jne     .L80                                        # }while(--counter)

xor bl, 1循环内部将是无关紧要的；乱序执行可能与其他工作重叠。

如果对函数的引用var转义了函数，因此编译器必须在非内联函数调用（包括对 pthread 库函数）之前使其在内存中同步，我们会期望类似xor byte ptr [rsp+8], 1. 这也将非常便宜，并且可能大部分被乱序执行隐藏，尽管加载/ALU/存储可能是在耗尽存储缓冲区时必须等待完整屏障的东西。

加速你的std::atomic代码：

您似乎有意避免执行原子 RMW，而是加载到 tmp var 并执行单独的存储。如果您只使用 release 而不是 seq_cst，那么它可以编译为 x86 上的普通存储指令。（或者在大多数其他 ISA 上设置更便宜的壁垒）。

      bool tmp = _value.load(std::memory_order_relaxed);   // or acquire
      _value.store(!tmp, std::memory_order_release);

这应该以每个反转大约 6 个周期运行，只是一个 ALU 操作的延迟加上存储/重新加载的存储转发延迟。mfence与（https://uops.info/）的最佳吞吐量相比，每次迭代可能需要 33 个周期。

或者由于这是一个非原子修改，只需存储交替值而不重新读取旧值。通常只能在只有一个生产者写入值而其他线程正在读取的情况下才能避免原子 RMW。因此，让生产者将其正在修改的值保存在寄存器（非原子本地变量）中，如果存在则存储副本。

   bool var = true;
   for(size_t counter = 0; counter < CYCLES; counter++)
   {
      var = !var;
      _value.store(var, std::memory_order_release);
   }

此外，不要在您自己的 var 名称中使用前导下划线。这些名称是为实现而保留的。（_带小写的单个仅在文件/全局范围内保留，但这仍然是不好的做法。）

于 2020-06-03T00:40:33.560 回答
0

有人告诉我，最初在互斥锁的某些实现中，它会首先在内部自旋锁。

这可能就是它看起来更快的原因。

如果进行系统调用，我怀疑你会得到相同的结果。

（我无法验证这一点，但我认为这可能是一个原因）

于 2020-06-02T17:17:56.467 回答