解析 RCU（Read-Copy-Update）模式：在 C++ 中实现高频读取场景的极致性能

各位同行，各位对高性能并发编程充满热情的开发者们，下午好。

今天，我们将深入探讨一种在并发编程领域具有卓越性能提升潜力的模式——RCU（Read-Copy-Update）。特别是在 C++ 环境中，面对高频读取、低频写入的场景，RCU 能够帮助我们突破传统锁机制的性能瓶颈，实现极致的读取性能。我们将从 RCU 的基本原理出发，逐步深入到其在 C++ 中的实现细节，包括内存管理、原子操作、内存屏障以及核心的“宽限期”机制，并最终通过一个具体的 C++ 代码示例来演示其工作方式。

1. 传统并发控制的困境与 RCU 的崛起

在现代多核处理器架构下，构建高性能的并发系统是软件开发面临的核心挑战之一。当多个线程需要访问和修改共享数据时，我们必须引入同步机制来保证数据的一致性和正确性。

最常见的同步机制是互斥锁（std::mutex、pthread_mutex）或自旋锁。它们通过强制一次只有一个线程访问临界区来避免数据竞争。然而，锁机制在带来正确性的同时，也引入了显著的性能开销：

• 锁竞争（Contention）：当多个线程争抢同一个锁时，未获得锁的线程会被阻塞，导致上下文切换，这带来了巨大的性能损失。
• 缓存失效（Cache Invalidation）：锁的获取和释放通常涉及对共享状态的修改，这会导致不同 CPU 核心之间的缓存线失效和同步，进一步降低性能。
• 死锁（Deadlock）：不当的锁使用可能导致死锁，使系统完全停滞。

对于读多写少的场景，例如配置信息、路由表、DNS 缓存、文件系统目录等，传统锁机制的劣势尤为明显。即使是读取操作，也常常需要获取读写锁（std::shared_mutex），虽然允许多个读者并发，但写者的出现仍然会阻塞所有读者和写者，且读写锁本身的开销也并非为零。

为了解决这些问题，研究人员提出了多种无锁（lock-free）或读无锁（wait-free for readers）的并发数据结构和模式。RCU（Read-Copy-Update）便是其中一种强大且广泛应用于高性能系统（特别是 Linux 内核）的模式，它专门针对读多写少场景进行了优化，旨在提供近乎无开销的读取操作。

RCU 的核心思想是：读者不阻塞写者，写者不阻塞读者，且读者之间完全并行，无需任何同步开销。 写者通过“复制-修改-更新”的策略来操作数据，并利用一个“宽限期”机制来安全地回收旧数据。

2. RCU 的核心原理：读、复制、更新与宽限期

RCU 这个名字本身就揭示了其工作机制的三个关键步骤：

1. Read (读)：读者直接访问共享数据，无需获取锁。这是 RCU 性能优越性的根本来源。读者可能会看到旧版本的数据，但在其读取过程中，数据的一致性是由写者的原子更新和宽限期机制保证的。
2. Copy (复制)：当写者需要修改数据时，它不会直接修改原始数据。相反，它会先创建一个原始数据的副本，然后在副本上进行修改。
3. Update (更新)：修改完成后，写者会使用一个原子操作（通常是原子指针交换）将指向原始数据的指针更新为指向修改后的新数据副本的指针。

这个过程完成后，系统中就存在了两个版本的数据：一个是由新指针指向的新版本，另一个是旧指针曾经指向的旧版本。此时，关键问题出现了：何时才能安全地释放旧版本的数据？ 如果立即释放，那些可能正在读取旧数据的线程就会访问到无效内存，导致程序崩溃。

这就是 RCU 中最重要的概念——宽限期（Grace Period）。

2.1 宽限期（Grace Period）

宽限期的目的是确保在旧数据被释放之前，所有可能正在访问它的读者都已经完成了对旧数据的访问，并切换到访问新数据。换句话说，宽限期是从写者完成原子更新（将新数据暴露给系统）开始，到所有正在访问旧数据的读者都“退出”其临界区（或者说，不再持有对旧数据的引用）为止的这段时间。

一旦宽限期结束，写者就可以安全地回收旧数据所占用的内存。

理解宽限期的关键在于：

• 读者不阻塞写者：写者完成更新后立即返回，无需等待读者。
• 写者不阻塞读者：读者在宽限期内可以继续读取旧数据。
• 内存安全：宽限期保证了在旧数据被释放时，没有活跃的读者会访问它。

2.2 RCU 的内存模型与原子性

RCU 的正确性高度依赖于 C++ 内存模型和原子操作。特别是在多核处理器上，编译器和 CPU 都可能对指令进行重排序，这可能破坏 RCU 的逻辑。因此，我们需要使用 std::atomic 和内存屏障（memory barriers）来强制特定的内存访问顺序。

写者侧的内存屏障：

1. 修改数据副本：在副本上进行修改，这通常是常规写操作。
2. std::atomic_store_explicit(ptr, new_data, std::memory_order_release)：将指向新数据的指针原子地存储到共享指针变量中。memory_order_release 确保所有在存储操作之前对新数据的修改都对其他线程可见。这会阻止编译器和 CPU 将对新数据的修改重排序到指针存储之后。

