[LockFree之美] 共享变量的并发读写

共享对象的并发读写

在高性能并发服务器中,对于共享对象的读写是最常见的操作之一,比如全局配置类对象的并发读取和更新,以及更复杂的如copy on write btree、堆栈等的并发读写,最基本的操作都可以简化理解为通过全局共享的指针,并发读取和更新指针所指向对象的操作。
最简单的模型如下所示,一个包含了多个字段的结构体:

struct GConf
{
  int64_t a;
  int64_t b;
  int64_t c;
};

可以通过一个全局指针struct gConf *g_conf进行访问。有多个线程会同时读取或更新这个对象,要求任何时刻读取到的内容一定是“一致的”,即读取到的a/b/c的值一定是相同一次更新的值,而不能出现读到的a是某一次更新的值,而b或c是另外一次更新的值。
我们假设研发平台为x86_64,指令级的原子操作最多是64位(实际intel提供了128bit的原子操作,但是本文后续的讨论的方法只用到64bit的原子操作),因为gConf包含三个8字节整数,无法使用原子操作指令对a/b/c一次性赋值,因此必须通过程序算法来保证并发情况下读取到的内容满足“一致性”。
下面开始从易到难讨论5种处理“共享对象并发读写”的方法。

1. 读写锁
最简单朴素的想法是为这个结构配备一个读写锁,读取操作在读锁保护下执行,更新操作在写锁保护下执行。这个方法在对象读取和更新操作都比较简单,且并发程度不高的情况下是一个不错的选择。但是由于读锁与写锁互相阻塞,在读取或更新操作比较费时的情况下,会出现更新或读取线程被阻塞的风险,在高并发的场景下,这种阻塞不可接受。一个实例是数据库系统的全局schema管理器,每次事务都需要读取schema的内容,而DDL会引起schema更新,它的内容比较复杂,更新操作比较费时,会阻塞住处理事务的线程。

2. 引用计数的问题
为了解决读写操作相互阻塞的问题,使用类似Copy on write的方式,每次更新时都构造新的对象,在更新过程中,读取操作还是读取旧的对象,更新操作完成后,原子的替换掉指针使其指向新的对象。而被替换掉的对象不能立即释放,而是要确保这个对象不再被继续使用的情况下,才能将其释放掉。因此这里使用引用计数机制,在读取前将引用计数加1,读取完毕后将引用计数减1,当引用计数减到0的时候,表示没人在使用这个对象,可以将其释放掉。
引用计数在没有并发保护情况下面临的问题是,取得全局指针g_conf和通过这个指针操作对象的引用计数这两个操作并非原子的。可能存在的情况是,一个线程T1取得g_conf的值保存到寄存器后,就被切换出去,另外一个线程T2将g_conf替换后,判断旧的对象引用计数为0,则将其释放。当T1被切换回来继续执行时,它在寄存器中保存的g_conf指向的对象已经被释放,访问对象引用计数的这一次操作就已经存在非法内存访问的风险。
下面两节讨论两种解决并发引用计数的方案。

3. 带锁的引用计数
解决第2点提出的原子性问题,一个最直接的方式是使用锁,将读写g_conf和修改引用计数两个操作放在一个独立全局锁(g_lock)的临界区执行,代码示例如下:

读取并将引用计数加1:

fetch()
{
  g_lock.rdlock();
  g_conf->inc_ref();
  ret = g_conf;
  g_lock.unlock();
}

将引用计数减1:

revert(ptr)
{
  g_lock.rdlock();
  if(0 == ptr->dec_ref())
  {
    free(ptr);
  }
  g_lock.unlock();
}

使用新的对象替换旧对象:

set_new_conf(new_conf)
{
  g_lock.wrlock()
  if (0 == g_conf->dec_ref())
  {
    free(g_conf);
  }
  new_conf->inc_ref();
  g_conf = new_conf;
  g_lock.unlock();
}

