AtomicQueue.hpp

Go to the documentation of this file.

// ======================================================================
// \title  AtomicQueue.hpp
// \author B. Duckett
// \brief  A lock-free FIFO queue using atomics for thread/ISR safety
//
// \copyright
// Copyright 2026, by the California Institute of Technology.
// ALL RIGHTS RESERVED.  United States Government Sponsorship
// acknowledged.
//
// ======================================================================

#ifndef OS_GENERIC_TYPES_ATOMIC_QUEUE_HPP
#define OS_GENERIC_TYPES_ATOMIC_QUEUE_HPP

#include <Fw/FPrimeBasicTypes.hpp>
#include <Fw/Types/ByteArray.hpp>
#include <Fw/Types/MemAllocator.hpp>
#include <Os/CountingSemaphore.hpp>
#include <atomic>

// Forward declaration for test-only friend access
class AtomicQueueWrapAroundTest;

namespace Types {

class AtomicQueue {
    friend class ::AtomicQueueWrapAroundTest;  // Test-only accessor for counter manipulation
  public:
    AtomicQueue();

    ~AtomicQueue();

    // Rule of Five: prevent accidental copy/move of resource-managing class
    AtomicQueue(const AtomicQueue&) = delete;
    AtomicQueue& operator=(const AtomicQueue&) = delete;
    AtomicQueue(AtomicQueue&&) = delete;
    AtomicQueue& operator=(AtomicQueue&&) = delete;

    void create(FwSizeType numBuffers, FwSizeType bufferSize, Fw::MemAllocator& allocator, FwEnumStoreType allocatorId);

    void teardown();

    bool enqueue(const U8* buffer, FwSizeType size);

    bool enqueueBlocking(const U8* buffer, FwSizeType size, bool blockIfFull);

    bool dequeue(U8* buffer, FwSizeType capacity, FwSizeType& actualSize);

    bool isFull() const;

    bool isEmpty() const;

    FwSizeType getSize() const;

    FwSizeType getCapacity() const;

    FwSizeType getBufferSize() const;

    bool isCreated() const { return this->m_slots != nullptr && this->m_capacity > 0; }

  private:
    static constexpr FwSizeType MAX_CAS_RETRIES = 100;

    struct Slot {
        U8* buffer;                        // Embedded message buffer
        FwSizeType size;                   // Actual message size stored
        std::atomic<FwSizeType> sequence;  // Coordination sequence number
    };

    inline FwSizeType getIndex(FwSizeType pos) const {
        return (this->m_mask != 0) ? (pos & this->m_mask) : (pos % this->m_capacity);
    }

    static U32 computeChecksum(const U8* buffer, FwSizeType size);

    bool enqueueInternal(const U8* buffer, FwSizeType size);

    // Private members:
    Slot* m_slots;                         // Circular slot array
    U8* m_bufferMemory;                    // Contiguous buffer memory block
    FwSizeType m_capacity;                 // Number of message buffers
    FwSizeType m_bufferSize;               // Size of each message buffer
    FwSizeType m_mask;                     // Bitmask if power-of-2, else 0
    std::atomic<FwSizeType> m_enqueuePos;  // Next enqueue position (producer cursor)
    std::atomic<FwSizeType> m_dequeuePos;  // Next dequeue position (consumer cursor)
    Fw::MemAllocator* m_allocator;         // Memory allocator (nullptr if not using allocator)
    FwEnumStoreType m_allocatorId;         // Allocator identifier for deallocation (also used to identify asserts)
    Os::CountingSemaphore* m_notFullSem;   // Semaphore for blocking enqueue (all platforms)
};

}  // namespace Types

#endif  // OS_GENERIC_TYPES_ATOMIC_QUEUE_HPP