Designing a Zero-Allocation Circular Buffer Log Writer in C++ Using Atomic Memory Operations and Memory-Mapped Files — code snippets

snippet-1.txt

#include <atomic>
#include <cstdint>

struct alignas(64) LogSlot {
    std::atomic<uint32_t> status; // 0 = Empty, 1 = Writing, 2 = Ready
    uint32_t length;
    char payload[248];            // 256 bytes total slot size (perfectly fits 4 cache lines)
};

struct alignas(64) LogControlBlock {
    uint64_t magic;
    uint64_t version;
    
    alignas(64) std::atomic<uint64_t> write_index;
    alignas(64) std::atomic<uint64_t> read_index;
    
    uint64_t capacity;
    uint8_t reserved[32];
};

snippet-2.txt

#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <string>
#include <stdexcept>
#include <system_error>

class MappedLogFile {
public:
    MappedLogFile(const std::string& path, size_t size) : size_(size) {
        fd_ = ::open(path.c_str(), O_RDWR | O_CREAT, 0644);
        if (fd_ < 0) {
            throw std::system_error(errno, std::generic_category(), "Failed to open file");
        }

        // Pre-allocate physical blocks on disk to prevent runtime page faults
        int falloc_res = ::posix_fallocate(fd_, 0, size_);
        if (falloc_res != 0) {
            ::close(fd_);
            throw std::system_error(falloc_res, std::generic_category(), "posix_fallocate failed");
        }

        // Map file with MAP_POPULATE to eagerly allocate and populate page tables
        addr_ = ::mmap(nullptr, size_, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd_, 0);
        if (addr_ == MAP_FAILED) {
            ::close(fd_);
            throw std::system_error(errno, std::generic_category(), "mmap failed");
        }

        // Inform the kernel about sequential write intentions
        ::madvise(addr_, size_, MADV_SEQUENTIAL | MADV_WILLNEED);
    }

    ~MappedLogFile() {
        if (addr_ != MAP_FAILED) {
            ::munmap(addr_, size_);
        }
        if (fd_ >= 0) {
            ::close(fd_);
        }
    }

    void* get_address() const { return addr_; }

private:
    int fd_{-1};
    size_t size_;
    void* addr_{MAP_FAILED};
};

snippet-3.txt

#include <atomic>
#include <optional>

class RingBufferWriter {
public:
    RingBufferWriter(LogControlBlock* control, LogSlot* slots, uint64_t capacity)
        : control_(control), slots_(slots), capacity_(capacity) {}

    std::optional<uint64_t> reserve_slot() {
        uint64_t write_idx = control_->write_index.load(std::memory_order_relaxed);
        while (true) {
            uint64_t read_idx = control_->read_index.load(std::memory_order_acquire);
            
            // Check if buffer capacity is saturated
            if (write_idx - read_idx >= capacity_) {
                return std::nullopt; // Buffer full
            }
            
            // Attempt to reserve the slot index atomically.
            // If compare_exchange_weak fails, write_idx is updated with the current
            // value of write_index, allowing immediate retry without manual reload.
            if (control_->write_index.compare_exchange_weak(
                    write_idx, write_idx + 1, std::memory_order_release, std::memory_order_relaxed)) {
                return write_idx;
            }
        }
    }

private:
    LogControlBlock* control_;
    LogSlot* slots_;
    uint64_t capacity_;
};

snippet-4.txt

#include <string_view>
#include <cstring>

class LogWriter {
public:
    LogWriter(LogControlBlock* control, LogSlot* slots, uint64_t capacity)
        : writer_(control, slots, capacity), slots_(slots), capacity_(capacity) {}

    bool write(std::string_view msg) {
        if (msg.size() > sizeof(LogSlot::payload)) {
            return false; // Message exceeds slot payload capacity
        }

        auto opt_idx = writer_.reserve_slot();
        if (!opt_idx) {
            return false; // Buffer full
        }

        uint64_t idx = *opt_idx;
        LogSlot& slot = slots_[idx % capacity_];

        // Spin until the slot is marked Empty (0) by the consumer
        uint32_t expected = 0;
        while (!slot.status.compare_exchange_weak(expected, 1, std::memory_order_acquire)) {
            expected = 0;
#if defined(__x86_64__) || defined(_M_X64)
            __builtin_ia32_pause();
#elif defined(__aarch64__)
            asm volatile("yield" ::: "memory");
#endif
        }

        // Copy payload directly into the mapped slot
        std::memcpy(slot.payload, msg.data(), msg.size());
        slot.length = static_cast<uint32_t>(msg.size());

        // Mark the slot as Ready (2) for the consumer
        slot.status.store(2, std::memory_order_release);
        return true;
    }

private:
    RingBufferWriter writer_;
    LogSlot* slots_;
    uint64_t capacity_;
};

snippet-5.txt

#include <thread>
#include <atomic>
#include <string_view>

class LogConsumer {
public:
    LogConsumer(LogControlBlock* control, LogSlot* slots, uint64_t capacity)
        : control_(control), slots_(slots), capacity_(capacity), running_(true) {}

    void start() {
        worker_ = std::thread([this]() { drain_loop(); });
    }