读者侧的内存屏障：

1. std::atomic_load_explicit(ptr, std::memory_order_acquire)：原子地加载共享指针变量的值。memory_order_acquire 确保所有在加载操作之后对数据的访问都能够看到在存储操作（memory_order_release）之前完成的写操作。这会阻止编译器和 CPU 将对数据的访问重排序到指针加载之前。

通过 memory_order_release 和 memory_order_acquire 的配对使用，我们创建了一个“同步点”，确保写者对新数据的修改在读者看到新指针时是完全可见的。

2.3 RCU 的性能优势

RCU 在高频读取场景下性能卓越，主要体现在以下几个方面：

• 无锁读取：读者无需获取任何锁，直接访问数据。这消除了锁竞争、上下文切换和缓存失效等开销。
• 高并发性：理论上，任意数量的读者都可以同时并行读取数据，不会相互影响。
• 低开销：读取路径上几乎没有额外的指令开销，仅涉及一个指针解引用。
• 缓存友好：读者访问的数据通常是稳定的，有助于 CPU 缓存命中。

然而，RCU 并非没有代价。写者的开销相对较高：它需要复制数据、进行修改，然后执行原子指针交换，并管理宽限期。因此，RCU 最适合于读写比例极高的场景。

3. C++ 中 RCU 的实现挑战与策略

在 C++ 用户空间实现 RCU，最大的挑战是如何安全地实现宽限期和内存回收。Linux 内核提供了 call_rcu() 等机制，但用户空间没有内核的协作环境。我们通常需要自行设计宽限期管理系统。

常见的用户空间 RCU 宽限期实现策略包括：

1. 基于 Quiescent State（静默状态）的 RCU (QSBR)：

   • 每个读者线程在进入 RCU 保护的临界区时，会标记自己处于“活跃”状态。
   • 在离开临界区时，它会标记自己处于“静默”状态。
   • 写者在更新数据后，会等待所有线程都进入一次静默状态（即，所有读者都至少一次离开了 RCU 临界区）。
   • 这种方式的缺点是需要读者主动报告状态，且需要一个机制来协调所有线程的状态。

2. 基于 Epoch（纪元）的 RCU：

   • 系统维护一个全局的纪元计数器。
   • 每个读者线程在进入 RCU 临界区时，会记录当前的全局纪元。
   • 写者在更新数据后，会等待足够长的时间，或者等待所有记录了旧纪元的读者都推进到新的纪元。
   • 这种方式比 QSBR 更为自动化，但仍然需要读者在某些点更新其本地纪元。

3. Hazard Pointers（危险指针）：

   • 每个读者线程维护一个或多个“危险指针”槽位。
   • 当读者要访问一个 RCU 保护的数据结构中的某个节点时，它会将该节点的指针地址原子地写入自己的危险指针槽位。
   • 写者在更新数据并想要回收旧数据时，会检查所有线程的危险指针槽位。如果一个旧数据节点的地址仍然被任何危险指针指向，则说明该节点可能仍在被某个读者使用，不能立即回收。写者会将这些“危险”的旧数据节点放入一个延迟回收列表。
   • 当一个读者不再需要访问某个节点时，它会清空对应的危险指针槽位。
   • 写者或一个独立的回收线程会定期扫描延迟回收列表和所有危险指针槽位，回收那些不再被任何危险指针指向的节点。

在 C++ 用户空间，Hazard Pointers 是实现 RCU 内存回收相对常见且直观的一种机制。 它将内存回收问题转化为一个垃圾收集问题，由危险指针系统负责判断何时可以安全释放内存。下面我们将重点介绍基于 Hazard Pointers 的 RCU 实现。

4. 基于 Hazard Pointers 的 C++ RCU 实现

我们将构建一个简单的 RCU 容器，它持有一个指向数据的指针。读者通过这个容器获取数据，写者通过它更新数据。

4.1 核心组件设计

1. Data 结构：我们希望 RCU 保护的实际数据类型。
2. HazardPointer 类：表示一个危险指针，它是一个线程局部的槽位，用于存储读者正在访问的对象的地址。
3. HazardPointerManager 类：管理所有线程的危险指针，并负责将旧对象加入回收列表，以及在宽限期结束后安全地回收它们。
4. RcuGuardedPtr<T> 类：用户面对的 RCU 容器，提供 get() 方法供读者使用，以及 update() 方法供写者使用。

4.2 HazardPointer 和 HazardPointerManager

HazardPointer 类：
每个线程可能需要多个危险指针来保护在链表遍历等操作中同时访问的多个节点。这里我们简化为每个线程一个危险指针。

#include <atomic>
#include <thread>
#include <vector>
#include <map>
#include <set>
#include <mutex>
#include <functional>
#include <chrono>
#include <iostream>
#include <memory> // For std::unique_ptr

// 前向声明， HazardPointerManager 需要知道 RcuGuardedPtr 的存在
template<typename T> class RcuGuardedPtr;

