大家好，又见面了，我是你们的朋友全栈君。

[译] https://beta.boost.org/doc/libs/1_72_0/doc/html/lockfree.html

Boost.Lockfree

1. 介绍和动机

1.1 简介和术语

  术语非阻塞表示并发数据结构，这种数据结构不使用传统的同步原语（如守卫）来确保线程安全。Maurice Herlihy 和Nir Shavit（比较“多处理器编程的艺术”）区分了3中类型的非阻塞数据结构，每一种都有不同的属性：

• 如果每个并发操作都保证在有限数量的步骤中完成，则数据结构是无等待的。因此，可以为操作次数提供最坏情况的保证。
• 数据结构是无锁的，如果一些并发操作保证在有限的步骤中完成。虽然理论上某些操作可能永远不会取得任何进展但在实际应用中却不太可能发生。
• 数据结构是无阻塞的，如果并发操作保证在有限数量的步骤中完成，除非另一个并发操作干扰。

 一些数据结构只能以无锁方式实现，如果它们在某些限制下使用的话。实现的相关方面boost.lockfree是生产者和消费者线程的数量。单生产者（sp）或多生产者（mp）意味着只允许单个线程或多个并发线程向数据结构添加数据。单消费者（sr）或多消费者（mc）表示从数据结构中删除数据的等价物。

1.2 非阻塞数据结构的属性

  非阻塞数据结构不依赖锁和互斥锁来确保线程安全。同步完全在用户空间中完成，无需与操作系统进行任何直接交互。这意味着它们不容易出现优先级反转（低优先级线程需要等待高优先级线程）之类的安全问题。
非阻塞数据结构不依赖守卫，而是需要原子操作（不中断地执行特定cpu指令）。这意味着任何线程都可以看到操作之前或之后的状态，但无法观察到中间状态。并非所有硬件都支持同一组原子指令。如果它在硬件中不可用。则可以使用警卫在软件中对其进行仿真。然而，这具有失去无锁属性的明显缺陷。

1.3 非阻塞数据结构的性能

  在讨论非阻塞数据结构的性能时，必须区分摊销成本和最坏情况成本。‘lock-free’和‘wait-free’的定义只能提到了操作的上限。因此，无锁数据结构不一定是每个用例的最佳选择。为了最大化应用程序的吞吐量，应该考虑高性能并发数据结构。
  为了优化系统的延迟和避免优先级反转，无锁数据结构将是最好的选择，这在实时应用程序中可能是必需的。一般来说，我们建议考虑是否需要无锁数据结构或并发数据结构是否足够。无论如何，我们建议针对工作负载使用不同的数据结构执行基准测试。

1.4 阻塞行为的来源

  除了锁和互斥锁(boost.lockfree无论如何我们都不会使用)，还有其他三个方面可能违法锁自由：
原子操作
  一些架构在硬件中没有提供必要的原子操作。如果不是这种情况，则使用自旋锁在软件中模拟它们，二自旋锁本身就是阻塞的。
内存分配
  从操作系统分配内存不是无锁的。这使得真正的动态大小的非阻塞数据结构成为不可能。基于节点的数据boost.lockfree使用内存池来分配内部节点。如果这个内存池用完了，新节点的内存必须从操作系统中分配。然而，所有的数据结构boost.lockfree都可以配置为避免内存分配(而不是特定的调用将失败)。这对于需要无锁内存分配的实时系统特别有用。
异常处理
  c++异常处理不保证其实时行为。因此，我们不鼓励在无锁代码中使用异常和异常处理。

2.数据结构

2.1 结构

  boost.lockfree实现了三个无锁数据结构:

boost::lockfree::queue 
//一个无锁的多生产者/多消费者队列

boost::lockfree::stack
//无锁的多生产者/多消费者堆栈

boost::lockfree::spsc_queue
//一个无等待的单生产者/单消费者队列(通常被称为ringbuffer)

2.2数据结构配置

  可以使用Boost.Parameter样式模板配置数据结构:

boost::lockfree:fixed_sized
/*将数据结构配置为固定大小。内部节点存储在数组中，并通过数组索引来寻址。 这将队列的可能大小限制为索引类型可以寻址的元素数量（通常为2^16-2），但在缺少 双宽度比较和交换指令的平台上，这是最好的实现无锁的方法。 */
boost::lockfree::capacity
//在编译时设置数据结构的容量。这意味着数据结构是固定大小的。

boost::lockfree::allocator
//定义分配器。boost.lockfree支持有状态分配器并与Boost.Interprocess分配器兼容。

3.例子

3.1 Queue

  该类boost::lockfree:queue实现了一个多写/多读队列。以下示例显示了4个线程如何生成和使用整数值:

