C++并发实战19：lock free编程

大家好，又见面了，我是你们的朋友全栈君。     涉及到并行/并发计算时，通常都会想到加锁，加锁可以保护共享的数据，不过也会存在一些问题：

1. 由于临界区无法并发运行，进入临界区就需要等待，加锁使得效率的降低。多核CPU也不能发挥全部马力

2. 在复杂的情况下，很容易造成死锁，并发进程、线程之间无止境的互相等待。

3. 在中断/信号处理函数中不能加锁，给并发处理带来困难。

4. 加锁影响实时性，等待时间不确定

5. 优先级反转，优先级高的等待优先级低的

6. 若一个线程带着锁挂了，那么将会影响其它等待该锁的线程

总之，在基于锁的多线程/多进程编程，你需要保证对竞争条件很敏感的共享数据上的任何操作，都通过加锁或解锁一个独占(mutex)来实现原子操作。

现有的加锁/无锁编程的种类如下：

      其中标注为红色字体的方案为 Blocking synchronization需要锁，黑色字体为 Non-blocking synchronization无锁。Lock-based 和 Lockless-based 两者之间的区别仅仅是加锁粒度的不同。图中最底层的方案就是大家经常使用的 mutex 和 semaphore 等方案，代码复杂度低，但运行效率也最低。

      可以使用std::atomic实现lock free，但这里并不是真正的无锁，只有atomic_flag是无锁的，其它的atomic内部都是有锁的只不过粒度很小.atomic::compare_exchange_weak/strong等于是个CAS(比较并交换)操作，在C++11之前该操作是平台相关的，现在atomic将其实现为成员函数。

      一个lock free的栈：

#include <atomic>
#include <memory>

template<typename T>
class lock_free_stack//栈的底层数据结构采用单向链表实现
{
private:
    struct node
    {
        std::shared_ptr<T> data;//这里采用shared_ptr管理的好处在于：若栈内存放对象pop中return栈顶对象可能拷贝异常，栈内只存储指针还可以提高性能
        node* next;
        node(T const& data_):
            data(std::make_shared<T>(data_))//注意make_shared比直接shared_ptr构造的内存开销小
        {}
    };
    std::atomic<node*> head;//采用原子类型管理栈顶元素，主要利用atomic::compare_exchange_weak实现lock free
public:
    void push(T const& data)
    {
        node* const new_node=new node(data);
        new_node->next=head.load();//每次从链表头插入
        while(!head.compare_exchange_weak(new_node->next,new_node));//若head==new_node->next则更新head为new_node,返回true结束循环，插入成功; 若head!=new_node->next表明有其它线程在此期间对head操作了，将new_node->next更新为新的head，返回false，继续进入下一次while循环。这里采用atomic::compare_exchange_weak比atomic::compare_exchange_strong快，因为compare_exchange_weak可能在元素相等的时候返回false所以适合在循环中，而atomic::compare_exchange_strong保证了比较的正确性，不适合用于循环
    }
    std::shared_ptr<T> pop()
    {
        node* old_head=head.load();//拿住栈顶元素，但是可能后续被更新，更新发生在head!=old_head时
        while(old_head &&!head.compare_exchange_weak(old_head,old_head->next));//这里注意首先要先判断old_head是否为nullptr防止操作空链表，然后按照compare_exchange_weak语义更新链表头结点。若head==old_head则更新head为old_head->next并返回true，结束循环，删除栈顶元素成功;若head!=old_head表明在此期间有其它线程操作了head，因此更新old_head为新的head,返回false进入下一轮循环，直至删除成功。
        return old_head ? old_head->data : std::shared_ptr<T>();//这里注意空链表时返回的是一个空的shared_ptr对象
    }//这里只是lock free，由于while循环可能无限期循环不能在有限步骤内完成，故不是wait free
};