在实际场景中,通常读取g_conf的操作更多且并发量大,而替换g_conf的操作不会经常发生,因此读取g_conf时加共享锁,替换g_conf时加互斥锁。
使用锁保护引用计数的方式在高并发的情况下存在一些局限:一方面可能存在读者或写者被饿死的情况;另一方面多个写者无法并发,应用在copy on write btree等数据结构中时,多个写者相互阻塞的问题会影响到整个数据结构的并发性能。

4. 无锁的引用计数
因此,一个更进一步的设计是使用无锁的引用计数,即无论读取g_conf还是替换g_conf都不在锁中进行,如下图所示:share_obj_seq_ref

RefNode的结构包含了指向GConf对象的指针data_ptr,以及引用计数ref_cnt,自增序列号seq_id,ref_cnt和seq_id长度都为4字节且地址连续,可以经由一次8字节的原子操作进行原子性的修改和读取。每次构造新的GConf对象的要同时分配新的RefNode来保存GConf对象的地址。
RefNode对象的分配和释放由专用的分配器进行分配和重用,一旦分配就不会释放回系统。引用计数的加1和减1,操作的是RefNode对象的ref_cnt字段,当引用计数减为0时,将data_ptr指向的对象释放,并将seq_id加1,然后将RefNode对象释放给内存分配器。已释放回分配器的RefNode对象仍然可以被安全的访问到。
替代全局GConf指针,我们增加了一个被称为GNode结构体的全局对象称为g_node,它包含了指向RefNode对象的指针称为node_idx,和这个RefNode中seq_id字段的值node_seq。node_idx和seq_id长度都为4字节且地址连续,可以经由一次8字节的原子操作进行原子性的读取和替换。
读取和替换操作的代码示例如下:
将引用计数加1并返回:

fetch()
{
  GNode tmp_g_node = atomic_load(g_node);
  while (true)
  {
    if (tmp_g_node.node_idx->atomic_check_seq_and_inc_ref(tmp_g_node.node_seq))
    {
      ret = tmp_g_node.node_idx;
      break;
    }
    else
    {
      tmp_g_node = atomic_load(g_node);
    }
  }
}

将引用计数减1:

revert(node_idx)
{
  if (node_idx->atomic_dec_ref_and_inc_seq())
  {
    free(node_idx->data_ptr);
    free_list.free(node_idx);
  }
}

使用新的对象替换旧对象:

set_new_conf(new_conf)
{
  RefNode *new_node = free_list.alloc();
  new_node->ref_cnt = 1;
  new_node->data_ptr = new_conf;

  GNode new_g_node;
  new_g_node.node_seq = new_node->seq_id;
  new_g_node.node_idx = new_node;

  GNode old_g_node = atomic_store_and_fetch_old(&g_node, new_g_node);
  if (old_g_node.node_idx->atomic_dec_ref_and_inc_seq())
  {
    free(old_g_node.node_idx->data_ptr);
    free_list.free(old_g_node.node_idx);
  }
}

其中几个原子操作的作用:
1. RefNode::atomic_check_seq_and_inc_ref原子性的判断如果seq_id等于传入参数就将ref_cnt加1,并返回true,否则返回false。
2. atomic_store_and_fetch_old原子性的将第二个参数赋值给第一个参数,并返回第一个参数的旧值。
3. RefNode::atomic_dec_ref_and_inc_seq原子性的将ref_cnt减1,并判断ref_cnt如果已减为0,则将seq_id加1并返回true,否则返回false。
工程实现上,为了达到node_idx长度只能有4字节的要求,RefNode分配器设计为定长数组,node_idx为数组下标,使用定长无锁队列作为分配器来维护空闲的node指针。一般来说一个线程只会增加一次引用计数,因此对于维护GConf这样全局只有一个的对象,定长RefNode数组的长度初始化为与线程个数相等就够用了。

