实现无锁的栈与队列(4)
现在我们来尝试解决前一篇文章提到的问题。
(一)
首先是内存释放的问题。
这个问题乍看起来很棘手：我们现在要访问一段内存，但无从知道这段内存是否还合法，是否已被释放。
很直接的一个想法是，看看有没别的办法判断该内存是否合法，这个想法很单纯，但从前面几篇文章的讨论我们得知，任何时候直接去碰队列上的节点都是不安全的，当前线程永远不知道下一秒后会发生了什么事情，这就是为什么lock free queue需要引入一个dummy 头结点的原因。
既然这样，那么我们能不能干脆简单点，直接就不允许释放链表的节点呢？
这个方案确实是最直接易用的，所付出的代价也最小，无非就是多费点内存，空间换效率，太划算了，boost 的lock free 库就采用了这种方法。
(1)创建队列的时候，分配好全部的内存，比如说，2048个节点。
(2)重复实现一套无锁分配节点的方法。
其中第二条看起来有些为难，这不正是我们现在所要解决的问题吗？
事实上这不大一样，在这里我们不需要再分配内部节点！因此，我们不需要担心内存回收的问题！只要处理好aba问题就行了！
struct Node { Node* next; //用于在lock free queue中指向下一个指点 Node* next2;// 指向内部队列 void* data; }; Node g_FreeList[N];
Node* head; void Init() { g_FreeList = (Node*)malloc(sizeof(Node)*N); for (int i = 0; i < N -1; ++i) { g_FreeList[i].next2 = &g_FreeList[i+1]; } g_FreeList[N-1].next2 = NULL; } Node* AllocNode() { Node* old_head; do { old_head = head; if (old_head == NULL) return NULL;
//下面的一行仍有aba问题，后面再解决。 if (CAS(&head, old_head, old_head->next2)) break; } while(1); return old_head; } void ReleaseNode(Node* node) {
assert(node);// more advance check is necessary Node* old_head; do { old_head = head; node->next2 = old_head; if (CAS(&head, old_head, node)) break; } while( 1); }
(二)
现在我们来看看ABA问题，回过头仔细观察一下ABA问题， 它的起因简单来说就在于dequeue的时候，无法确认head是否还是当初的head, 也无法确认它的内容是否已经发生变化，因此无法更新当前的头结点指针。所以解法最直观的无外乎两个：
1) 在当前线程还在操作该节点时，不允许别的线程释放这个节点。
2) 给节点做标志，使得每个插入的节点有一个唯一的标记，这样，就能检测当前的节点是否已发生变化。
其中第一种做法在C/C++中不容易做到，它们在语言层面上没有GC, 对内存的操作都要靠程序员自己来把控，使得在处理资源的回收时，虽然更灵活，但也更不容易实现一些诸如自动回收这样的高级功能，不过这难不倒聪明人，2004年时候，Maged.M.Machel(对，又是他), 在IEEE的期刊 Transactions on Parallel and Distributed Systems上发一发表了一篇论文：Hazard Pointers: Safe Memory Reclamation for Lock-Free Objects
该论文引入一个叫作hazard pointer的东西来处理ABA问题，关于Hazard pointer的简单介绍可以参考一下wiki中的条目。归根到底，hazard pointer是实现了一种reference的机制，使得链表的节点如果还有线程在读，就不允许该节点被释放，这个方法实现起来有很多的细节要处理，并不是件容易做的事情，维基百科的附录里面介绍了好几种不同的人的实现方案，有兴趣的读者可以自行去研究研究。我在前一篇文章中提到过的Christian Hergert也在他的博客中介绍了他自己的hazard pointer的实现，代码放到了github上，非常强大！
阻止内存过早被释放这个做法不是件容易的事情，但如果做到了，就连我们上面讨论的内存访问的问题都一并解决了。Memory reclamation是无锁算法里最棘手的两个问题之一了，Hazard Pointer在这个难题上是个很完美的解决方案。
但是Hazard Pointer来头太大，也太麻烦了，有没更轻量一点的方法呢？现在我们来看看第二种解法。为了说明第二种方法，我们来回顾一下lock free queue中dequeue的操作。
1 gpointer queue_dequeue(Queue *q) 2 { 3 Node *node, *tail, *next; 4 5 while (TRUE) { 6 head = q->head; 7 tail = q->tail; 8 next = head->next; 9 if (head != q->head) 10 continue; 11 12 if (next == NULL) 13 return NULL; // Empty 14 15 if (head == tail) { 16 CAS(&q->tail, tail, next); 17 continue; 18 } 19 20 data = next->data; 21 if (CAS(&q->head, head, next)) 22 break; 23 } 24 25 g_slice_free(Node, head); // This isn't safe 26 return data; 27 }
所有的问题归结起来，就在于第21行进行cas操作时，head虽然还是head，但head->next已经发生了变化。那么，我们应该怎样来识别这些变化呢？
从本质上来说，既然head已经发生了变化，那接下来的CAS就应该要失败才是正确的行为。ABA问题的根源就在于该失败的CAS操作没有失败，所以，我们现在的目标就是要纠正CAS的这个错误行为，让它在该失败的时候就彻底的失败。
回头来分析一下cas操作
1 bool cas(type*ptr, type old, type new)
这个函数纯粹只是比较一下ptr 与 old的值，然后决定下一步的操作：
如果 *ptr == old，就*ptr = new，否则什么也不做(暂且这样理解)。
在我们的场景下，我们希望在aba问题出现了的时候，cas能够失败。为了做到这点，我们自然希望*ptr != old，但aba问题出现时，*ptr 是等于 old的，因此我们在进行cas时不应该只比较*ptr == old, 而应该想办法在*ptr中加入些不同的东西来加以区别，比如说再多比较几个字节，再决定是否更新*ptr: 我们需要cas能比较的字节数要大于字长(sizeof(void*))，这个要求显然是需要cpu的支持的。因此，我们现在讨论的这个解法并不具备普遍性，是要依赖硬件的。这大概也是为什么Maged.M.Machel花了大心思是去研究出hazard pointer的原因。好消息是，x86平台上较新的cpu都是支持double wide cas的，也就是通常指的CAS2，具体来说，就是支持cmpxchg8b, cmpxchg16b这两条指令。
有了CAS2的支持，我们就可以对指向指向节点的指针加一个tag作一标记。
1 union DoublePointer 2 { 3 void* vals[2]; 4 atomic_longlong val; 5 };
DoublePointer包含了指向结点的指针，以及一个tag，每次插入一个节点时，都用一个DoublePointer来指向这个新插入的结点，每个DoublePointer中包含了唯一的标记符，每次插入新结点或取出结点，都用CAS2来更新double pointer，从而就做到区别对待每一个新插入的结点，从根本上去除了ABA问题。
语言上比较难说得清楚，还好可以用代码来说话。
有兴趣取的读者可以看看我放在github上的代码，在x86-64/32上都进行了一定的压力测试，应该是没问题的。
https://github.com/kmalloc/back-end-facility/blob/master/misc/src/lock-free-list.cc
后话
好了，写lock free queue的目标到此算是基本完成了，花了一个多月的时间，一开始先是读了很多的文章，从无到有，算是在内存模型，cpu结构方面有了些前所未有的了解，不过就算这样，真正写起来还是比想像中的困难太多了，尤其是debug的过程，刚开始时，遇到问题简直束手无策，事实证明，思路清晰才是解决问题的根本方法，不能一发现问题就挂gdb，那是没用的，特别是多线程的情况下，必须一点一点的分析代码，认真推敲，查找漏洞，挂gdb应该只做验证之用，打Log其实更好了。
四篇文章写下来，lock free queue的实现过程基本是个重造轮子的过程，说到通用性，可靠性那是没法和 boost相比的，性能的话，也不一定比得上，唯一值得安慰的地方，就是它们是我的亲儿子了T_T.