Fasterkv Epoch 保护框架

Epoch 支持对任意全局操作的惰性同步（lazy synchronization）。虽然像 Silo、Masstree 和 Bw-Tree 这样的系统已经将 epoch 用于特定的目的。

Epoch Basics。系统维护一个共享的原子计数器 E，称为当前 epoch，可以由任何线程递增。每个线程 T 都有一个本地版本的 E，用 ET 表示。线程定期刷新它们的本地 epoch 值。所有线程本地的 epoch 值 ET 存储在一个共享的 epoch table 中，每个线程有一个高速缓存行。如果所有线程都具有比 c 更高的 ET，即任意线程的 ET > c，则称 epoch c 是安全的。请注意，如果 epoch c 是安全的，则小于 c 的所有 epoch 也是安全的。我们还维护了一个全局计数器 ES，它跟踪当前最大的安全 epoch。ES 通过扫描 epoch 表中所有的条目得到，并在线程刷新其 epoch 时更新。系统保持以下不变式：任意线程 : ES < ET ≤ E。

Trigger Actions。当一个 epoch 变的安全时，使用触发器操作来增强基础的 epoch 框架，从而提高执行任意全局操作的能力。当增加当前 epoch 时，比如说从 c 到 c +1，线程可以另外关联一个 action，该 action 将由系统在 epoch c 是安全时触发。这可以使用 drain-list 来实现，这是一个 (epoch,action) 对列表，其中 action 是在 epoch 安全后必须调用的回调代码片段。该表是使用一个小数组来实现的，每当 ES 更新时，该数组都会扫描准备触发的操作。我们在数组上使用原子的 compare-and-swap 来确保一个 action 只执行一次。为了增强可伸缩性，我们只在当前 epoch 发生变化时才重新计算 ES 并扫描排出列表。

在 Faster 中使用 epoch 框架来协调系统操作，Fasterkv 的 Epoch 保护框架代码实现在 FASTER/cc/src/core/light_epoch.h 中，主要使用到该保护框架的地方有：内存安全的垃圾回收、循环缓冲区维护和页面刷新、共享日志页面边界维护和检查点。

malloc_fixed_page_size.h
FixedPageArray* MallocFixedPageSize::ExpandArray() 内 array 的删除
 
persistent_memory_malloc.h
Address PersistentMemoryMalloc::ShiftReadOnlyToTail()
inline void PersistentMemoryMalloc::PageAlignedShiftHeadAddress()
inline void PersistentMemoryMalloc::PageAlignedShiftReadOnlyAddress()
 
file_system_disk.h
core::Status OpenSegment(uint64_t segment)
void TruncateSegments()
storage.h
void ShiftRemoteOffset(Address tailFile)

Bw-Tree 利用 Epoch 完成了安全的垃圾回收

The Bw-Tree: A B-tree for New Hardware Platforms

D. Garbage Collection

A latch-free environment does not permit exclusive access to shared data structures (e.g., Bw-tree pages), meaning one or more readers can be active in a page state even as it is being updated. We do not want to deallocate memory still accessed by another thread. For example, during consolidation, a thread “swaps out” the old state of a page (i.e., delta chain plus base page) for a new consolidated state and requests that the old state be garbage collected. However, we must take care not to deallocate the old page state while another thread still accesses it. Similar concerns arise when a page is removed from the Bw-tree. That is, other threads may still be able to access the now removed page. We must protect these threads from accessing reclaimed and potentially “repurposed” objects by preventing reclamation until such access is no longer possible. This is done by a thread executing within an “epoch”.

An epoch mechanism is a way of protecting objects being deallocated from being re-used too early. A thread joins an epoch when it wants to protect objects it is using (e.g., searching) from being reclaimed. It exits the epoch when this dependency is finished. Typically, a thread’s duration in an epoch is for a single operation (e.g. insert, next-record). Threads “enrolled” in epoch E might have seen earlier versions of objects that are being deallocated in epoch E. However, a thread enrolled in epoch E cannot have seen objects deallocated in epoch E-1 because it had not yet started its dependency interval. Hence, once all threads enrolled in epoch E have completed and exited the epoch (“drained”), it is safe to recycle the objects deallocated in epoch E. We use epochs to protect both storage and deallocated PIDs. Until the epoch has drained such objects cannot be recycled.