#include <boost/thread/thread.hpp>
#include <boost/lockfree/queue.hpp>
#include <iostream>
#include <boost/atomic.hpp>
boost::atomic_int producer_count(0);
boost::atomic_int consumer_count(0);
boost::lockfree::queue<int> queue(128);
const int iterations = 10000000;
const int producer_thread_count = 4;
const int consumer_thread_count = 4;
void producer(void)
{ 

for (int i = 0; i != iterations; ++i) { 

int value = ++producer_count;
while (!queue.push(value))
;
}
}
boost::atomic<bool> done (false);
void consumer(void)
{ 

int value;
while (!done) { 

while (queue.pop(value))
++consumer_count;
}
while (queue.pop(value))
++consumer_count;
}
int main(int argc, char* argv[])
{ 

using namespace std;
cout << "boost::lockfree::queue is ";
if (!queue.is_lock_free())
cout << "not ";
cout << "lockfree" << endl;
boost::thread_group producer_threads, consumer_threads;
for (int i = 0; i != producer_thread_count; ++i)
producer_threads.create_thread(producer);
for (int i = 0; i != consumer_thread_count; ++i)
consumer_threads.create_thread(consumer);
producer_threads.join_all();
done = true;
consumer_threads.join_all();
cout << "produced " << producer_count << " objects." << endl;
cout << "consumed " << consumer_count << " objects." << endl;
}

程序的输出为:

produced 40000000 objects.
consumed 40000000 objects.

4.2 stack

  该类boost::lockfree::stack实现了一个多写入器/多多读取器堆栈。以下示例显示了4个线程如何生成和使用整数值:

#include <boost/thread/thread.hpp>
#include <boost/lockfree/stack.hpp>
#include <iostream>
#include <boost/atomic.hpp>
boost::atomic_int producer_count(0);
boost::atomic_int consumer_count(0);
boost::lockfree::stack<int> stack(128);
const int iterations = 1000000;
const int producer_thread_count = 4;
const int consumer_thread_count = 4;
void producer(void)
{ 

for (int i = 0; i != iterations; ++i) { 

int value = ++producer_count;
while (!stack.push(value))
;
}
}
boost::atomic<bool> done (false);
void consumer(void)
{ 

int value;
while (!done) { 

while (stack.pop(value))
++consumer_count;
}
while (stack.pop(value))
++consumer_count;
}
int main(int argc, char* argv[])
{ 

using namespace std;
cout << "boost::lockfree::stack is ";
if (!stack.is_lock_free())
cout << "not ";
cout << "lockfree" << endl;
boost::thread_group producer_threads, consumer_threads;
for (int i = 0; i != producer_thread_count; ++i)
producer_threads.create_thread(producer);
for (int i = 0; i != consumer_thread_count; ++i)
consumer_threads.create_thread(consumer);
producer_threads.join_all();
done = true;
consumer_threads.join_all();
cout << "produced " << producer_count << " objects." << endl;
cout << "consumed " << consumer_count << " objects." << endl;
}

程序输出为：

produced 4000000 objects.
consumed 4000000 objects.

4.3 waitfree 单生产者/单消费者队列

  该类boost::lockfree::spsc_queue 实现了一个无等待的单生产者/单消费者队列。以下示例显示了 2 个单独的线程如何生成和使用整数值：

#include <boost/thread/thread.hpp>
#include <boost/lockfree/spsc_queue.hpp>
#include <iostream>
#include <boost/atomic.hpp>
int producer_count = 0;
boost::atomic_int consumer_count (0);
boost::lockfree::spsc_queue<int, boost::lockfree::capacity<1024> > spsc_queue;
const int iterations = 10000000;
void producer(void)
{ 

for (int i = 0; i != iterations; ++i) { 

int value = ++producer_count;
while (!spsc_queue.push(value))
;
}
}
boost::atomic<bool> done (false);
void consumer(void)
{ 

int value;
while (!done) { 

while (spsc_queue.pop(value))
++consumer_count;
}
while (spsc_queue.pop(value))
++consumer_count;
}
int main(int argc, char* argv[])
{ 

using namespace std;
cout << "boost::lockfree::queue is ";
if (!spsc_queue.is_lock_free())
cout << "not ";
cout << "lockfree" << endl;
boost::thread producer_thread(producer);
boost::thread consumer_thread(consumer);
producer_thread.join();
done = true;
consumer_thread.join();
cout << "produced " << producer_count << " objects." << endl;
cout << "consumed " << consumer_count << " objects." << endl;
}

程序的输出为：

produced 10000000 objects.
consumed 10000000 objects.