// HazardPointer 定义
// 每个线程可能需要多个危险指针槽位来保护多个对象
// 这里我们实现一个简化的版本，每个线程维护一个危险指针槽位
class HazardPointer
{
public:
    // 默认构造函数，初始化为 nullptr
    HazardPointer() : ptr_(nullptr) {}

    // 设置危险指针
    void set(void* p) { ptr_.store(p, std::memory_order_release); }

    // 获取危险指针的值
    void* get() const { return ptr_.load(std::memory_order_acquire); }

    // 清空危险指针
    void clear() { ptr_.store(nullptr, std::memory_order_release); }

private:
    std::atomic<void*> ptr_;
};

// 线程局部存储，用于每个线程的 HazardPointer 实例
// 在实际应用中，可能需要一个 HazardPointer 数组
thread_local HazardPointer g_thread_hazard_pointer;

// HazardPointerManager 类
// 负责管理所有线程的危险指针，并协调内存回收
class HazardPointerManager
{
public:
    // 注册一个 HazardPointer。在我们的简化模型中，g_thread_hazard_pointer 已经全局可见
    // 但如果 HazardPointer 是动态分配的，则需要注册。
    // 这里我们直接通过 g_thread_hazard_pointer 访问。
    // 实际上，更健壮的管理器会维护一个所有活跃 HazardPointer 的列表。
    // 为了简化，我们假设只有一个全局的 thread_local HazardPointer。

    // 线程局部存储，用于每个线程的延迟回收列表
    // 每个线程有自己的待回收对象列表，避免锁竞争
    struct RetiredObject
    {
        void*                 ptr;
        std::function<void(void*)> deleter;
        long long             retired_epoch; // 记录被标记为待回收时的纪元
    };
    thread_local std::vector<RetiredObject> g_thread_retired_objects;

    // 全局原子纪元计数器，用于宽限期管理
    // 当写者更新数据时，会增加此计数器
    // 当回收器回收时，会等待所有读者都通过这个纪元
    static std::atomic<long long> global_epoch;

    HazardPointerManager() = default;
    ~HazardPointerManager() = default;

    // 将旧对象添加到当前线程的延迟回收列表
    // 对象的实际删除将在宽限期结束后进行
    void retire_object(void* obj_ptr, std::function<void(void*)> deleter)
    {
        g_thread_retired_objects.push_back({obj_ptr, deleter, global_epoch.load(std::memory_order_relaxed)});
        // 尝试触发回收，但不必每次都回收
        // 实际的回收逻辑可能由一个独立的线程或定期触发
        // 为了演示，我们在这里尝试扫描和回收
        try_scan_and_reclaim();
    }

    // 扫描所有线程的危险指针，并回收不再危险的对象
    void try_scan_and_reclaim()
    {
        // 实际的扫描和回收逻辑可能比较复杂，并且需要考虑性能。
        // 理想情况下，会有一个专门的回收线程，或者在写者操作后异步触发。
        // 为了简化，我们在这里直接执行同步扫描。

        // 步骤 1: 收集所有活跃的危险指针
        // 在我们的简化模型中，只有 g_thread_hazard_pointer 是活跃的。
        // 在实际系统中，可能需要遍历所有注册的 HazardPointer 实例。
        std::set<void*> active_hazard_ptrs;

        // 注意：这里需要遍历所有正在运行的线程的 thread_local g_thread_hazard_pointer。
        // C++ 标准库没有直接暴露所有 thread_local 变量的实例。
        // 在真实的高性能库中，通常会有一个注册机制，例如每个线程启动时向管理器注册其 HazardPointer 实例。
        // 为了本示例的演示目的，我们假设可以“看到”所有活跃的 HazardPointer。
        // 实际上，这里需要更复杂的机制，例如，管理器维护一个 `std::vector<HazardPointer*>`。
        // 简化处理：假设只有一个活跃线程，或者说，我们只能检查当前线程的危险指针。
        // 这在多线程环境下是不够的，因为它无法看到其他线程的危险指针。
        // 为了模拟，我们需要一个全局的列表来注册所有的 HazardPointer
        // 例如：
        static std::vector<HazardPointer*> all_hazard_ptrs;
        static std::mutex all_hazard_ptrs_mutex; // 保护 all_hazard_ptrs

        // 这是为了模拟如何收集所有线程的 Hazard Pointers。
        // 实际应用中，线程会在启动时注册其 HazardPointer 实例。
        {
            std::lock_guard<std::mutex> lock(all_hazard_ptrs_mutex);
            // 假设 g_thread_hazard_pointer 已经被正确注册到 all_hazard_ptrs
            // 每次调用 try_scan_and_reclaim 时，收集所有注册的危险指针
            for (HazardPointer* hp_ptr : all_hazard_ptrs) {
                void* p = hp_ptr->get();
                if (p != nullptr) {
                    active_hazard_ptrs.insert(p);
                }
            }
        }

        // 如果没有注册的 HazardPointer，此简化版本将无法正确工作。
        // 我们需要一个更健壮的注册/注销机制。
        // 例如，我们可以让 `RcuGuardedPtr` 在其构造函数中注册一个 `HazardPointer`
        // 或者让 `HazardPointerManager` 提供一个 `register_thread_hazard_pointer` 方法。

        // 步骤 2: 遍历当前线程的待回收列表
        std::vector<RetiredObject> new_retired_objects;
        for (const auto& obj : g_thread_retired_objects)
        {
            // 检查当前对象是否在任何活跃的危险指针列表中
            if (active_hazard_ptrs.count(obj.ptr) == 0)
            {
                // 如果不在，且所有读者都通过了该对象被淘汰时的纪元，则可以安全回收
                // 这里的 epoch 检查是 Hazard Pointer 和 Epoch RCU 的结合思想
                // 实际上，纯 Hazard Pointer 只依赖 active_hazard_ptrs
                // 加上 epoch 是为了在某些情况下提供额外的宽限期保证
                // 如果 `global_epoch` 已经显著超过 `obj.retired_epoch`，则可以安全删除
                // 通常，我们会等待 `global_epoch - obj.retired_epoch >= 2`
                // 确保所有读者都已看到新的 epoch
                if (global_epoch.load(std::memory_order_acquire) - obj.retired_epoch >= 2) {
                    obj.deleter(obj.ptr); // 执行删除
                } else {
                    new_retired_objects.push_back(obj); // 暂时保留，等待下一个回收周期
                }
            }
            else
            {
                new_retired_objects.push_back(obj); // 仍然被危险指针指向，保留
            }
        }
        g_thread_retired_objects = std::move(new_retired_objects);
    }