5. HazardPointer
引用计数基本可以解决共享对象并发读写的问题,但它仍然存在一些不足:第一,每次读取都需要使用原子操作修改全局引用计数,高并发情况下的对同一个变量的原子操作可能成为性能瓶颈;第二,管理RefNode对象的无锁分配器有额外的管理和维护成本,增加的实现复杂度;第三,针对每个共享对象都需要维护N个(N为线程个数)RefNode,如果共享对象数量较多(比如Btree节点),为节省空间,还需要额外的机制来避免大量使用RefNode。
因此再介绍一种被称为HazardPointer的思路,它由Maged M. Michael在论文“Hazard Pointers: Safe Memory Reclamation for Lock-Free Objects”中提出,基本思路是将可能要被访问到的共享对象指针(成为hazard pointer)先保存到线程局部,然后再访问,访问完成后从线程局部移除。而要释放一个共享对象时,则要先遍历查询所有线程的局部信息,如果尚有线程局部保存有这个共享对象的指针,说明这个线程有可能将要访问这个对象,因此不能释放,只有所有线程的局部信息中都没有保存这个共享对象的指针情况下,才能将其释放。
读取和替换操作的代码示例如下:

g_hp_array[N]; //保存每个线程局部信息的数组,N为线程个数

读取操作:

while(true)
{
  ret = g_conf;
  g_hp_array[thread_id] = ret;
  if (g_conf == ref)
  {
    break;
  }
  else
  {
    g_hp_array[thread_id] = NULL;
  }
}
// 读取逻辑代码…
g_hp_array[thread_id] = NULL;

使用新的对象替换旧对象:

set_new_conf(new_conf)
{
  retired_ptr = atomic_store_and_fetch_old(&g_conf, new_conf);
  found = false;
  for (i = 0; i < g_hp_array.length(); i++)
  {
    if (retired_ptr == g_hp_array[i])
    {
      found = true;
      break;
    }
  }
  if (!found)
  {
    free(retired_ptr);
  }
}

HazardPointer的设计解决了引用计数原子操作的性能瓶颈,避免了实现额外RefNode无锁分配器的复杂度。但是同时引入一个新的问题:由于回收内存的操作需要遍历查询全局数组g_hp_array,代价比较大,通常工程上的做法是将待回收的指针暂存在线程局部,等数量超过一个配置门限时,才进行批量回收。使得回收内存产生延迟,并且回收操作本身由于需要遍历数组,也会消耗较多的时间。
关于并发场景的问题,由于回收操作便利g_hp_array并不能保证原子的遍历,而HazardPointer的设计要求对象已经被放入g_hp_array后是能够安全访问的,因此可能存在如下时序:
1. 读取线程拿到g_conf指针存入局部变量ret
2. 更新线程将g_conf指针替换,扫描g_hp_array没有找到g_conf,释放g_conf内存
3. 读取线程将刚刚拿到的ret放入g_hp_array,然后开始使用ret,而此时ret指向的内存已经释放
为了处理这个问题,如上述伪代码所示,读取线程需要增加double check逻辑,判断如果从拿到g_conf到放入g_hp_array之间,可能有回收逻辑遍历过g_hp_array,则要重试。
因为替换g_conf意味着要将旧的g_conf释放,而释放意味着要先会遍历g_hp_array,因此读取线程只需要在将g_conf放入g_hp_array后,判断g_conf是否变化,如果变化了,则说明g_conf可能正好在从拿到g_conf到放入g_hp_array之间被释放了,因此需要重试。

6. HazardVersion

HazardPointer的实现简单,但是与引用计数法有着相似的不足:需要为每个共享变量维护O(N)个(N为线程个数)槽位,使用者需要仔细分析算法以尽量减少同时存在的hazard pointer 或者引入其他机制封装多个共享变量称为一个hazard pointer。这样就给使用者带来了比较大的难度,HazardPointer机制也与具体数据结构的实现比较紧的耦合在一起。

