Designing a Lock-Free Read-Copy-Update (RCU) Cache in C++ for Ultra-Low Latency Routing Tables — code snippets

snippet-1.txt

#include <atomic>
#include <vector>
#include <thread>
#include <mutex>
#include <algorithm>
#include <limits>

struct Node {
    Node* next{nullptr};
    virtual ~Node() = default;
};

class EpochManager {
public:
    static constexpr uint64_t INACTIVE = std::numeric_limits<uint64_t>::max();
    static constexpr size_t CACHE_LINE_SIZE = 64;

    struct alignas(CACHE_LINE_SIZE) ThreadState {
        std::atomic<uint64_t> active_epoch{INACTIVE};
        std::atomic<bool> active{false};
        std::thread::id tid{};
    };

    struct RetiredObject {
        void* ptr;
        void (*deleter)(void*);
        uint64_t epoch;
    };

private:
    std::atomic<uint64_t> global_epoch_{0};
    std::vector<std::unique_ptr<ThreadState>> threads_;
    std::vector<RetiredObject> retired_list_;
    std::mutex mutex_;

    EpochManager() = default;

public:
    static EpochManager& instance() {
        static EpochManager inst;
        return inst;
    }

    ThreadState* register_thread() {
        std::lock_guard<std::mutex> lock(mutex_);
        auto state = std::make_unique<ThreadState>();
        state->tid = std::this_thread::get_id();
        state->active.store(true, std::memory_order_relaxed);
        auto* raw = state.get();
        threads_.push_back(std::move(state));
        return raw;
    }

    uint64_t get_global_epoch() const noexcept {
        return global_epoch_.load(std::memory_order_acquire);
    }

    void advance_epoch() noexcept {
        global_epoch_.fetch_add(1, std::memory_order_acq_rel);
    }

    void retire(void* ptr, void (*deleter)(void*)) {
        std::lock_guard<std::mutex> lock(mutex_);
        uint64_t current_epoch = global_epoch_.load(std::memory_order_relaxed);
        retired_list_.push_back({ptr, deleter, current_epoch});
    }

    void reclaim();
};

snippet-2.txt

#include <atomic>

thread_local EpochManager::ThreadState* local_state = nullptr;

class RcuGuard {
public:
    RcuGuard() noexcept {
        if (!local_state) [[unlikely]] {
            local_state = EpochManager::instance().register_thread();
        }
        // seq_cst memory order is required to prevent store-load reordering.
        // The registration of the active epoch MUST be visible to all cores
        // before any pointer chasing occurs within the critical section.
        uint64_t epoch = EpochManager::instance().get_global_epoch();
        local_state->active_epoch.store(epoch, std::memory_order_seq_cst);
    }

    ~RcuGuard() noexcept {
        // Resetting active epoch requires seq_cst to guarantee that no subsequent
        // reads or writes spill outside of the RCU read-side critical section.
        local_state->active_epoch.store(EpochManager::INACTIVE, std::memory_order_seq_cst);
    }
};

snippet-3.txt

#include <atomic>
#include <optional>
#include <cstring>

struct RouteEntry {
    uint32_t ip_address;
    uint32_t subnet_mask;
    char gateway[16];
    uint32_t interface_id;
};

struct RouteNode : public Node {
    RouteEntry entry;
};

class RcuRoutingCache {
private:
    static constexpr size_t BUCKET_COUNT = 4096;
    std::atomic<Node*> buckets_[BUCKET_COUNT]{};

    size_t hash(uint32_t ip) const noexcept {
        ip ^= ip >> 16;
        ip *= 0x85ebca6b;
        ip ^= ip >> 13;
        ip *= 0xc2b2ae35;
        ip ^= ip >> 16;
        return ip % BUCKET_COUNT;
    }

public:
    std::optional<RouteEntry> lookup(uint32_t ip) noexcept {
        RcuGuard guard; // Signals entering the RCU critical section
        size_t idx = hash(ip);
        
        // We load the bucket head pointer using acquire semantics.
        // This ensures all writes to the new node's contents are visible.
        Node* curr = buckets_[idx].load(std::memory_order_acquire);
        while (curr) {
            auto* rnode = static_cast<RouteNode*>(curr);
            if ((ip & rnode->entry.subnet_mask) == (rnode->entry.ip_address & rnode->entry.subnet_mask)) {
                return rnode->entry; // Safe to read, deferred memory reclamation is active
            }
            curr = curr->next;
        }
        return std::nullopt;
    }

    void update(const RouteEntry& new_entry);
};

snippet-4.txt