d .垃圾收集

无锁环境不允许对共享数据结构(例如，Bw-tree页面)的独占访问，这意味着即使在页面状态正在更新时，也可以有一个或多个 read 处于活动状态。我们不希望释放仍然被另一个线程访问的内存。例如，在整合期间，线程将页面的旧状态(即增量链加上基页)“交换”为新的整合状态，并请求对旧状态进行垃圾收集。但是，我们必须注意不要在另一个线程仍然访问旧的页面状态时释放它。当从 bw-tree 中删除页面时，会出现类似的问题。也就是说，其他线程仍然可以访问现在删除的页面。我们必须防止这些线程访问已回收和潜在的“重新使用”对象，方法是阻止回收，直到这种访问不再可能。这是由在“epoch”内执行的线程完成的。

Epoch 机制是一种保护被释放对象不被过早重用的方法。当线程想要保护它正在使用的对象(例如，搜索)不被回收时，它会加入一个 epoch。当此依赖项完成时，它将退出 epoch。通常，线程在一个 epoch 中的持续时间是针对单个操作(例如插入、下一个记录)的。在 epoch E 中“注册”的线程可能已经看到了在 epoch E 中被释放的对象的早期版本。然而，在 epoch E 中注册的线程无法看到在epoch E-1 中被释放的对象，因为它还没有开始它的依赖间隔。因此，一旦在 epoch E 中注册的所有线程都完成并退出了 epoch(“耗尽”)，回收在 epoch E 中释放的对象是安全的。我们使用 epoch 来保护存储和释放的 pid。在时代耗尽之前，这些物体不能被回收。

Fasterkv 中的实现，在调用 BumpCurrentEpoch() 时会将 current_epoch 递增，并计算出当前安全的 Es=min(table_[entry].local_current_epoch)-1;。接着处理 drain_list_ 内小于  Es的  action。完成上述操作后，将本阶段的 Et 和 action 组成 (epoch,action) 并添加到 drain_list_ 中。

/// This thread is not protecting any epoch.
static constexpr uint64_t kUnprotected = 0;
 
/// Enter the thread into the protected code region
inline uint64_t Protect() {
  uint32_t entry = Thread::id();
  table_[entry].local_current_epoch = current_epoch.load();
  return table_[entry].local_current_epoch;
}
 
/// Exit the thread from the protected code region.
void Unprotect() {
  table_[Thread::id()].local_current_epoch = kUnprotected;
}
 
/// Compute latest epoch that is safe to reclaim, by scanning the epoch table
uint64_t ComputeNewSafeToReclaimEpoch(uint64_t current_epoch_) {
  uint64_t oldest_ongoing_call = current_epoch_;
  for(uint32_t index = 0; index < kTableSize; ++index) {
    uint64_t entry_epoch = table_[index].local_current_epoch;
    if(entry_epoch != kUnprotected && entry_epoch < oldest_ongoing_call) {
      oldest_ongoing_call = entry_epoch;
    }
  }
  safe_to_reclaim_epoch = oldest_ongoing_call - 1;
  return safe_to_reclaim_epoch;
}
 
void Drain(uint64_t nextEpoch) {
  ComputeNewSafeToReclaimEpoch(nextEpoch);
  for(uint32_t idx = 0; idx < kDrainListSize; ++idx) {
    uint64_t trigger_epoch = drain_list_[idx].epoch.load();
    // safe_to_reclaim_epoch 就是 Es
    if(trigger_epoch <= safe_to_reclaim_epoch) {
      if(drain_list_[idx].TryPop(trigger_epoch)) {
        if(--drain_count_ == 0) {
          break;
        }
      }
    }
  }
}
 
/// Increment the current epoch (global system state)
uint64_t BumpCurrentEpoch() {
  uint64_t nextEpoch = ++current_epoch;
  if(drain_count_ > 0) {
    Drain(nextEpoch);
  }
  return nextEpoch;
}
 
