c++ - 为什么 std::mutex 比 std::atomic 快？

Question

我想在多线程模式下将对象放入 std::vector 中。所以我决定比较两种方法:一种使用 std::atomic，另一种使用 std::mutex。我看到第二种方法比第一种方法更快。为什么？

我使用 GCC 4.8.1，在我的机器(8 线程)上，我看到第一个解决方案需要 391502 微秒，第二个解决方案需要 175689 微秒。

#include <vector>
#include <omp.h>
#include <atomic>
#include <mutex>
#include <iostream>
#include <chrono>

int main(int argc, char* argv[]) {
    const size_t size = 1000000;
    std::vector<int> first_result(size);
    std::vector<int> second_result(size);
    std::atomic<bool> sync(false);

    {
        auto start_time = std::chrono::high_resolution_clock::now();
        #pragma omp parallel for schedule(static, 1)
        for (int counter = 0; counter < size; counter++) {
            while(sync.exchange(true)) {
                std::this_thread::yield();
            };
            first_result[counter] = counter;
            sync.store(false) ;
        }
        auto end_time = std::chrono::high_resolution_clock::now();
        std::cout << std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time).count() << std::endl;
    }

    {
        auto start_time = std::chrono::high_resolution_clock::now();
        std::mutex mutex; 
        #pragma omp parallel for schedule(static, 1)
        for (int counter = 0; counter < size; counter++) {
            std::unique_lock<std::mutex> lock(mutex);       
            second_result[counter] = counter;
        }
        auto end_time = std::chrono::high_resolution_clock::now();
        std::cout << std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time).count() << std::endl;
    }

    return 0;
}

Answer 1

我不认为您的问题可以仅引用标准来回答——互斥锁尽可能地依赖于平台。但是，有一件事应该提到。

互斥体并不慢。您可能看过一些文章，将它们的性能与自定义自旋锁和其他“轻量级”东西进行比较，但这不是正确的方法 - 它们不可互换。

自旋锁 相当快，当它们被锁定(获取)的时间相对较短时 - 获取它们非常便宜，但其他也在尝试锁定的线程处于事件状态整个时间(不断循环运行)。

自定义自旋锁可以这样实现:

class SpinLock
{
private:
    std::atomic_flag _lockFlag;

public:
    SpinLock()
    : _lockFlag {ATOMIC_FLAG_INIT}
    { }

    void lock()
    {
        while(_lockFlag.test_and_set(std::memory_order_acquire))
        { }
    }

    bool try_lock()
    {
        return !_lockFlag.test_and_set(std::memory_order_acquire);
    }

    void unlock()
    {
        _lockFlag.clear();
    }
};

Mutex 是一个原语，它要复杂得多。特别是，在 Windows 上，我们有两个这样的原语 - Critical Section ，在每个进程的基础上工作，并且 Mutex , 而没有这样的限制。

锁定互斥量(或关键部分)的成本要高得多，但操作系统有能力真正让其他等待线程进入“ sleep ”状态，从而提高性能并帮助任务调度程序进行有效的资源管理。

为什么要写这个？因为现代互斥锁通常是所谓的“混合互斥锁”。当此类互斥锁被锁定时，它的行为就像一个普通的自旋锁 - 其他等待线程执行一定数量的“自旋”，然后重互斥锁被锁定以防止浪费资源。

在您的例子中，互斥锁在每次循环迭代中被锁定以执行此指令:

second_result[counter] = omp_get_thread_num();

它看起来像一个快速的，所以“真正的”互斥量可能永远不会被锁定。这意味着，在这种情况下，您的“互斥量”可以与基于原子的解决方案一样快(因为它本身成为基于原子的解决方案)。

此外，在第一个解决方案中，您使用了某种类似自旋锁的行为，但我不确定这种行为在多线程环境中是否可预测。我很确定，“锁定”应该有 acquire语义，而解锁是一个 release操作。 Relaxed对于此用例，内存排序可能太弱。

---

我将代码编辑得更紧凑和正确。它使用 std::atomic_flag ，这是唯一的类型(不像 std::atomic<> 特化)，保证是无锁的(即使是 std::atomic<bool> 也不给你)。

另外，引用下面关于“不屈服”的评论:这是具体情况和要求的问题。自旋锁是多线程编程中非常重要的部分，通常可以通过稍微修改其行为来提高其性能。例如，Boost 库实现了 spinlock::lock()如下:

void lock()
{
    for( unsigned k = 0; !try_lock(); ++k )
    {
        boost::detail::yield( k );
    }
}

来源:boost/smart_ptr/detail/spinlock_std_atomic.hpp

在哪里detail::yield()是(Win32 版本):

inline void yield( unsigned k )
{
    if( k < 4 )
    {
    }
#if defined( BOOST_SMT_PAUSE )
    else if( k < 16 )
    {
        BOOST_SMT_PAUSE
    }
#endif
#if !BOOST_PLAT_WINDOWS_RUNTIME
    else if( k < 32 )
    {
        Sleep( 0 );
    }
    else
    {
        Sleep( 1 );
    }
#else
    else
    {
        // Sleep isn't supported on the Windows Runtime.
        std::this_thread::yield();
    }
#endif
}

[来源:http://www.boost.org/doc/libs/1_66_0/boost/smart_ptr/detail/yield_k.hpp]

首先，线程旋转一些固定次数(在本例中为 4 次)。如果互斥体仍然被锁定， pause instruction is used (如果可用)或 Sleep(0)被调用，这基本上会导致上下文切换，并允许调度程序给另一个阻塞的线程一个机会来做一些有用的事情。然后，Sleep(1)被调用以执行实际(短) sleep 。非常好!

另外，这个声明:

The purpose of a spinlock is busy waiting

不完全正确。自旋锁的目的是作为一种快速、易于实现的锁原语——但它仍然需要正确编写，并考虑到某些可能的场景。例如，Intel says (关于 Boost 对 _mm_pause() 的使用，作为在 lock() 内部产生的方法):

In the spin-wait loop, the pause intrinsic improves the speed at which the code detects the release of the lock and provides especially significant performance gain.

所以，像这样的实现 void lock() { while(m_flag.test_and_set(std::memory_order_acquire)); }可能不像看起来那么好。