    // 注册线程的 HazardPointer 实例
    // 这是一个关键的补充，用于管理所有线程的 HazardPointer
    static void register_thread_hazard_pointer(HazardPointer* hp) {
        std::lock_guard<std::mutex> lock(all_hazard_ptrs_mutex);
        all_hazard_ptrs.push_back(hp);
    }

    // 注销线程的 HazardPointer 实例 (例如，线程退出时)
    static void unregister_thread_hazard_pointer(HazardPointer* hp) {
        std::lock_guard<std::mutex> lock(all_hazard_ptrs_mutex);
        all_hazard_ptrs.erase(std::remove(all_hazard_ptrs.begin(), all_hazard_ptrs.end(), hp), all_hazard_ptrs.end());
    }
};

// 静态成员变量的定义
std::atomic<long long> HazardPointerManager::global_epoch = 0;
// 静态成员变量的定义
std::vector<HazardPointer*> HazardPointerManager::all_hazard_ptrs;
std::mutex HazardPointerManager::all_hazard_ptrs_mutex;

// 辅助类：RAII 风格的 HazardPointer 管理器
// 确保在线程生命周期内正确注册和注销 HazardPointer
class ThreadHazardPointerRegistrar {
public:
    ThreadHazardPointerRegistrar() {
        HazardPointerManager::register_thread_hazard_pointer(&g_thread_hazard_pointer);
    }
    ~ThreadHazardPointerRegistrar() {
        // 在线程退出时，需要清除该线程的危险指针，并尝试回收。
        // 这需要更复杂的机制来确保所有线程的回收列表都被处理。
        // 简化起见，这里只注销。实际中，退出线程的延迟回收列表也需要被处理。
        g_thread_hazard_pointer.clear();
        HazardPointerManager::unregister_thread_hazard_pointer(&g_thread_hazard_pointer);
    }
};

// 每个线程一个注册器，确保 HazardPointer 的生命周期管理
thread_local ThreadHazardPointerRegistrar g_thread_hazard_pointer_registrar;

解释：

• HazardPointer：每个线程有一个 thread_local 的 HazardPointer 实例 g_thread_hazard_pointer。set 和 get 方法使用 std::atomic 来保证对指针值的原子读写，并使用 memory_order_release 和 memory_order_acquire 来确保内存可见性。
• HazardPointerManager：
  • g_thread_retired_objects：每个线程维护一个自己的待回收对象列表。这样可以减少线程间对一个全局列表的锁竞争。
  • global_epoch：一个全局的原子计数器，写者每次更新数据时会递增它。这有助于在回收时判断一个对象是否已经“足够旧”，即使它不被任何危险指针指向，也可以增加一个安全延迟。
  • retire_object：将旧对象及其删除器添加到当前线程的待回收列表。
  • try_scan_and_reclaim：这是核心的回收逻辑。
    1. 收集活跃危险指针：遍历所有注册的 HazardPointer 实例，将它们当前指向的地址收集到一个 std::set 中，以便快速查找。（注意：这里是简化处理，实际需要更复杂的全局注册机制，我们通过 register_thread_hazard_pointer 和 all_hazard_ptrs 尝试模拟）
    2. 遍历待回收列表：检查当前线程的 g_thread_retired_objects 中的每个对象。
    3. 判断并回收：如果一个对象不再被任何活跃的危险指针指向，并且其 retired_epoch 已经足够老（例如，global_epoch - obj.retired_epoch >= 2，这意味着至少两个完整的纪元已经过去，保证了所有读者都已越过该点），那么就可以安全地通过其 deleter 函数释放该对象。否则，将其保留到 new_retired_objects 中，等待下一次回收扫描。