/// Increment the current epoch (global system state) and register
/// a trigger action for when older epoch becomes safe to reclaim
uint64_t BumpCurrentEpoch(EpochAction::callback_t callback, IAsyncContext* context) {
  uint64_t prior_epoch = BumpCurrentEpoch() - 1;
  uint32_t i = 0, j = 0;
  while(true) {
    uint64_t trigger_epoch = drain_list_[i].epoch.load();
    if(trigger_epoch == EpochAction::kFree) {
      if(drain_list_[i].TryPush(prior_epoch, callback, context)) {
        ++drain_count_;
        break;
      }
    } else if(trigger_epoch <= safe_to_reclaim_epoch.load()) {
      if(drain_list_[i].TrySwap(trigger_epoch, prior_epoch, callback, context)) {
        break;
      }
    }
    if(++i == kDrainListSize) {
      i = 0;
      if(++j == 500) {
        j = 0;
        std::this_thread::sleep_for(std::chrono::seconds(1));
        fprintf(stderr, "Slowdown: Unable to add trigger to epoch\\n");
      }
    }
  }
  return prior_epoch + 1;
}

在Fasterkv 里面，record 级别的并发操作还是要用 cas 或加锁。

混合日志（HybridLog）是一种新颖的数据结构，它结合了就地更新（在内存中）和日志结构组织（在磁盘上），同时提供了对记录的无锁并发访问。混合日志跨越内存和二级存储，其中内存中的数据充当热点数据的缓存，并适应不断变化的热点数据，提高了局部性。

在混合日志中，逻辑地址空间被分成 3 个连续的区域：（1）稳定区域（2）只读区域和（3）可变区域，如图 5 所示。稳定区域部分在磁盘上。内存中的部分由只读区域和可变区域组成。可变区域中的记录可以就地修改，而只读区域中的记录不能。为了更新当前只读区域中的记录，遵循 Read-Copy-Update（RCU）策略：在可变区域中创建新的副本，然后进行更新。对这种记录的进一步更新就地执行，只要它停留在可变区域中。

追加日志分配器上实现了 HybridLog，使用了一个额外的标记称为只读偏移量（read-only offset），它对应于驻留在内存循环缓冲区中的逻辑地址。在 head offset 和 read-only offset 之间的区域为只读区域，在 read-only offset 之后的区域为可变区域。如果一个记录的逻辑地址大于read-only offset，它将被就地更新。如果地址在 read-only offset 和 head offset 之间，我们在日志尾部创建一个更新的副本，并更新哈希索引以指向新的位置；如果地址小于 head-offset，意味着它不在内存中，因此发出一个异步 I/O 请求从磁盘中检索记录。一旦获得，在尾部创建新的更新副本，然后更新哈希索引。表 1 总结了这一更新方案。

read-only offset 与 tail offset 保持恒定的滞后，并且仅在与 head-offset 类似的跨域页面边界处更新。由于逻辑地址小于 read-only offset 的页都不会并发更新，因此将它们刷新到磁盘是安全的。随着 tail offset 的增加，read-only offset 也随之移动，使页面随时可以刷新。一旦它们被安全地存储到磁盘上，就可以使用 head offset 和关闭状态数组（像第 5 节）将它们从循环缓冲区中逐出。因此，read-only offset 充当准备刷新到磁盘的页面的轻量指示器。请注意，混合日志中的 read-only offset 允许对可变区域中的记录进行无锁访问，而在传统设计中，记录（或页面）必须在更新之前固定在缓冲池中，以防止在将它们刷新到磁盘时进行并发更新。

read-only offset 和 tail offset 之间的滞后决定了将内存缓冲区容量划分为快速的就地可更新（fast in-place updatable）和不可变的只读（immutable read-only）区域。除了帮助将页面安全地刷新到磁盘中，read-only 区域还充当一个二级缓存。在第 6.4 节中，我们讨论了缓存行为和混合日志区域大小对 Faster 性能的影响。

Lost-Update 异常

read-only offset 被自动更新和读取。但是，线程仍然可能基于偏移量的旧值来决定更新方案，从而导致不正确的执行。考虑图 6 所示的场景，来自我们的计数存储示例。线程 T1 和 T3 从 Faster 哈希索引中获得相同的逻辑地址。T1 决定进行就地更新，因为 L 大于当前只读偏移 R1。同时，由于 tail offset 的移动，线程 T2 更新了从 R1 到 R2 的 read-only offset。现在，线程 T3 将 L 与 R2 进行比较，并决定在 L' 创建一个新记录，更新值为 5。但是，线程 T1 在 L 时将该值更新为5，以后的所有访问都将使用 L' 时的值，因此我们丢失了 T1 的更新。（下表应为 L > R1）