因此在HazardPointer基础上发展出了被称为HazardVersion技术,它提供类似lock一样的acquire/release接口,支持无限制个数共享对象的管理。

与HazardPointer的实现不同:首先全局要维护一个int64_t类型的GlobalVersion;要访问共享对象前,先将当时的GlobalVersion保存到线程局部,称为hazard version;而每次要释放共享对象的时候先将当前GlobalVersion保存在共享对象,然后将GlobalVersion原子的加1,然后遍历所有线程的局部信息,找到最小的version称为reclaim version,判断如果待释放的对象中保存的version小于reclaim version则可以释放。

读取和替换操作的代码示例如下:

g_version; //全局GlobalVersion
g_hv_array[N]; //保存每个线程局部信息的数组,N为线程个数

读取操作:

g_hv_array[thread_id] = g_version;
ret = g_conf;
// 读取逻辑代码…
g_hv_array[thread_id] = INT64_MAX;

使用新的对象替换旧对象:

set_new_conf(new_conf)
{
  retired_ptr = atomic_store_and_fetch_old(&g_conf, new_conf);
  retired_ptr->set_version(atomic_fetch_and_add(&g_version, 1));
  reclaim_version = INT64_MAX;
  for (i = 0; i < g_hv_array.length(); i++)
  {
    if (reclaim_version > g_hv_array[i])
    {
      reclaim_version = g_hv_array[i];
    }
  }
  if (reclaim_version > retired_ptr->get_version())
  {
    free(retired_ptr);
  }
}

可以看到,读取操作先保存了GlobalVersion到线程局部,然后去读取g_conf指针,虽然可能同时发生的回收操作遍历g_hv_array并不保证原子性,但是即使一个更小的hazard version被放入g_hv_array而错过了遍历,也不会有风险。

因为回收操作是先更新g_conf指针后遍历g_hv_array,而读取操作是先更新g_hv_array后读取g_conf指针,所以最坏情况下读取操作与回收操作同时发生,且读取操作放入g_hv_array的hazard version被回收操作遍历时错过,那么读取操作读到的g_conf一定是被回收操作更新之后的,本次不会被回收,可以安全的访问。

HazardVersion相对HazardPointer,在用法上更加简单,无需针对具体的数据结构进行分析,局限是需要在被管理的对象中保存version。其次,与HazardPointer一样,由于遍历g_hv_array比较费时,因此也需要引入类似的延迟回收机制。需要注意的是,它的缺点也比较明显,一旦出现一个线程长时间不释放HazardVersin时,会阻塞住后面更大Version对象的释放。

7. 总结
本文讨论了“读写锁”、“带锁的引用计数”、“无锁的引用计数”、“HazardPointer”、“HazardVersion”五种方法来处理共享对象的并发读写问题,在实际工程领域,“带锁的引用计数”比较常见,在特定场景下优化读写锁可以得到不错的性能,并且很易于理解和实现;“无锁的引用计数”实现复杂但并发读写性能卓越,在已开源的OceanBase版本中也有实践;“HazardPointer”实现简单,但是理解和使用都较复杂;“HazardVersion”是一种基于“HazardPointer”的创新思路,有一定的局限性,凭借简单易用的接口,可以在很多场景替代“HazardPointer”。

[LockFree之美] 共享变量的并发读写》上有4条评论

  1. nnkken

    我想请教一下关於Hazard Pointer和Hazard Version的部分,如果在set_new_conf时,旧对象因为有线程正在使用而没有free,那麽这个对象由谁负责free?
    在两者的代码里,都只有set_new_conf会执行free,而它只负责free「上一个旧g_conf」,所以如果它没有执行free的话,这个g_conf不就泄漏了吗?

    回复
    1. yubai 文章作者

      文中有提到
      “将待回收的指针暂存在线程局部,等数量超过一个配置门限时,才进行批量回收”

      回复

发表评论

电子邮件地址不会被公开。 必填项已用*标注