void RcuRoutingCache::update(const RouteEntry& new_entry) {
    size_t idx = hash(new_entry.ip_address);
    
    // Writer-side mutex to serialize writes. Lookups remain lock-free.
    static std::mutex writer_mutex;
    std::lock_guard<std::mutex> lock(writer_mutex);

    Node* old_head = buckets_[idx].load(std::memory_order_relaxed);
    Node* new_head = new RouteNode{ {nullptr}, new_entry };
    Node* curr_new = new_head;

    Node* curr = old_head;
    while (curr) {
        auto* rnode = static_cast<RouteNode*>(curr);
        if (!(rnode->entry.ip_address == new_entry.ip_address && 
              rnode->entry.subnet_mask == new_entry.subnet_mask)) {
            // Copy unchanged nodes to the new chain to keep them visible
            curr_new->next = new RouteNode{ {nullptr}, rnode->entry };
            curr_new = curr_new->next;
        }
        curr = curr->next;
    }

    // Publish the new bucket head atomically.
    // Release order guarantees that all node writes are visible before the pointer swap.
    buckets_[idx].store(new_head, std::memory_order_release);

    // Defer the deletion of the old chain
    if (old_head) {
        EpochManager::instance().retire(old_head, [](void* ptr) {
            Node* curr = static_cast<Node*>(ptr);
            while (curr) {
                Node* next = curr->next;
                delete curr; // Invokes virtual destructor safely
                curr = next;
            }
        });
    }
}

snippet-5.txt

void EpochManager::reclaim() {
    std::lock_guard<std::mutex> lock(mutex_);
    if (retired_list_.empty()) return;

    uint64_t min_active = std::numeric_limits<uint64_t>::max();
    for (const auto& t : threads_) {
        if (t->active.load(std::memory_order_relaxed)) {
            uint64_t thread_epoch = t->active_epoch.load(std::memory_order_acquire);
            if (thread_epoch != INACTIVE) {
                min_active = std::min(min_active, thread_epoch);
            }
        }
    }

    // Reclaim objects retired in epochs strictly older than the minimum active epoch.
    auto it = std::remove_if(retired_list_.begin(), retired_list_.end(),
        [min_active](const RetiredObject& obj) {
            if (obj.epoch < min_active) {
                obj.deleter(obj.ptr);
                return true;
            }
            return false;
        });
    retired_list_.erase(it, retired_list_.end());
}

snippet-6.txt

#include <new>
#include <atomic>

#ifdef __cpp_lib_hardware_interference_size
using std::hardware_destructive_interference_size;
#else
constexpr size_t hardware_destructive_interference_size = 64;
#endif

// We align the state struct to the hardware destructive interference size.
// This forces the compiler to pad the structure, preventing multiple threads 
// from sharing the same L1 cache line when writing their epochs.
struct alignas(hardware_destructive_interference_size) OptimizedThreadState {
    std::atomic<uint64_t> active_epoch{EpochManager::INACTIVE};
    std::atomic<bool> active{false};
    std::thread::id tid{};
};

snippet-7.txt

#include <benchmark/benchmark.h>
#include <shared_mutex>
#include <vector>

class LockedRoutingCache {
private:
    std::vector<RouteEntry> table_;
    mutable std::shared_mutex rw_lock_;

public:
    void update(const RouteEntry& entry) {
        std::unique_lock lock(rw_lock_);
        for (auto& item : table_) {
            if (item.ip_address == entry.ip_address && item.subnet_mask == entry.subnet_mask) {
                item = entry;
                return;
            }
        }
        table_.push_back(entry);
    }

    std::optional<RouteEntry> lookup(uint32_t ip) const {
        std::shared_lock lock(rw_lock_);
        for (const auto& entry : table_) {
            if ((ip & entry.subnet_mask) == (entry.ip_address & entry.subnet_mask)) {
                return entry;
            }
        }
        return std::nullopt;
    }
};

static void BM_LockedCacheLookup(benchmark::State& state) {
    static LockedRoutingCache cache;
    cache.update({0x0A000001, 0xFFFFFF00, "10.0.0.254", 1});
    
    for (auto _ : state) {
        auto res = cache.lookup(0x0A000001);
        benchmark::DoNotOptimize(res);
    }
}
BENCHMARK(BM_LockedCacheLookup)->ThreadRange(1, 64)->UseRealTime();

static void BM_RcuCacheLookup(benchmark::State& state) {
    static RcuRoutingCache cache;
    cache.update({0x0A000001, 0xFFFFFF00, "10.0.0.254", 1});

    for (auto _ : state) {
        auto res = cache.lookup(0x0A000001);
        benchmark::DoNotOptimize(res);
    }
}
BENCHMARK(BM_RcuCacheLookup)->ThreadRange(1, 64)->UseRealTime();