| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | // ====================================================================== | ||
| 2 | // \title Os/Generic/LocklessPriorityQueue.hpp | ||
| 3 | // \brief lockless ISR-safe priority queue implementation for Os::Queue | ||
| 4 | // ====================================================================== | ||
| 5 | #ifndef OS_GENERIC_LOCKLESSPRIORITYQUEUE_HPP | ||
| 6 | #define OS_GENERIC_LOCKLESSPRIORITYQUEUE_HPP | ||
| 7 | |||
| 8 | #include <atomic> | ||
| 9 | #include <limits> | ||
| 10 | #include <type_traits> | ||
| 11 | #include "Fw/FPrimeBasicTypes.hpp" | ||
| 12 | #include "Os/Queue.hpp" | ||
| 13 | #include "config/LocklessQueueCfg.hpp" | ||
| 14 | |||
| 15 | namespace Os { | ||
| 16 | namespace Generic { | ||
| 17 | |||
| 18 | static_assert(std::is_integral<LocklessStateTagType>::value && std::is_unsigned<LocklessStateTagType>::value, | ||
| 19 | "LocklessStateTagType must be an unsigned integral type"); | ||
| 20 | |||
| 21 | //! \brief compile-time lock-free possibility for an atomic of unsigned integral width WIDTH | ||
| 22 | //! | ||
| 23 | //! C++14 lacks `std::atomic<T>::is_always_lock_free` (C++17), so this is derived from the | ||
| 24 | //! standard `ATOMIC_*_LOCK_FREE` macros (0 = never, 1 = sometimes, 2 = always lock-free), | ||
| 25 | //! selected by matching the width of the corresponding builtin type so no particular ABI | ||
| 26 | //! (e.g. `sizeof(int) == 4`) is assumed. Rejects only never-lock-free (value 0) widths at | ||
| 27 | //! compile time; on sometimes-lock-free (value 1) platforms the runtime `is_lock_free()` | ||
| 28 | //! FW_ASSERT in create() is the authoritative gate. | ||
| 29 | template <FwSizeType WIDTH> | ||
| 30 | struct LocklessAtomicLockFree { | ||
| 31 | static constexpr bool value = ((sizeof(unsigned char) == WIDTH) && (ATOMIC_CHAR_LOCK_FREE != 0)) || | ||
| 32 | ((sizeof(unsigned short) == WIDTH) && (ATOMIC_SHORT_LOCK_FREE != 0)) || | ||
| 33 | ((sizeof(unsigned int) == WIDTH) && (ATOMIC_INT_LOCK_FREE != 0)) || | ||
| 34 | ((sizeof(unsigned long) == WIDTH) && (ATOMIC_LONG_LOCK_FREE != 0)) || | ||
| 35 | ((sizeof(unsigned long long) == WIDTH) && (ATOMIC_LLONG_LOCK_FREE != 0)); | ||
| 36 | }; | ||
| 37 | |||
| 38 | static_assert(LocklessAtomicLockFree<sizeof(LocklessStateTagType)>::value, | ||
| 39 | "std::atomic<LocklessStateTagType> is never lock-free on this platform; " | ||
| 40 | "configure a narrower type in config/LocklessQueueCfg.hpp"); | ||
| 41 | static_assert(LocklessAtomicLockFree<sizeof(U32)>::value, "std::atomic<U32> is never lock-free on this platform"); | ||
| 42 | static_assert(LocklessAtomicLockFree<sizeof(FwQueuePriorityType)>::value, | ||
| 43 | "std::atomic<FwQueuePriorityType> is never lock-free on this platform"); | ||
| 44 | |||
| 45 | static_assert(LOCKLESS_QUEUE_MAX_RETRY_PASSES >= 1, "LOCKLESS_QUEUE_MAX_RETRY_PASSES must be at least 1"); | ||
| 46 | |||
| 47 | // A zero backoff could livelock a high-priority blocking caller against a lower-priority | ||
| 48 | // thread on a strict-priority scheduler. | ||
| 49 | static_assert(LOCKLESS_QUEUE_BLOCKING_BACKOFF_US > 0, "LOCKLESS_QUEUE_BLOCKING_BACKOFF_US must be greater than 0"); | ||
| 50 | |||
| 51 | static_assert((LOCKLESS_QUEUE_SLOT_ALIGNMENT & (LOCKLESS_QUEUE_SLOT_ALIGNMENT - 1)) == 0, | ||
| 52 | "LOCKLESS_QUEUE_SLOT_ALIGNMENT must be a power of two"); | ||
| 53 | |||
| 54 | //! \brief slot lifecycle states for the lockless priority queue | ||
| 55 | //! | ||
| 56 | //! Each slot in the queue moves through a four-state state machine. Producers transition slots | ||
| 57 | //! `FREE -> WRITING -> READY`. Consumers transition slots `READY -> READING -> FREE`. State values | ||
| 58 | //! occupy the low bits of a packed atomic; the remaining bits are used as an ABA tag. | ||
| 59 | enum LocklessSlotState : LocklessStateTagType { | ||
| 60 | LOCKLESS_SLOT_FREE = 0, //!< slot contains no data; available to a producer | ||
| 61 | LOCKLESS_SLOT_WRITING = 1, //!< producer has reserved the slot and is filling it | ||
| 62 | LOCKLESS_SLOT_READY = 2, //!< slot contains a published message available to a consumer | ||
| 63 | LOCKLESS_SLOT_READING = 3 //!< consumer has reserved the slot and is draining it | ||
| 64 | }; | ||
| 65 | |||
| 66 | //! \brief per-slot data for the lockless priority queue | ||
| 67 | //! | ||
| 68 | //! Each slot carries an atomic state-and-tag word that is the only synchronizing element of the | ||
| 69 | //! queue. Fields `m_priority` and `m_sequence` are atomic because they are read during the | ||
| 70 | //! consumer's scan phase (while the slot is `READY`) without exclusive ownership. They use | ||
| 71 | //! `relaxed` ordering; the happens-before relationship is established through the release/acquire | ||
| 72 | //! on `m_stateTag`. The non-atomic `m_size` field is only accessed under exclusive ownership | ||
| 73 | //! (`WRITING` by the producer, `READING` by the consumer). | ||
| 74 | struct alignas(LOCKLESS_QUEUE_SLOT_ALIGNMENT) LocklessSlot { | ||
| 75 | //! Number of low bits used for the state value within `m_stateTag`. | ||
| 76 | static constexpr U32 STATE_BITS = 2; | ||
| 77 | //! Mask for the state portion of `m_stateTag`. | ||
| 78 | static constexpr LocklessStateTagType STATE_MASK = | ||
| 79 | (static_cast<LocklessStateTagType>(1) << STATE_BITS) - static_cast<LocklessStateTagType>(1); | ||
| 80 | //! Number of bits available for the ABA epoch tag (62 with the default U64 word). | ||
| 81 | static constexpr U32 TAG_BITS = static_cast<U32>(std::numeric_limits<LocklessStateTagType>::digits) - STATE_BITS; | ||
| 82 | |||
| 83 | //! Packed (tag << STATE_BITS) | state word. Updated only via atomic operations. The TAG_BITS | ||
| 84 | //! tag increments on every transition; a stale CAS is defeated unless a thread stalls | ||
| 85 | //! between its scan and CAS across an exact multiple of 2^TAG_BITS transitions of one slot. | ||
| 86 | //! In that case the consumer dequeues a valid message out of priority order — never | ||
| 87 | //! corrupted, lost, or duplicated (see SDD sections 4 and 15). | ||
| 88 | std::atomic<LocklessStateTagType> m_stateTag; | ||
| 89 | //! Sequence number assigned at publication time. Used as a FIFO tiebreaker when priorities | ||
| 90 | //! are equal. Atomic because consumers read it during the scan phase without ownership. | ||
| 91 | std::atomic<U32> m_sequence; | ||
| 92 | //! Stored message size, less than or equal to the queue's configured message size. | ||
| 93 | //! Only accessed under exclusive ownership (WRITING or READING). | ||
| 94 | FwSizeType m_size; | ||
| 95 | //! Stored message priority. Atomic because consumers read it during the scan phase. | ||
| 96 | std::atomic<FwQueuePriorityType> m_priority; | ||
| 97 | |||
| 98 | //! Construct a slot in the FREE state with a zero tag. | ||
| 99 | LocklessSlot(); | ||
| 100 | }; | ||
| 101 | |||
| 102 | //! \brief handle for the lockless priority queue | ||
| 103 | //! | ||
| 104 | //! All persistent state for the queue is contained in this handle. Memory pointed to by `m_slots` | ||
| 105 | //! and `m_data` is allocated exactly once during `LocklessPriorityQueue::create` and freed during | ||
| 106 | //! `LocklessPriorityQueue::teardown`. | ||
| 107 | struct LocklessPriorityQueueHandle : public QueueHandle { | ||
| 108 | //! Pre-allocated array of `m_depth` slots, aligned within `m_slotsAllocation`. | ||
| 109 | LocklessSlot* m_slots; | ||
| 110 | //! Raw allocation backing `m_slots`; retained because allocators may ignore alignment. | ||
| 111 | void* m_slotsAllocation; | ||
| 112 | //! Pre-allocated array of `m_depth * m_messageSize` bytes for message payloads. | ||
| 113 | U8* m_data; | ||
| 114 | //! Configured queue depth in messages. | ||
| 115 | FwSizeType m_depth; | ||
| 116 | //! Configured maximum size of a single message. | ||
| 117 | FwSizeType m_messageSize; | ||
| 118 | //! Sequence assigned to messages on publication for FIFO tiebreak; may wrap (compared modularly). | ||
| 119 | std::atomic<U32> m_sequence; | ||
| 120 | //! Occupancy count (claimed-or-queued slots) used only for the high-water mark. | ||
| 121 | std::atomic<U32> m_count; | ||
| 122 | //! Receivable message count: incremented after a slot is published READY, decremented at | ||
| 123 | //! the successful READY->READING claim. Backs getMessagesAvailable(). | ||
| 124 | std::atomic<U32> m_available; | ||
| 125 | //! Maximum value `m_count` has ever held. Updated by producers via a bounded CAS loop. | ||
| 126 | std::atomic<U32> m_highMark; | ||
| 127 | //! Identifier passed to the memory allocator at create() time and reused at teardown(). | ||
| 128 | FwEnumStoreType m_id; | ||
| 129 | |||
| 130 | //! Default-construct a handle in the uncreated state. | ||
| 131 | LocklessPriorityQueueHandle(); | ||
| 132 | }; | ||
| 133 | |||
| 134 | //! \brief lockless ISR-safe priority queue implementation for Os::QueueInterface | ||
| 135 | //! | ||
| 136 | //! The lockless priority queue stores messages in a fixed pool of pre-allocated slots. Each slot | ||
| 137 | //! is governed by an atomic state machine that uses an embedded ABA tag, allowing producers and | ||
| 138 | //! consumers to manipulate the queue concurrently without taking any operating-system lock. | ||
| 139 | //! | ||
| 140 | //! \section flight_software_properties Flight-software properties | ||
| 141 | //! | ||
| 142 | //! - Memory: All memory is allocated through the configured `Fw::MemAllocator` exactly once | ||
| 143 | //! during `create`. No allocation occurs during `send`, `receive`, `getMessagesAvailable`, or | ||
| 144 | //! `getMessageHighWaterMark`. | ||
| 145 | //! - Loops: All non-blocking control paths are bounded by the configured queue depth multiplied | ||
| 146 | //! by `MAX_RETRY_PASSES`. The high-water-mark CAS loop is bounded by depth because the mark | ||
| 147 | //! only increases and never exceeds depth. Blocking paths poll with a fixed `Os::Task::delay` | ||
| 148 | //! backoff until the requested condition is satisfied; unlike the condition-variable-based | ||
| 149 | //! `Os::Generic::PriorityQueue`, an idle blocking caller wakes periodically rather than | ||
| 150 | //! sleeping until signaled. | ||
| 151 | //! - Determinism: All operations execute in time bounded by queue depth; no per-message dynamic | ||
| 152 | //! work scales with the number of producers or consumers. | ||
| 153 | //! | ||
| 154 | //! \section isr_safety ISR safety | ||
| 155 | //! | ||
| 156 | //! Non-blocking operations (`send` and `receive` with `BlockingType::NONBLOCKING`) are safe to | ||
| 157 | //! invoke from interrupt context because they use only lock-free atomic operations and bounded | ||
| 158 | //! `memcpy`. Blocking calls spin on the same atomic state and therefore must not be invoked from | ||
| 159 | //! ISR context. | ||
| 160 | //! | ||
| 161 | //! \section ordering Priority ordering | ||
| 162 | //! | ||
| 163 | //! Consumers pop the slot with the highest priority. When multiple slots share the same priority, | ||
| 164 | //! the one with the smallest sequence number (i.e. the earliest publication) is selected. | ||
| 165 | //! Sequence numbers are assigned by an atomic counter at publication time. The U32 counter may | ||
| 166 | //! wrap during a long mission; comparison uses unsigned modular subtraction so that wrap is | ||
| 167 | //! still ordered correctly within the queue's active window (create() enforces `depth < 2^31`). | ||
| 168 | //! FIFO tie-breaking assumes equal-priority messages do not remain queued across 2^31 | ||
| 169 | //! intervening sends; see the SDD (`docs/sdd-lockless-queue.md` section 7) for details. | ||
| 170 | class LocklessPriorityQueue final : public Os::QueueInterface { | ||
| 171 | public: | ||
| 172 | //! Maximum number of retry passes through the slot array before a non-blocking operation | ||
| 173 | //! gives up (configurable in config/LocklessQueueCfg.hpp). Each pass scans up to the | ||
| 174 | //! configured queue depth; the worst-case work per non-blocking call is therefore | ||
| 175 | //! `depth * MAX_RETRY_PASSES`. | ||
| 176 | static constexpr FwSizeType MAX_RETRY_PASSES = LOCKLESS_QUEUE_MAX_RETRY_PASSES; | ||
| 177 | |||
| 178 | //! \brief decide whether a candidate (priority, sequence) is preferred over the current best | ||
| 179 | //! | ||
| 180 | //! Highest priority wins; on a tie, the smallest sequence in the wrap-aware modular ordering | ||
| 181 | //! wins. Exposed as a static member so unit tests exercise the shipped comparison. | ||
| 182 | //! | ||
| 183 | //! \param candidatePriority: priority of the candidate message | ||
| 184 | //! \param candidateSequence: sequence of the candidate message | ||
| 185 | //! \param bestPriority: priority of the current best message | ||
| 186 | //! \param bestSequence: sequence of the current best message | ||
| 187 | //! \return true if the candidate should be preferred over the current best | ||
| 188 | static bool isCandidatePreferred(FwQueuePriorityType candidatePriority, | ||
| 189 | U32 candidateSequence, | ||
| 190 | FwQueuePriorityType bestPriority, | ||
| 191 | U32 bestSequence); | ||
| 192 | |||
| 193 | //! \brief default constructor | ||
| 194 | ✗ | LocklessPriorityQueue() = default; | |
| 195 | |||
| 196 | //! \brief destructor | ||
| 197 | //! | ||
| 198 | //! The destructor does **not** free queue resources. Owners must call `teardown()` | ||
| 199 | //! explicitly before destroying the queue (or its hosting `Os::Queue`). This matches the | ||
| 200 | //! `Os::Generic::PriorityQueue` contract and avoids a static-destruction-order fault | ||
| 201 | //! where the underlying `Fw::MemAllocatorRegistry` may already have been destroyed by the | ||
| 202 | //! time the destructor runs, which would manifest as a `pure virtual method called` abort | ||
| 203 | //! when `MemAllocator::deallocate` is invoked through its v-table. | ||
| 204 | ~LocklessPriorityQueue() override; | ||
| 205 | |||
| 206 | //! \brief constructing from a base reference is forbidden | ||
| 207 | LocklessPriorityQueue(const QueueInterface& other) = delete; | ||
| 208 | |||
| 209 | //! \brief constructing from a pointer is forbidden | ||
| 210 | LocklessPriorityQueue(const QueueInterface* other) = delete; | ||
| 211 | |||
| 212 | //! \brief assignment operator is forbidden | ||
| 213 | LocklessPriorityQueue& operator=(const QueueInterface& other) override = delete; | ||
| 214 | |||
| 215 | //! \brief create queue storage | ||
| 216 | //! | ||
| 217 | //! Allocates the slot pool and message-data region through the registered memory allocator. | ||
| 218 | //! | ||
| 219 | //! \warning allocates memory exactly once through the memory allocator registry; subsequent | ||
| 220 | //! `send` and `receive` calls do not allocate. | ||
| 221 | //! | ||
| 222 | //! \param id: identifier for the queue, used for memory allocation | ||
| 223 | //! \param name: name of queue (unused by this implementation) | ||
| 224 | //! \param depth: depth of queue in number of messages | ||
| 225 | //! \param messageSize: maximum size of an individual message | ||
| 226 | //! \return: status of the creation | ||
| 227 | Status create(FwEnumStoreType id, | ||
| 228 | const Fw::ConstStringBase& name, | ||
| 229 | FwSizeType depth, | ||
| 230 | FwSizeType messageSize) override; | ||
| 231 | |||
| 232 | //! \brief tear down the queue | ||
| 233 | //! | ||
| 234 | //! Returns memory acquired in `create` to the configured memory allocator. Safe to call | ||
| 235 | //! repeatedly; only the first call returns memory. | ||
| 236 | //! | ||
| 237 | //! \warning not thread-safe: the caller must guarantee no concurrent `send`, `receive`, | ||
| 238 | //! or `teardown` (including from ISR context) is in flight when this is invoked. | ||
| 239 | void teardown() override; | ||
| 240 | |||
| 241 | //! \brief send a message into the queue | ||
| 242 | //! | ||
| 243 | //! When `blockType` is `NONBLOCKING`, the operation completes in time bounded by | ||
| 244 | //! `depth * MAX_RETRY_PASSES` and returns `FULL` if no slot can be claimed. Because slots | ||
| 245 | //! held mid-operation by concurrent producers or consumers are not claimable, a | ||
| 246 | //! non-blocking send may return `FULL` under contention even though fewer than `depth` | ||
| 247 | //! messages are queued. When `blockType` is `BLOCKING`, the operation spins until a slot | ||
| 248 | //! becomes available. | ||
| 249 | //! | ||
| 250 | //! \warning `BLOCKING` calls must not be invoked from ISR context. | ||
| 251 | //! | ||
| 252 | //! \warning A spurious `FULL` triggers overflow handling on async ports (including | ||
| 253 | //! `assert`-on-overflow ports). Size queues with margin or select the mutex-based | ||
| 254 | //! `Os::Generic::PriorityQueue` where that is unacceptable. | ||
| 255 | //! | ||
| 256 | //! \param buffer: message data; must be non-null | ||
| 257 | //! \param size: size of message data; must be no greater than the configured message size | ||
| 258 | //! \param priority: priority of the message | ||
| 259 | //! \param blockType: BLOCKING to spin until space is available; NONBLOCKING to fail fast | ||
| 260 | //! \return: status of the send | ||
| 261 | Status send(const U8* buffer, FwSizeType size, FwQueuePriorityType priority, BlockingType blockType) override; | ||
| 262 | |||
| 263 | //! \brief receive a message from the queue | ||
| 264 | //! | ||
| 265 | //! Selects the highest-priority slot and, on a tie, the slot with the smallest sequence | ||
| 266 | //! number. When `blockType` is `NONBLOCKING`, the operation completes in time bounded by | ||
| 267 | //! `depth * MAX_RETRY_PASSES` and returns `EMPTY` if no slot is available. Because slots | ||
| 268 | //! held mid-operation by concurrent producers or consumers are not claimable, a | ||
| 269 | //! non-blocking receive may return `EMPTY` under contention even though messages are | ||
| 270 | //! queued. When `blockType` is `BLOCKING`, the operation spins until a slot is published. | ||
| 271 | //! | ||
| 272 | //! \warning `BLOCKING` calls must not be invoked from ISR context. | ||
| 273 | //! | ||
| 274 | //! \param destination: destination for message data; must be non-null | ||
| 275 | //! \param capacity: maximum size of message data the destination can hold; asserted to be | ||
| 276 | //! at least the size of the dequeued message. Supplying the configured message size is | ||
| 277 | //! always sufficient. | ||
| 278 | //! \param blockType: BLOCKING to spin for a message; NONBLOCKING to fail fast | ||
| 279 | //! \param actualSize: (output) actual size of the message read on success | ||
| 280 | //! \param priority: (output) priority of the message read on success | ||
| 281 | //! \return: status of the receive | ||
| 282 | Status receive(U8* destination, | ||
| 283 | FwSizeType capacity, | ||
| 284 | BlockingType blockType, | ||
| 285 | FwSizeType& actualSize, | ||
| 286 | FwQueuePriorityType& priority) override; | ||
| 287 | |||
| 288 | //! \brief get number of messages currently receivable | ||
| 289 | //! | ||
| 290 | //! A message counts only from its READY publication until a consumer claims it, so a | ||
| 291 | //! nonzero return means a receive of at least one message can complete. | ||
| 292 | //! | ||
| 293 | //! \return number of receivable messages currently in the queue | ||
| 294 | FwSizeType getMessagesAvailable() const override; | ||
| 295 | |||
| 296 | //! \brief get the maximum number of messages that have been queued at once | ||
| 297 | //! | ||
| 298 | //! \return high-water mark of message count | ||
| 299 | FwSizeType getMessageHighWaterMark() const override; | ||
| 300 | |||
| 301 | //! \brief return the underlying queue handle | ||
| 302 | QueueHandle* getHandle() override; | ||
| 303 | |||
| 304 | //! Persistent queue state. | ||
| 305 | LocklessPriorityQueueHandle m_handle; | ||
| 306 | }; | ||
| 307 | |||
| 308 | } // namespace Generic | ||
| 309 | } // namespace Os | ||
| 310 | |||
| 311 | #endif // OS_GENERIC_LOCKLESSPRIORITYQUEUE_HPP | ||
| 312 |