• ThreadHazardPointerRegistrar：这是一个 RAII 类，利用 thread_local 变量在每个线程启动时自动注册其 HazardPointer 到 HazardPointerManager，并在线程退出时注销。这是为了解决 HazardPointerManager 需要“看到”所有线程的危险指针的问题。

4.3 RcuGuardedPtr<T>：RCU 保护的指针

这是用户直接交互的类，它包装了一个 std::atomic<T*>。

template<typename T>
class RcuGuardedPtr
{
public:
    RcuGuardedPtr(T* initial_data = nullptr) : ptr_(initial_data) {
        // 确保 HazardPointerManager 已经初始化，并且线程的 HazardPointer 已经注册
    }

    // 读者获取数据的方法
    // 返回一个 RAII 包装器，确保 HazardPointer 的正确设置和清除
    class RcuReadGuard
    {
    public:
        RcuReadGuard(const RcuGuardedPtr<T>& rcu_ptr) : rcu_ptr_(rcu_ptr)
        {
            // 循环直到我们获得一个稳定的指针
            // 在获得 ptr_ 后，立即将其设置为危险指针
            // 然后再次检查 rcu_ptr_ 的值是否仍然相同
            // 如果不相同，说明在设置危险指针期间发生了更新，需要重试
            T* current_ptr;
            do {
                g_thread_hazard_pointer.clear(); // 清除旧的危险指针
                current_ptr = rcu_ptr_.ptr_.load(std::memory_order_acquire); // 获取当前指针
                g_thread_hazard_pointer.set(current_ptr); // 设置危险指针
            } while (g_thread_hazard_pointer.get() != rcu_ptr_.ptr_.load(std::memory_order_acquire)); // 再次检查是否稳定

            // 现在 current_ptr 是稳定的，并且被危险指针保护
            // 如果 current_ptr 是 nullptr，则不需要设置危险指针，但上述循环已处理
            guarded_data_ = current_ptr;
        }

        ~RcuReadGuard()
        {
            // 清除当前线程的危险指针
            g_thread_hazard_pointer.clear();
            // 在某些实现中，这里可以触发一次回收扫描，但通常由写者或专门的回收线程负责。
            // HazardPointerManager::get_instance().try_scan_and_reclaim();
        }

        // 访问数据
        T* get() const { return guarded_data_; }
        T& operator*() const { return *guarded_data_; }
        T* operator->() const { return guarded_data_; }

    private:
        const RcuGuardedPtr<T>& rcu_ptr_;
        T* guarded_data_; // 实际保护的数据指针
    };

    // 读者接口
    RcuReadGuard read() const {
        return RcuReadGuard(*this);
    }

    // 写者更新数据的方法
    // old_data_to_retire 是旧的数据指针，需要被回收
    void update(T* new_data)
    {
        HazardPointerManager::global_epoch.fetch_add(1, std::memory_order_release); // 增加纪元
        T* old_data = ptr_.exchange(new_data, std::memory_order_release); // 原子交换指针

        if (old_data != nullptr)
        {
            // 将旧数据添加到回收队列
            // 使用 lambda 捕获 deleter，以确保正确释放内存
            HazardPointerManager().retire_object(old_data, [](void* p){ delete static_cast<T*>(p); });
        }
    }

    // 显式删除当前持有的数据，用于清理
    void clear() {
        T* old_data = ptr_.exchange(nullptr, std::memory_order_release);
        if (old_data != nullptr) {
            HazardPointerManager().retire_object(old_data, [](void* p){ delete static_cast<T*>(p); });
        }
    }

private:
    std::atomic<T*> ptr_;
};

解释：

• RcuGuardedPtr：

  • ptr_：一个 std::atomic<T*> 成员，这是 RCU 保护的实际指针。
  • update(T* new_data)：写者方法。
    1. HazardPointerManager::global_epoch.fetch_add(1, std::memory_order_release)：在更新前，递增全局纪元计数器。这有助于宽限期机制判断旧对象是否“足够老”。
    2. ptr_.exchange(new_data, std::memory_order_release)：原子地将 ptr_ 更新为 new_data，并返回旧的指针。memory_order_release 确保 new_data 的所有修改在指针更新前完成。
    3. HazardPointerManager().retire_object(...)：将旧指针（即 old_data）交给 HazardPointerManager 进行延迟回收。

• RcuReadGuard：这是一个嵌套的 RAII 类，用于管理读者对 RCU 数据的访问。

  • 构造函数：
    1. 它首先清除当前线程的危险指针 (g_thread_hazard_pointer.clear())。
    2. 然后加载 RCU 容器的当前指针 (rcu_ptr_.ptr_.load(std::memory_order_acquire))。
    3. 将加载到的指针设置为危险指针 (g_thread_hazard_pointer.set(current_ptr))。
    4. 关键的循环检查：再次加载 RCU 容器的指针，并与危险指针进行比较。如果两者不一致，说明在设置危险指针的瞬间，RCU 容器的指针被写者更新了。在这种情况下，当前线程可能正在访问一个即将被回收的指针，因此需要重新尝试（do-while 循环）。这个循环确保读者总是访问一个被其危险指针正确保护的、稳定的指针。
    5. memory_order_acquire 确保在加载指针后，对数据内容的访问能看到写者在 memory_order_release 之前完成的所有修改。
  • 析构函数：在 RcuReadGuard 离开作用域时，自动清除当前线程的危险指针 (g_thread_hazard_pointer.clear())。

