实现无锁的栈与队列(5)：Hazard Pointer

两年多之前随手写了点与 lock free 相关的笔记：1,2,3,4，质量都不是很高其实(读者见谅)，但两年来陆陆续续竟也有些阅读量了(可见剑走偏锋的技巧是多容易吸引眼球)。笔记当中在解决内存释放和 ABA 问题时提到了 Hazard Pointer 这个东西，有两三个读者来信问这是什么，让详细讲一下，我想了想，反正之前在看这东西的时候也记了些东西，干脆整理一下发出来。

前面写的那几篇笔记都来源于 Maged Michael 的学术论文，Hazard pointer 也是他的创想，academic paper 的特色之一就是常常有些美好的假设，关于 hazard pointer 也一样如此，如下的讨论均假设内存模型是 sequential consistent 的，不然仍是问题多多。

核心问题

Hazard Pointer(如下简称为 HP) 要解决的核心问题是怎样安全地释放内存，该问题的解决在实现无锁算法时有两个关键的影响：

1. 保证了关键节点的访问是合法的，不会致使程序尝试去读取已经释放了的内存。
2. 保证了 ABA 问题不会出现，程序逻辑正确的前提。

这两个问题在写无锁代码时基本是没法避免的，走这条路终会赶上，多少人所以费尽心力穷尽技巧各类花样，只为把这问题完全解决。HP 就是这众多花样各类技巧中的一种，它的作法以个人愚见也不是很完美，但实现上比较简单，不依赖具体系统，也不对硬件有特殊要求(固然 CAS 操做仍是要的)，从效果上看也凑和，所以不管怎样是值得参考学习的。

具体实现

在无锁算法中释放内存之因此难，主要缘由在于，当一个线程准备释放一块内存时，它没法知道是否另有别的线程也同时持有该块内存的指针并须要访问，所以解决这个难点的一个直接想法就是，在每一个线程获取了一个关键内存的指针后，该线程将设置一个标志，代表"我正在操做这个关键数据，大家谁都别给我随便就释放了"。固然，这个标志须要放在一个公共区域，使得任何线程均可以去读。当另外一个线程想要释放一块内存时，它就去把每一个线程的标志都看一下，看看是否有别的线程也在操做这块内存，从而决定是否立刻释放该内存：若是有别的线程在操做该内存，则暂时不释放，等下次。具体实现以下：

1. 创建一个全局数组 HP hp[N]，数组中的元素为指针，称为 Hazard pointer，数组的大小为线程的数目，即每一个线程拥有一个 HP。
2. 约定每一个线程只能修改本身的 HP，而不容许修改别的线程的 HP，但能够去读别的线程的 HP 值。
3. 当线程尝试去访问一个关键数据节点时，它得先把该节点的指针赋给本身的 HP，即告诉别人不要释放这个节点。
4. 每一个线程维护一个私有链表(free list)，当该线程准备释放一个节点时，把该节点放入本身的链表中，当链表数目达到一个设定数目 R 后，遍历该链表把能释放的节点统统释放。
5. 当一个线程要释放某个节点时，它须要检查全局的 HP 数组，肯定若是没有任何一个线程的 HP 值与当前节点的指针相同，则释放之，不然不释放，仍旧把该节点放回本身的链表中。

HP 算法主要用在实现无锁的队列上，所以前面的具体步骤其实基于如下几个假设：

1. 队列上的元素任什么时候候，只可能被其中一个线程成功地从队列上取下来，所以每一个线程的 free list 中的元素确定是惟一的。
2. 线程在操做无锁队列时，任什么时候候基本只须要处理一个节点，所以每一个线程只须要一个 HP 就够了，若是有特殊需求，固然 HP 的数目也能够相应扩展。
3. 对于某个节点来讲，多个线程同时持有该节点的指针这个现象，在时间上是很是短暂有限的，只有当这几个线程同时尝试去取下该节点，它们才可能同时持有该节点的指针，一旦某个线程成功地将节点取下，其它线程很快就会发现，并尝试继续去操做下一下节点，然后续再来取节点的线程则再也不可能得到已经不在无琐队列上的节点的指针，所以：当某个线程尝试去检查其它线程的 HP 时，它只须要将 HP 数组遍历一遍就够了，不用担忧各线程 HP 值的变化。

如下为我从论文里翻译过来的伪代码，入队列的函数不涉及删除节点所以不会操做 HP，难点都在处理出队列的函数上：

using hp_t = void*;

hp_t hp[N] = {0};

// 如下为队列的头指针。
node_t* top;