    void stop() {
        running_.store(false, std::memory_order_release);
        if (worker_.joinable()) {
            worker_.join();
        }
    }

private:
    void drain_loop() {
        uint64_t local_read_idx = control_->read_index.load(std::memory_order_relaxed);
        while (running_.load(std::memory_order_relaxed)) {
            LogSlot& slot = slots_[local_read_idx % capacity_];

            // Spin-wait until slot status becomes Ready (2)
            uint32_t status = slot.status.load(std::memory_order_acquire);
            if (status != 2) {
#if defined(__x86_64__) || defined(_M_X64)
                __builtin_ia32_pause();
#endif
                std::this_thread::yield();
                continue;
            }

            // Reference payload directly without allocation or copying
            std::string_view payload(slot.payload, slot.length);
            process_log(payload);

            // Mark slot as Empty (0)
            slot.status.store(0, std::memory_order_release);

            // Advance the read index
            local_read_idx++;
            control_->read_index.store(local_read_idx, std::memory_order_release);
        }
    }

    void process_log(std::string_view msg) {
        // Output path logic goes here (e.g. streaming to secondary storage)
    }

    LogControlBlock* control_;
    LogSlot* slots_;
    uint64_t capacity_;
    std::atomic<bool> running_;
    std::thread worker_;
};

snippet-6.txt

#include <iostream>
#include <atomic>

class LogRecovery {
public:
    static void recover(LogControlBlock* control, LogSlot* slots, uint64_t capacity) {
        constexpr uint64_t EXPECTED_MAGIC = 0x5A414C5752545231ULL; // "ZALWRTR1"
        
        if (control->magic != EXPECTED_MAGIC) {
            std::cerr << "Cold startup or invalid log file. Formatting control block...\n";
            control->magic = EXPECTED_MAGIC;
            control->version = 1;
            control->write_index.store(0, std::memory_order_release);
            control->read_index.store(0, std::memory_order_release);
            control->capacity = capacity;
            for (uint64_t i = 0; i < capacity; ++i) {
                slots[i].status.store(0, std::memory_order_release);
            }
            return;
        }

        uint64_t write_idx = control->write_index.load(std::memory_order_acquire);
        uint64_t read_idx = control->read_index.load(std::memory_order_acquire);

        std::cout << "Recovering buffer index. Write index: " << write_idx 
                  << ", Read index: " << read_idx << std::endl;

        // Scan the active buffer window for incomplete writes
        for (uint64_t i = read_idx; i < write_idx; ++i) {
            LogSlot& slot = slots[i % capacity];
            uint32_t status = slot.status.load(std::memory_order_acquire);
            if (status == 1) { // Process crashed mid-write
                std::cout << "Detected torn write at index " << i 
                          << ". Resetting status and rolling back write_index.\n";
                slot.status.store(0, std::memory_order_release);
                control->write_index.store(i, std::memory_order_release);
                break;
            }
        }
    }
};

snippet-7.txt

#include <charconv>
#include <chrono>
#include <system_error>

char* format_timestamp(char* buf_ptr, char* buf_end) {
    auto now = std::chrono::system_clock::now();
    auto time_t_now = std::chrono::system_clock::to_time_t(now);
    std::tm tm_buf;
    ::localtime_r(&time_t_now, &tm_buf);
    
    // Write format: YYYY-MM-DD HH:MM:SS (19 bytes)
    int written = std::strftime(buf_ptr, buf_end - buf_ptr, "%Y-%m-%d %H:%M:%S", &tm_buf);
    if (written <= 0) return buf_ptr;
    return buf_ptr + written;
}

char* format_log_entry(char* buf, size_t size, std::string_view level, std::string_view msg, int code) {
    char* ptr = buf;
    char* end = buf + size;

    // 1. Format Timestamp
    ptr = format_timestamp(ptr, end);
    if (ptr + 4 >= end) return buf;
    
    // Append delimiter
    *ptr++ = ' ';
    *ptr++ = '[';
    
    // 2. Format Severity Level
    std::memcpy(ptr, level.data(), std::min(level.size(), static_cast<size_t>(end - ptr)));
    ptr += std::min(level.size(), static_cast<size_t>(end - ptr));
    if (ptr + 2 >= end) return buf;
    *ptr++ = ']';
    *ptr++ = ' ';

    // 3. Format message payload
    std::memcpy(ptr, msg.data(), std::min(msg.size(), static_cast<size_t>(end - ptr)));
    ptr += std::min(msg.size(), static_cast<size_t>(end - ptr));
    if (ptr + 10 >= end) return buf;

    // 4. Format integer code using zero-allocation std::to_chars
    *ptr++ = ' ';
    *ptr++ = '(';
    auto res = std::to_chars(ptr, end, code);
    if (res.ec == std::errc()) {
        ptr = res.ptr;
    }
    if (ptr < end) {
        *ptr++ = ')';
    }
    return ptr;
}

snippet-8.txt

#ifndef MAP_HUGE_2MB
#define MAP_HUGE_2MB (21 << MAP_HUGE_SHIFT)
#endif

// Example modification inside snippet-2 to allocate 2MB Huge Pages
void* allocate_huge_mmap(int fd, size_t size) {
    void* addr = ::mmap(nullptr, size, PROT_READ | PROT_WRITE, 
                         MAP_SHARED | MAP_POPULATE | MAP_HUGETLB | MAP_HUGE_2MB, fd, 0);
    return addr;
}