4.4 完整代码示例

#include <atomic>
#include <thread>
#include <vector>
#include <map>
#include <set>
#include <mutex>
#include <functional>
#include <chrono>
#include <iostream>
#include <memory>
#include <algorithm> // For std::remove

// 前向声明
template<typename T> class RcuGuardedPtr;

// HazardPointer 定义
class HazardPointer
{
public:
    HazardPointer() : ptr_(nullptr) {}
    void set(void* p) { ptr_.store(p, std::memory_order_release); }
    void* get() const { return ptr_.load(std::memory_order_acquire); }
    void clear() { ptr_.store(nullptr, std::memory_order_release); }
private:
    std::atomic<void*> ptr_;
};

// 线程局部存储的 HazardPointer 实例
thread_local HazardPointer g_thread_hazard_pointer;

// HazardPointerManager 类
class HazardPointerManager
{
public:
    struct RetiredObject
    {
        void*                 ptr;
        std::function<void(void*)> deleter;
        long long             retired_epoch;
    };
    thread_local static std::vector<RetiredObject> g_thread_retired_objects; // 线程局部静态成员

    static std::atomic<long long> global_epoch;
    static std::vector<HazardPointer*> all_hazard_ptrs; // 全局注册的 HazardPointer 列表
    static std::mutex all_hazard_ptrs_mutex; // 保护 all_hazard_ptrs

    HazardPointerManager() = default;
    ~HazardPointerManager() = default;

    static void retire_object(void* obj_ptr, std::function<void(void*)> deleter)
    {
        g_thread_retired_objects.push_back({obj_ptr, deleter, global_epoch.load(std::memory_order_relaxed)});
        // 每次 retire_object 都尝试扫描回收，但可以优化为批量回收
        // 比如当 retired_objects 数量达到阈值时
        try_scan_and_reclaim();
    }

    static void try_scan_and_reclaim()
    {
        // 收集所有活跃的危险指针
        std::set<void*> active_hazard_ptrs;
        {
            std::lock_guard<std::mutex> lock(all_hazard_ptrs_mutex);
            for (HazardPointer* hp_ptr : all_hazard_ptrs) {
                void* p = hp_ptr->get();
                if (p != nullptr) {
                    active_hazard_ptrs.insert(p);
                }
            }
        }

        // 遍历当前线程的待回收列表
        std::vector<RetiredObject> new_retired_objects;
        for (const auto& obj : g_thread_retired_objects)
        {
            if (active_hazard_ptrs.count(obj.ptr) == 0)
            {
                // 如果对象不再被任何危险指针指向，且纪元足够老，则可以回收
                // 这里的 epoch 检查是为了更保守地确保宽限期
                // 通常，等待 global_epoch 至少比 retired_epoch 大2，以确保所有读者都已通过该纪元
                if (global_epoch.load(std::memory_order_acquire) - obj.retired_epoch >= 2) {
                    obj.deleter(obj.ptr);
                } else {
                    new_retired_objects.push_back(obj);
                }
            }
            else
            {
                new_retired_objects.push_back(obj);
            }
        }
        g_thread_retired_objects = std::move(new_retired_objects);
    }

    static void register_thread_hazard_pointer(HazardPointer* hp) {
        std::lock_guard<std::mutex> lock(all_hazard_ptrs_mutex);
        all_hazard_ptrs.push_back(hp);
    }

    static void unregister_thread_hazard_pointer(HazardPointer* hp) {
        std::lock_guard<std::mutex> lock(all_hazard_ptrs_mutex);
        all_hazard_ptrs.erase(std::remove(all_hazard_ptrs.begin(), all_hazard_ptrs.end(), hp), all_hazard_ptrs.end());
    }
};

// 静态成员变量的定义和初始化
std::atomic<long long> HazardPointerManager::global_epoch = 0;
std::vector<HazardPointer*> HazardPointerManager::all_hazard_ptrs;
std::mutex HazardPointerManager::all_hazard_ptrs_mutex;
thread_local std::vector<HazardPointerManager::RetiredObject> HazardPointerManager::g_thread_retired_objects;

// RAII 风格的 HazardPointer 注册器
class ThreadHazardPointerRegistrar {
public:
    ThreadHazardPointerRegistrar() {
        HazardPointerManager::register_thread_hazard_pointer(&g_thread_hazard_pointer);
    }
    ~ThreadHazardPointerRegistrar() {
        // 在线程退出时，需要清理该线程的危险指针和待回收列表
        // 确保本线程的待回收对象也被扫描和处理
        g_thread_hazard_pointer.clear();
        HazardPointerManager::unregister_thread_hazard_pointer(&g_thread_hazard_pointer);
        // 在实际应用中，线程退出时，其 g_thread_retired_objects 里的对象也需要被回收。
        // 这可能需要主线程或某个回收线程来接管这些待回收对象。
        // 这里为了简化，我们假设这些对象在线程生命周期内会被处理。
        // 否则，未回收的对象将导致内存泄漏。
        HazardPointerManager::try_scan_and_reclaim(); // 尝试回收当前线程的剩余对象
    }
};