（1）键值对 (k, v) 位于只读偏移量 R1 后，偏移量记为 L。
（2）线程 T1 和线程 T3 同时获取了 L。
（3）线程 T1 比较 L > R1，准备执行 in-place 更新。
（4）线程 T2 执行了其他更新操作，将尾偏移量往后递增，并将只读偏移量递增到 R2。
（5）线程 T3 比较 L< R3，执行 RCU，拷贝并追加了 (k, v + 1)。
（6）线程 T1 开始执行就地更新，将 L 上的 (k, v) -> (k, v + 1)，但实际上应为 (k, v + 2)，即更新丢失。

发生上述异常是因为线程 T2 更新了 read-only offset，而 T1 基于当前值执行操作（T2的修改对其他线程不可见）。可以通过在 T1 操作的整个过程中获取 read-only offset 上的读锁来防止这种情况的发生。然而，这样的锁方案是昂贵的，并且不必要地延迟了 read-only offset 的移动，而 read-only offset 的移动对于维护循环缓冲区是不可或缺的。另一方面，即使 read-only offset 发生了偏移，也会出现异常，因为一个线程（T1）基于旧值做出更新决定，而另一个线程（T2）基于新的偏移量值做出更新决定。

使用另一个称为安全只读偏移量的标记来消除这种不正确的执行。该标记跟踪所有线程看到的 read-only offset。它是基于以下不变式设计的：安全的 read-only offset 是任何活跃的 Faster 线程看到的最小的 read-only offset。使用如下的 epoch 触发操作机制来维护它：每当 read-only offset 被更新时，将当前 epoch 与一个触发操作结合起来，该触发操作将安全的 read-only offset 更新为新值。这种基于 epoch 的安全的 read-only offset 更新满足不变式，因为所有跨越当前 epoch 的线程都必须看到 read-only offset 的新值。

使用附加标记，安全的 read-only offset，HybridLog 被分成 4 个区域。我们将**安全的 read-only offset 和 read-only offset 之间的区域称为模糊区域（fuzzy region），因为一些线程可能会在 read-only offset 之后看到它，而另一些线程可能会在之前看到它。**只有当线程刷新它们的 epoch 时，它们才能保证获得安全的 read-only offset 和 read-only offset 的最新值。因此，每个线程可能都有这些标记的线程本地视图，如图 7 所示。

蓝色框框是每个线程所看到的 Fuzzy region，蓝框上面是每个线程自身视角的安全read-only offset，蓝框下面是每个线程自身视角的 read-only offset。全局的安全 read-only offset，取的是每个线程视角中最小值的 read-only offset。

线程 T4 具有最高的 read-only offset，因为它最近刷新了其 epoch，而线程 T3 是过时的值，因为它最近没有刷新。但是，请注意，任何线程的安全的 read-only offset 是最小 read-only offset，这是由我们的 epoch protection 框架确保的。当记录的逻辑地址小于安全 read-only 时，线程可能会尝试并发地创建一个新记录（因为在只读区域），由于对 Faster 哈希索引的原子 CAS 操作，所以只有一个记录会成功。

上述过程通过触发 epoch action 让该偏移量对所有线程可见：

• 它追踪所有线程可见的read-onlyoffset，取其中最小的。
• 使用epoch保护框架进行更新：

当只读偏移量更新时，调用 BumpEpoch(Action) 注册回调，更新安全只读偏移量到新的值。

其他线程若跨越了上面调用时的 epoch，则会触发安全只读偏移量的刷新，那么安全只读偏移量就可见了。

• 当键值对地址小于"安全只读偏移量"，则需要执行RCU更新。

Hlog及涉及到 Epoch 保护框架的地方：

typedef PersistentMemoryMalloc<disk_t> hlog_t;
hlog_t hlog;
 
class PersistentMemoryMalloc {
 public:
  typedef D disk_t;
  typedef typename D::file_t file_t;
  typedef typename D::log_file_t log_file_t;
  typedef PersistentMemoryMalloc<disk_t> alloc_t;
 
  log_file_t* file;
 
......
  // 这三个函数内部会掉异步写写入 log 信息
  AsyncFlushPages();
  AsyncFlushPagesToFile();
  AsyncFlushPage();
}