使用Boost.Lockfree队列比使用互斥锁慢

Question

直到现在我std::queue在我的项目中使用.我测量了此队列上特定操作所需的平均时间.

在两台机器上测量时间:我的本地Ubuntu VM和远程服务器.使用时std::queue,两台机器的平均值几乎相同:约750微秒.

然后我"升级" std::queue到boost::lockfree::spsc_queue,所以我可以摆脱保护队列的互斥锁.在我的本地虚拟机上,我可以看到巨大的性能提升,平均现在是200微秒.然而,在远程机器上,平均值高达800微秒,这比以前慢了.

首先我认为这可能是因为远程机器可能不支持无锁实现:

从Boost.Lockfree页面:

并非所有硬件都支持同一组原子指令.如果硬件不可用,则可以使用防护装置在软件中进行仿真.然而,这具有失去无锁属性的明显缺点.

要确定是否支持这些指令,请boost::lockfree::queue调用一个方法bool is_lock_free(void) const;.但是,boost::lockfree::spsc_queue没有这样的功能,对我来说,这意味着它不依赖于硬件而且总是无锁 - 在任何机器上.

性能损失的原因是什么？

---

例子(生产者/消费者)

// c++11 compiler and boost library required

#include <iostream>
#include <cstdlib>
#include <chrono>
#include <async>
#include <thread>
/* Using blocking queue:
 * #include <mutex>
 * #include <queue>
 */
#include <boost/lockfree/spsc_queue.hpp>


boost::lockfree::spsc_queue<int, boost::lockfree::capacity<1024>> queue;

/* Using blocking queue:
 * std::queue<int> queue;
 * std::mutex mutex;
 */

int main()
{
    auto producer = std::async(std::launch::async, [queue /*,mutex*/]() 
    {
        // Producing data in a random interval
        while(true)
        {
            /* Using the blocking queue, the mutex must be locked here.
             * mutex.lock();
             */

            // Push random int (0-9999)
            queue.push(std::rand() % 10000);

            /* Using the blocking queue, the mutex must be unlocked here.
             * mutex.unlock();
             */

            // Sleep for random duration (0-999 microseconds)
            std::this_thread::sleep_for(std::chrono::microseconds(rand() % 1000));
        }
    }

    auto consumer = std::async(std::launch::async, [queue /*,mutex*/]() 
    {
        // Example operation on the queue.
        // Checks if 1234 was generated by the producer, returns if found.

        while(true)
        {
            /* Using the blocking queue, the mutex must be locked here.
             * mutex.lock();
             */

            int value;
            while(queue.pop(value)
            {
                if(value == 1234)
                    return;
            }

            /* Using the blocking queue, the mutex must be unlocked here.
             * mutex.unlock();
             */

            // Sleep for 100 microseconds
            std::this_thread::sleep_for(std::chrono::microseconds(100));
        }
    }

    consumer.get();
    std::cout << "1234 was generated!" << std::endl;
    return 0;
}

Answer 1

无锁算法通常比基于锁的算法执行得更差.这是他们几乎不经常使用的关键原因.

无锁算法的问题在于它们通过允许竞争线程继续竞争来最大化争用.锁通过取消调度竞争线程来避免争用.只有在无法取消调度竞争线程时,才应使用无锁算法,这是第一次近似.这很少适用于应用程序级代码.

让我给你一个非常极端的假设.想象一下,在一个典型的现代双核CPU上运行四个线程.线程A1和A2正在操作集合A.线程B1和B2正在操作集合B.

首先,让我们假设该集合使用锁.这意味着如果线程A1和A2(或B1和B2)尝试同时运行,其中一个将被锁定阻止.因此,很快,一个A线程和一个B线程将运行.这些线程将非常快速地运行并且不会竞争.任何时候线程都试图争用,冲突的线程将被取消调度.好极了.

现在,想象一下该集合不使用锁.现在,线程A1和A2可以同时运行.这将导致持续争用.集合的缓存行将在两个核心之间进行乒乓.核心间总线可能会饱和.表现太糟糕了.

再次,这是非常夸张的.但是你明白了.你想避免争用,而不是尽可能多地遭受损失.

但是,现在再次运行这个思想实验,其中A1和A2是整个系统中唯一的线程.现在,无锁集合可能更好(尽管你可能会发现在这种情况下只有一个线程更好!).

几乎每个程序员都经历了一个阶段,他们认为锁是坏的并且避免锁会使代码变得更快.最终,他们意识到这种争用会使事情变得缓慢并且锁定,正确使用,最大限度地减少争用.