-
lock-free的概念
什么是lock-free，简单的说就是不直接使用锁，减少锁在系统中占用的开销。相比于基于锁的算法而言，Lock-free 算法具有明显的特征：某个线程在执行数据访问时挂起不会阻碍其他的线程继续执行（某个线程持有锁之后挂起，导致其他的线程没有办法申请锁）。这意味着在任意时刻，多个 lock-free 线程可以同时访问同一数据而不产生数据竞争。上面的定义保证了 lock-free 程序中的一组线程中至少有一个可以顺利的执行而不产生阻塞，从而确保整个程序的顺利执行。lock-free中常见的就是使用cas来代替加锁。
-
CAS
什么是CAS？CAS的本质是把compare-and-swap封装为一个原子操作。伪代码如下：// 伪代码，整个过程是 atomic，不可被中断 bool compare_and_swap(int * p, int old_val, int new_val) { if (*p == old_val) { *p = new_val; return true; } return false; } -
gcc提供的类似的接口:
// for gcc bool __sync_bool_compare_and_swap(type *ptr, type oldval, type newval, ...) type __sync_val_compare_and_swap(type *ptr, type oldval, type newval, ...) // for C++11 template <typename T> bool atomic_compare_exchange(T* pointer, T* expected, T desired); // TODO compare_exchange_weak -
使用CAS的接口可以很方便的实现一个无锁的容器。比如实现一个无锁的栈(stack)
// push 方法 void Stack::push(int val, int debug_thread_info) { while (true) { Node * new_obj = new Node(val); if (new_obj == NULL) {return;} new_obj->extra_info = debug_thread_info; Node *next_ptr = this->_top; new_obj->next = next_ptr; // 将栈顶指针指向新节点，CAS 直到成功 if (_top.compare_exchange_weak(next_ptr, new_obj)) { return; } } }// pop 方法 Node* Stack::pop() { while (true) { Node *cur_ptr = this->_top; if (!cur_ptr) { return NULL; } Node *next_ptr = cur_ptr->next; // 将栈顶指针指向下一节点，CAS 直到成功 if (this->_top.compare_exchange_weak(cur_ptr, next_ptr)) { return cur_ptr; } } }完整代码：
https://github.com/zhaozhengcoder/CoderNoteBook/tree/master/example_code/cas -
ABA问题
一个线程在cas的判断中被切换出去，另外一个线程把a改成了b，又改回了a。等到原线程切换回来，比较当前值和a，发现一致，于是把当前值cas交换，设置为新值。举个栗子：thread1: top a -> b -> c ^ ^ cur_ptr next_ptr 1. thread1 即将执行cas 2. thread2 切入，thread1挂起 3. thread2 连续pop 两个元素 4. thread2 push元素，恰好是a 于是： thread2： a -> c 5. thread1切回来继续执行 thread1判断 top == cur_ptr，于是吧top设置为b。 但是b已经不存在这个链表了，于是stack的top指针被设置错误。 -
如何避免ABA问题
ABA问题的本质是内存回收。用刚才的例子来说，需要保证a被弹出之后，内存不会立刻被回收，下次分配内存的时候，不会在恰好把这段内存分配出去。以至于CAS判断的时候，通过地址判断发现是一致的，但是实际上是有可能被修改过的。常见的一些解决办法:
- 使用版本号
- 使用hazard pointer
- 使用定长数组，这样就可以提前分配好内存，就不涉及内存分配，避免aba出现。（这是一个非常巧妙的思路 很多项目都用到了这招）
无锁队列
对于一些性能要求比较高的场景里面，往往会使用无锁队列来代替锁，提供系统的性能。
- boost提供了三种无锁方案，分别适用不同使用场景。
boost::lockfree::queue 是支持多个生产者和多个消费者线程的无锁队列。 boost::lockfree::stack 是支持多个生产者和多个消费者线程的无锁栈。 boost::lockfree::spsc_queue 是仅支持单个生产者和单个消费者线程的无锁队列，比boost::lockfree::queue性能更好。Boost无锁数据结构的API通过轻量级原子锁实现lock-free，不是真正意义的锁 - Github上面的一些开源实现
单读单写的队列
多读多写的队列实现
-
简单的分析一下它们实现的原理：
1）如果使用的场景是单读单写的话，可以维护一个ringbuff，生产线程和消费线程各自维护一个write index 和 read index。各自维护（写）自己的index。 这种情况可以无需加锁。
2）如果是场景是多读多写的话，由于有多个生产者和消费者，这里需要进行cas操作来维护index。并且由于ring buffer是固定长度的数组，来进行cas操作的时候，不涉及到内存分配，不会引入ABA问题。ringbuff的原理如下：
当空队列时，head与tail相等；当有元素进队，则tail后移；有元素出队，则head后移。
自己实现了一个乞丐版本 github链接
相关参考
网友评论