// 每个线程一个注册器，确保 HazardPointer 的生命周期管理
thread_local ThreadHazardPointerRegistrar g_thread_hazard_pointer_registrar;

// RcuGuardedPtr 类
template<typename T>
class RcuGuardedPtr
{
public:
    RcuGuardedPtr(T* initial_data = nullptr) : ptr_(initial_data) {}

    // 读者获取数据的方法
    class RcuReadGuard
    {
    public:
        RcuReadGuard(const RcuGuardedPtr<T>& rcu_ptr) : rcu_ptr_(rcu_ptr)
        {
            T* current_ptr;
            do {
                g_thread_hazard_pointer.clear();
                current_ptr = rcu_ptr_.ptr_.load(std::memory_order_acquire);
                g_thread_hazard_pointer.set(current_ptr);
            } while (g_thread_hazard_pointer.get() != rcu_ptr_.ptr_.load(std::memory_order_acquire));

            guarded_data_ = current_ptr;
        }

        ~RcuReadGuard()
        {
            g_thread_hazard_pointer.clear();
        }

        T* get() const { return guarded_data_; }
        T& operator*() const { return *guarded_data_; }
        T* operator->() const { return guarded_data_; }

    private:
        const RcuGuardedPtr<T>& rcu_ptr_;
        T* guarded_data_;
    };

    RcuReadGuard read() const {
        return RcuReadGuard(*this);
    }

    void update(T* new_data)
    {
        HazardPointerManager::global_epoch.fetch_add(1, std::memory_order_release);
        T* old_data = ptr_.exchange(new_data, std::memory_order_release);

        if (old_data != nullptr)
        {
            HazardPointerManager::retire_object(old_data, [](void* p){ delete static_cast<T*>(p); });
        }
    }

    void clear() {
        T* old_data = ptr_.exchange(nullptr, std::memory_order_release);
        if (old_data != nullptr) {
            HazardPointerManager::retire_object(old_data, [](void* p){ delete static_cast<T*>(p); });
        }
    }

private:
    std::atomic<T*> ptr_;
};

// 示例数据结构
struct MyData {
    int value;
    std::string name;

    MyData(int v, const std::string& n) : value(v), name(n) {
        // std::cout << "MyData(" << value << ", " << name << ") constructed at " << this << std::endl;
    }
    ~MyData() {
        // std::cout << "MyData(" << value << ", " << name << ") destructed at " << this << std::endl;
    }
};

// 主函数和测试
int main() {
    std::cout << "RCU (Read-Copy-Update) Demo with Hazard Pointers in C++" << std::endl;

    RcuGuardedPtr<MyData> rcu_data(new MyData(100, "Initial Data"));

    std::atomic<bool> stop_threads(false);
    std::atomic<long long> read_count(0);
    std::atomic<long long> write_count(0);

    // 读者线程
    auto reader_func = [&](int id) {
        std::cout << "Reader " << id << " started." << std::endl;
        while (!stop_threads.load(std::memory_order_relaxed)) {
            RcuGuardedPtr<MyData>::RcuReadGuard guard = rcu_data.read();
            MyData* data = guard.get();
            if (data) {
                // 模拟读取操作
                // std::cout << "Reader " << id << ": Value = " << data->value << ", Name = " << data->name << std::endl;
                read_count.fetch_add(1, std::memory_order_relaxed);
            }
            std::this_thread::yield(); // 减少 CPU 占用，避免忙循环
        }
        std::cout << "Reader " << id << " stopped. Read count: " << read_count.load() << std::endl;
    };

    // 写者线程
    auto writer_func = [&](int id) {
        std::cout << "Writer " << id << " started." << std::endl;
        int current_value = 101;
        while (!stop_threads.load(std::memory_order_relaxed)) {
            // 模拟数据修改
            MyData* new_data = new MyData(current_value++, "Updated Data " + std::to_string(current_value));
            rcu_data.update(new_data);
            write_count.fetch_add(1, std::memory_order_relaxed);
            // std::cout << "Writer " << id << ": Updated data to value " << new_data->value << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(50)); // 模拟写操作间隔
        }
        std::cout << "Writer " << id << " stopped. Write count: " << write_count.load() << std::endl;
    };

    std::vector<std::thread> readers;
    for (int i = 0; i < 4; ++i) { // 4 个读者线程
        readers.emplace_back(reader_func, i);
    }

    std::thread writer(writer_func, 0); // 1 个写者线程

    // 运行一段时间
    std::this_thread::sleep_for(std::chrono::seconds(5));

    stop_threads.store(true, std::memory_order_relaxed);

    for (auto& t : readers) {
        t.join();
    }
    writer.join();

    rcu_data.clear(); // 清理最后的旧数据

    // 强制执行最后一次回收，以确保所有对象都被删除
    HazardPointerManager::try_scan_and_reclaim();
    HazardPointerManager::try_scan_and_reclaim(); // 多次调用以确保宽限期
    HazardPointerManager::try_scan_and_reclaim();

    std::cout << "nTotal reads: " << read_count.load() << std::endl;
    std::cout << "Total writes: " << write_count.load() << std::endl;
    std::cout << "RCU Demo Finished." << std::endl;

    return 0;
}

