shm_queue.h

Go to the documentation of this file.

#pragma once
#include "posix_shm.h"
#include "shm_span.h"
#include "shm_table.h"
#include <atomic>
#include <concepts>
#include <span>
#include <optional>
#include <string_view>
 
template<typename T, typename TableType = shm_table>
    requires std::is_trivially_copyable_v<T>
class shm_queue : public shm_span<T, posix_shm_impl<TableType>> {
private:
    struct QueueHeader {
        std::atomic<size_t> head{0};
        std::atomic<size_t> tail{0};
        size_t capacity;
    };
    
    QueueHeader* header() {
        return reinterpret_cast<QueueHeader*>(
            static_cast<char*>(this->shm.get_base_addr()) + this->offset
        );
    }
    
    const QueueHeader* header() const {
        return reinterpret_cast<const QueueHeader*>(
            static_cast<const char*>(this->shm.get_base_addr()) + this->offset
        );
    }
    
    T* data_start() {
        return reinterpret_cast<T*>(
            reinterpret_cast<char*>(header()) + sizeof(QueueHeader)
        );
    }
 
    const typename TableType::entry* _table_entry{nullptr};
    
public:
    template<typename ShmType>
    shm_queue(ShmType& shm, std::string_view name, size_t capacity = 0)
        : shm_span<T, posix_shm_impl<TableType>>(shm, 0, 0) {
        static_assert(std::is_same_v<typename ShmType::table_type, TableType>,
                      "SharedMemory table type must match queue table type");
        
        auto* table = static_cast<TableType*>(shm.get_base_addr());
        
        // Convert string_view to null-terminated string for find
        char name_buf[TableType::MAX_NAME_SIZE]{};
        size_t copy_len = std::min(name.size(), sizeof(name_buf) - 1);
        std::copy_n(name.begin(), copy_len, name_buf);
        
        auto* entry = table->find(name_buf);
        
        if (entry) {
            // Open existing queue
            if (capacity != 0 && entry->num_elem != capacity) {
                throw std::runtime_error("Queue capacity mismatch");
            }
            this->offset = entry->offset;
            this->num_elem = entry->num_elem;
            _table_entry = entry;
            
            // Validate that the header has been initialized
            auto* hdr = header();
            if (hdr->capacity == 0) {
                // Header wasn't properly initialized, fix it
                hdr->capacity = this->num_elem + 1;  // Internal capacity
                hdr->head.store(0, std::memory_order_relaxed);
                hdr->tail.store(0, std::memory_order_relaxed);
            }
        } else if (capacity > 0) {
            // Create new queue
            // Allocate capacity+1 to allow full use of capacity (circular queue needs one empty slot)
            size_t actual_capacity = capacity + 1;
            size_t required_size = sizeof(QueueHeader) + actual_capacity * sizeof(T);
            size_t current_used = table->get_total_allocated_size();
            
            // Data must come after the table
            this->offset = sizeof(TableType) + current_used;
            this->num_elem = capacity;
            
            // Initialize header with actual capacity
            new (header()) QueueHeader{.capacity = actual_capacity};
            
            // Register in table
            table->add(name_buf, this->offset, required_size, sizeof(T), capacity);
            // Find the entry we just added
            _table_entry = table->find(name_buf);
        } else {
            throw std::runtime_error("Queue not found and capacity not specified");
        }
    }
    
    [[nodiscard]] bool enqueue(const T& value) noexcept {
        auto* hdr = header();
        size_t current_tail = hdr->tail.load(std::memory_order_relaxed);
        size_t next_tail = (current_tail + 1) % hdr->capacity;
        
        if (next_tail == hdr->head.load(std::memory_order_acquire)) {
            return false; // Queue is full
        }
        
        data_start()[current_tail] = value;
        hdr->tail.store(next_tail, std::memory_order_release);
        return true;
    }
    
    [[nodiscard]] std::optional<T> dequeue() noexcept {
        auto* hdr = header();
        size_t current_head = hdr->head.load(std::memory_order_relaxed);
        
        if (current_head == hdr->tail.load(std::memory_order_acquire)) {
            return std::nullopt; // Queue is empty
        }
        
        T value = data_start()[current_head];
        size_t next_head = (current_head + 1) % hdr->capacity;
        hdr->head.store(next_head, std::memory_order_release);
        
        return value;
    }
    
    [[nodiscard]] bool dequeue(T& out_value) noexcept {
        if (auto val = dequeue()) {
            out_value = *val;
            return true;
        }
        return false;
    }
    
    [[nodiscard]] bool empty() const noexcept {
        const auto* hdr = header();
        return hdr->head.load(std::memory_order_acquire) == 
               hdr->tail.load(std::memory_order_acquire);
    }
    
    [[nodiscard]] bool full() const noexcept {
        const auto* hdr = header();
        size_t current_tail = hdr->tail.load(std::memory_order_acquire);
        size_t next_tail = (current_tail + 1) % hdr->capacity;
        return next_tail == hdr->head.load(std::memory_order_acquire);
    }
    
    [[nodiscard]] size_t size() const noexcept {
        const auto* hdr = header();
        size_t h = hdr->head.load(std::memory_order_acquire);
        size_t t = hdr->tail.load(std::memory_order_acquire);
        
        if (t >= h) {
            return t - h;
        } else {
            return hdr->capacity - h + t;
        }
    }
    
    [[nodiscard]] size_t capacity() const noexcept {
        // Return user-visible capacity
        // For new queues: stored in num_elem
        // For consistency, use the stored value
        return this->num_elem;
    }
    
    [[nodiscard]] std::string_view name() const noexcept {
        return _table_entry ? std::string_view(_table_entry->name.data()) : std::string_view{};
    }
};