data_t* Pop()
{
   node_t* t = null;
   while (true)
   {
      t = top;
      if (t == null) break;

      // 设置当前线程的 HP
      hp[this_thread] = t;

      // 如下这步是必须的，确认了当前 HP 在 t 被释放前已经被设置到当前线程的 HP 中。
      if (t != top) continue;

      node_t* next = t->next;
      if (CAS(&top, t, next)) break;
   }

   // 已经再也不持有任何节点，需将本身的 HP 设为空.
   hp[this_thread] = null;

   if (t == null) return null

   data_t* data = t->data;
   
   // 尝试释放节点
   DeleteNode(t);

   return data;
}

以上是出队列的代码，显然，所作的事情很是直白：线程拿到一个节点后将数据取出，并尝试释放节点。释放节点是另外一个关键点，具体实现参看以下伪代码：

thread_local vector<hp_t> free_list;

void DeleteNode(node_t* t)
{
   free_list.push_back(t);
   if (free_list.size() > R) FreeNode();
}

void FreeNode()
{
  vector<hp_t> hp_list;
  hp_list.reserve(N);

  // 获取全部线程的 HP，如非空则保存到 hp_list 中。
  for (int i = 0; i < N; ++i)
  {
    if (hp[i] == null) continue;

    hp_list.push_back(hp[i]);
  }

  std::sort(hp_list);

  vector<hp_t> not_free;  
  not_free.reserve(free_list.size());

  // 把当前线程的 free_list 遍历遂一进行释放。
  for (int i = 0;i < free_list.size(); ++i)
  {
    if (std::binary_search(hp_list.begin(), hp_list.end(), free_list[i]))
    {
      // 某个线程持有当前节点，故不能删除，仍是保留在队列里。
      not_free.push_back(free_list[i]);
      continue;
    }

    // 确认没有任何线程持有该节点，删除之。
    delete free_list[i];
  }

  free_list.swap(not_free);
}

存在的问题

看到这里相信读者对 Hazard Pointer 的原理已经大概了解了，那么咱们来简单总结一下上面的实现。

首先是效率问题，它够快吗？根据前面的伪代码，显然影响效率的关键点在FreeNode()这个函数上，该函数有一个双重循环，但还好第二重循环用了二分查找，所以删除 R 个节点总的时间效率理论上是 O(R*logN)，R 能够设置， N 是线程数目，一般也不会太大，所以均摊下来应该还好？我只能说不知道，看使用场景吧，用无琐通常有很高的效率需求，这里加个这样复杂度的处理是否会比加琐更快呢？也说不许，实现上复杂了是确定的，想用的话得好好测试测试看看划不划得来。

其次是易用性，HP 释放节点是累进制的，只有当一个线程积累到了必定数量的节点才批量进行释放，而生产环境里一般状况复杂，会不会某个线程积累了几个节点后，就再也不去队列里 pop 数据了呢？岂不是总有些节点不能释放？内心有些疙瘩。。除此，现代操做系统里线程建立销毁其实很频繁，某个线程若是要退出了，还得记得把本身头上的节点释放一下，也是件麻烦事。有人可能会以为为何删除节点时要把节点放到队列里再删？画蛇添足！直接遍历 HP 数组直到没有线程持有该节点不就行了 --- 放到队列里实际上是为效率，不然每 pop 一次就遍历一遍 HP list，并且搞很差还要反复等待某个线程释放节点，均摊下来效率过低。

最后，还有一个问题，相信读者忍了好久了，HP 数组那里，各个线程怎么 index 进去取出本身的 HP 呢？ thread id 吗？那这个数组不得很大很大很大？

一点改进

关于 HP 数组的实现上，做者其实也看到了问题，提出能够用 list 来管理 HP，由于不是每一个线程都必须固定分配一个 HP，事实上只有当该线程正在进行 pop 操做的时候它才须要，pop 完了立刻就能够把 HP 还回去了，所以数组能够用链表来替换，固然这个链表也得是 Lock free 的，但这个链表能够不用考虑回收和释放实现上容易多了，和我在本系列文章的第四篇里提到的思路是一致的。

但这样用 List 来代替数组在必定程度也增长了效率负担，由于每一个线程取出 HP 变得更慢了(首先是很容易引发多个线程冲突，其次用到了 CAS 以及函数调用的开销)，固然具体有多少效率损失还得看使用场景，须要好好测量一下---写无琐代码不能少作的事情。

无琐编程很难，但这并不表明它们所以只能是理论游戏，Maged Michael 的无琐系列文章启发了不少人，这其中也包括 c++ 里的大腕 Andrei Alexandrescu，呐呐，看这里。