为什么 std::mutex 比 std::atomic 快?

Why is std::mutex faster than std::atomic?

我想在多线程模式下将对象放入 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;
}

我不认为你的问题可以参考 只有 标准来回答 - 互斥锁尽可能地依赖于平台。但是,有一件事,应该提到。

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

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

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

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, that works in per-process basis and Mutex,它没有这样的限制。

锁定互斥体(或临界区)的开销要大得多,但OS有能力真正将其他等待线程放到"sleep",这提高了性能并有助于任务调度程序进行有效的资源管理.

我为什么要写这个?因为现代互斥量通常是所谓的 "hybrid mutexes"。当这种互斥锁被锁定时,它的行为就像一个普通的自旋锁 - 其他等待线程执行一些 "spins" 然后重互斥锁被锁定以防止浪费资源。

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

second_result[counter] = omp_get_thread_num();

看起来速度很快,所以 "real" 互斥量可能永远不会被锁定。这意味着,在这种情况下,您的 "mutex" 可以与基于原子的解决方案一样快(因为它本身成为基于原子的解决方案)。

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


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

另外,参考下面关于"not yielding"的评论:这是具体情况和要求的问题。自旋锁是多线程编程中非常重要的部分,通常可以通过稍微修改其行为来提高其性能。例如,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 次)。如果 mutex 仍然被锁定,pause instruction is used(如果可用)或 Sleep(0) 被调用,这基本上会导致上下文切换并允许调度程序给另一个被阻塞的线程一个机会来做一些有用的事情。然后,调用 Sleep(1) 执行实际(短)睡眠。很不错!

另外,这个声明:

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)); } 可能没有看起来那么好。

还有一个与您的问题相关的重要问题。高效的自旋锁永远不会“自旋”涉及(甚至可能)修改内存位置(例如 exchangetest_and_set)的操作。在典型的现代架构上,这些操作生成的指令要求具有锁定内存位置的缓存行处于独占状态,这是非常耗时的(尤其是当多个线程同时旋转时)。始终仅在 load/read 上自旋,并且仅当此操作有可能成功时才尝试获取锁。

例如,这里有一篇很好的相关文章:Correctly implementing a spinlock in C++