编译和运行：

使用 C++11 或更高版本编译，并启用多线程支持：
g++ -std=c++17 -O2 -pthread rcu_demo.cpp -o rcu_demo
./rcu_demo

这个示例展示了 RCU 的基本工作原理。读者线程通过 RcuReadGuard 安全地访问数据，而写者线程则通过 update 方法创建新数据并原子地交换指针。HazardPointerManager 负责在宽限期结束后回收旧数据。

5. RCU 的适用场景与进阶考量

5.1 RCU 的适用场景

RCU 最适合以下类型的应用：

• 读多写少：这是 RCU 的核心优势所在。如果写入操作非常频繁，复制数据和管理宽限期的开销可能会抵消读取的性能优势。
• 数据结构是可替换的：整个数据结构或其核心部分可以被原子地替换为一个新版本。例如，一个配置对象的指针、一个哈希表的根节点指针。
• 对读取延迟要求极高：例如实时系统、高吞吐量的网络服务等。
• 需要避免读者阻塞：即使写者正在修改数据，读者也不应被阻塞。

5.2 RCU 的局限性与挑战

尽管 RCU 强大，但它并非银弹：

• 实现复杂性：尤其是内存回收机制（宽限期管理）非常复杂且容易出错。
• 写者开销：写者需要复制整个数据结构（或其部分），这可能涉及显著的内存分配和复制成本。
• 内存膨胀：在宽限期内，旧版本和新版本的数据会同时存在，导致内存占用暂时增加。如果写操作非常频繁，并且宽限期较长，可能会导致大量的旧数据堆积，形成内存泄漏。
• 不适用于读写冲突频繁的场景：如果数据频繁地被修改且读者需要看到最新版本，或者写者之间也需要严格同步，RCU 可能不适用。

5.3 进阶优化与考虑

1. 细粒度 RCU：并非所有数据结构都需要整个复制。对于链表或树，可以只复制和修改受影响的节点，然后更新指针，这减少了写者开销。
2. 写者同步：虽然读者不阻塞，但写者之间通常仍然需要某种同步（例如，互斥锁）来避免同时修改和更新同一个数据结构，从而避免多个写者之间产生竞争性的 update。
3. 回收线程：在生产环境中，通常会有一个独立的后台线程来定期扫描 HazardPointerManager 的待回收列表并执行回收，而不是在每次 retire_object 时都同步执行。这可以平摊回收的开销。
4. 内存池/自定义分配器：频繁的 new/delete 可能导致堆碎片和性能开销。使用内存池可以显著改善性能。
5. RCU 与其他无锁技术结合：RCU 可以与其他无锁数据结构（如无锁队列）结合使用，构建更复杂的并发系统。
6. std::shared_ptr 与 RCU 的对比：
   • std::shared_ptr 通过引用计数来管理对象的生命周期。每次复制或赋值 shared_ptr 都会原子地修改引用计数。
   • 优点：使用简单，自动管理内存，对读者和写者都提供安全访问。
   • 缺点：引用计数的原子操作会带来性能开销，尤其是在高并发读取时，由于缓存线竞争，可能导致性能下降。这是 RCU 试图解决的核心问题。
   • 总结：RCU 适用于极致性能追求的读多写少场景，牺牲了写者便利性来换取读者零开销。shared_ptr 适用于一般并发场景，提供了更好的易用性和更通用的内存管理。它们解决的是不同层面的问题：shared_ptr 关心“谁拥有对象”，RCU 关心“何时可以安全地释放对象”。

6. RCU 模式的未来与应用

RCU 作为一种经过实践检验的并发模式，其在 Linux 内核中的广泛应用证明了其强大和可靠性。在用户空间，随着并发编程需求的不断增长，越来越多的高性能库和框架开始探索和采用 RCU 思想。例如，一些高性能的路由表、配置管理系统、内存数据库等都可能受益于 RCU。

正确实现 RCU 需要深刻理解 C++ 内存模型、原子操作和并发原语。它要求开发者在代码的正确性、性能和复杂性之间做出权衡。然而，对于那些对读取性能有极致要求的应用，RCU 提供的优势是无与伦比的。

RCU 的核心在于其精妙的宽限期机制，它将内存回收的责任从读者转移到写者和回收系统，从而实现了读者的无锁、无等待访问。掌握 RCU，就掌握了在高并发读场景下实现极致性能的关键技术之一。

---

通过今天的讲座，我们深入探讨了 RCU 的核心概念、C++ 实现的挑战与策略，并详细演示了基于 Hazard Pointers 的 RCU 示例。RCU 模式在读多写少的场景下，为我们提供了实现极致读取性能的强大工具，但其实现复杂性也要求我们投入更多的精力去理解和实践。在合适的场景下，运用 RCU 将能够显著提升您的并发系统的性能和可伸缩性。