| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | // ====================================================================== | ||
| 2 | // \title Os/Generic/LocklessPriorityQueue.cpp | ||
| 3 | // \brief lockless ISR-safe priority queue implementation for Os::Queue | ||
| 4 | // ====================================================================== | ||
| 5 | #include "Os/Generic/LocklessPriorityQueue.hpp" | ||
| 6 | #include <atomic> | ||
| 7 | #include <cstring> | ||
| 8 | #include <limits> | ||
| 9 | #include "Fw/LanguageHelpers.hpp" | ||
| 10 | #include "Fw/Time/TimeInterval.hpp" | ||
| 11 | #include "Fw/Types/Assert.hpp" | ||
| 12 | #include "Fw/Types/ByteArray.hpp" | ||
| 13 | #include "Fw/Types/MemAllocator.hpp" | ||
| 14 | #include "Os/Task.hpp" | ||
| 15 | #include "config/MemoryAllocatorTypeEnumAc.hpp" | ||
| 16 | |||
| 17 | namespace Os { | ||
| 18 | namespace Generic { | ||
| 19 | |||
| 20 | namespace { | ||
| 21 | |||
| 22 | //! Extract the state portion of a packed state-tag word. | ||
| 23 | 72590891 | constexpr LocklessStateTagType stateOf(LocklessStateTagType packed) { | |
| 24 | 72590891 | return packed & LocklessSlot::STATE_MASK; | |
| 25 | } | ||
| 26 | |||
| 27 | //! Extract the tag portion of a packed state-tag word. | ||
| 28 | 7897941 | constexpr LocklessStateTagType tagOf(LocklessStateTagType packed) { | |
| 29 | 7897941 | return packed >> LocklessSlot::STATE_BITS; | |
| 30 | } | ||
| 31 | |||
| 32 | //! Pack a (state, tag) pair into a single word. | ||
| 33 | 7913332 | constexpr LocklessStateTagType packStateTag(LocklessStateTagType state, LocklessStateTagType tag) { | |
| 34 | 7913332 | return (tag << LocklessSlot::STATE_BITS) | (state & LocklessSlot::STATE_MASK); | |
| 35 | } | ||
| 36 | |||
| 37 | } // namespace | ||
| 38 | |||
| 39 | 33944400 | bool LocklessPriorityQueue::isCandidatePreferred(FwQueuePriorityType candidatePriority, | |
| 40 | U32 candidateSequence, | ||
| 41 | FwQueuePriorityType bestPriority, | ||
| 42 | U32 bestSequence) { | ||
| 43 | 33944400 | bool preferred = false; | |
| 44 |
2/2✓ Branch 0 taken 4367411 times.
✓ Branch 1 taken 29576989 times.
|
33944400 | if (candidatePriority > bestPriority) { |
| 45 | 4367411 | preferred = true; | |
| 46 |
2/2✓ Branch 0 taken 10331306 times.
✓ Branch 1 taken 19245683 times.
|
29576989 | } else if (candidatePriority == bestPriority) { |
| 47 | // Modular subtraction: if (candidate - best) interpreted unsigned has its top bit set, | ||
| 48 | // candidate is "older" (smaller in the wrap-aware ordering) than best. | ||
| 49 | 10331306 | const U32 difference = candidateSequence - bestSequence; | |
| 50 | 10331306 | const U32 topBit = static_cast<U32>(1) << (std::numeric_limits<U32>::digits - 1); | |
| 51 | 10331306 | preferred = (difference & topBit) != 0; | |
| 52 | } | ||
| 53 | 33944400 | return preferred; | |
| 54 | } | ||
| 55 | |||
| 56 | 2280 | LocklessSlot::LocklessSlot() | |
| 57 | 2280 | : m_stateTag(packStateTag(LOCKLESS_SLOT_FREE, 0)), m_sequence(0), m_size(0), m_priority(0) {} | |
| 58 | |||
| 59 | 203 | LocklessPriorityQueueHandle::LocklessPriorityQueueHandle() | |
| 60 | : QueueHandle(), | ||
| 61 | 203 | m_slots(nullptr), | |
| 62 | 203 | m_slotsAllocation(nullptr), | |
| 63 | 203 | m_data(nullptr), | |
| 64 | 203 | m_depth(0), | |
| 65 | 203 | m_messageSize(0), | |
| 66 | 203 | m_sequence(0), | |
| 67 | 203 | m_count(0), | |
| 68 | 203 | m_available(0), | |
| 69 | 203 | m_highMark(0), | |
| 70 | 203 | m_id(0) {} | |
| 71 | |||
| 72 | 406 | LocklessPriorityQueue::~LocklessPriorityQueue() { | |
| 73 | // Intentionally empty: cleanup is the responsibility of an explicit `teardown()` call. | ||
| 74 | // Freeing here would call the `Fw::MemAllocatorRegistry` singleton, whose destruction | ||
| 75 | // order relative to global queue objects is unspecified (matches | ||
| 76 | // `Os::Generic::PriorityQueue::~PriorityQueue()`). | ||
| 77 | 406 | } | |
| 78 | |||
| 79 | 192 | QueueInterface::Status LocklessPriorityQueue::create(FwEnumStoreType id, | |
| 80 | const Fw::ConstStringBase& name, | ||
| 81 | FwSizeType depth, | ||
| 82 | FwSizeType messageSize) { | ||
| 83 | static_cast<void>(name); | ||
| 84 | |||
| 85 | // Ensure that the queue has not already been created. A double create would leak memory and | ||
| 86 | // is a programming error. | ||
| 87 | 192 | FW_ASSERT(this->m_handle.m_slots == nullptr); | |
| 88 | 192 | FW_ASSERT(this->m_handle.m_data == nullptr); | |
| 89 | |||
| 90 | // The state-tag word relies on the underlying atomic being lock-free for ISR safety. The | ||
| 91 | // static_assert in the header covers platforms with a compile-time guarantee; this runtime | ||
| 92 | // check is the authoritative gate (`is_always_lock_free` is C++17 and this code targets | ||
| 93 | // C++14). | ||
| 94 | 192 | std::atomic<LocklessStateTagType> probe(0); | |
| 95 | 192 | FW_ASSERT(probe.is_lock_free()); | |
| 96 | 192 | std::atomic<FwQueuePriorityType> priorityProbe(0); | |
| 97 | 192 | FW_ASSERT(priorityProbe.is_lock_free()); | |
| 98 | 192 | std::atomic<U32> counterProbe(0); | |
| 99 | 192 | FW_ASSERT(counterProbe.is_lock_free()); | |
| 100 | 192 | FW_ASSERT(depth > 0); | |
| 101 | 192 | FW_ASSERT(messageSize > 0); | |
| 102 | |||
| 103 | // Guard the multiplications used to compute allocation sizes against overflow. | ||
| 104 | 192 | const FwSizeType maxSize = std::numeric_limits<FwSizeType>::max(); | |
| 105 | 192 | FW_ASSERT(depth <= (maxSize / sizeof(LocklessSlot))); | |
| 106 | 192 | FW_ASSERT(depth <= (maxSize / messageSize)); | |
| 107 | |||
| 108 | // The modular comparison in isCandidatePreferred is exact only while queued equal-priority | ||
| 109 | // messages span less than half the U32 sequence domain (SDD section 7); depth bounds message | ||
| 110 | // count, not sequence spread. | ||
| 111 | 192 | FW_ASSERT(depth < (std::numeric_limits<U32>::max() / 2)); | |
| 112 | |||
| 113 |
3/3✓ Branch 1 taken 192 times.
✓ Branch 5 taken 192 times.
✓ Branch 8 taken 192 times.
|
384 | Fw::MemAllocator& allocator = Fw::MemAllocatorRegistry::getInstance().getAnAllocator( |
| 114 | 192 | Fw::MemoryAllocation::MemoryAllocatorType::OS_GENERIC_PRIORITY_QUEUE); | |
| 115 | |||
| 116 | 192 | LocklessSlot* slots = nullptr; | |
| 117 | 192 | void* slotsAllocation = nullptr; | |
| 118 | 192 | U8* data = nullptr; | |
| 119 | 192 | QueueInterface::Status status = QueueInterface::Status::OP_OK; | |
| 120 | |||
| 121 | // Allocate the slot array. Request padding for manual alignment: allocators are not required | ||
| 122 | // to honor the alignment argument (e.g. Fw::MallocAllocator ignores it). | ||
| 123 | 192 | FW_ASSERT(depth * sizeof(LocklessSlot) <= (maxSize - alignof(LocklessSlot))); | |
| 124 | 192 | FwSizeType slotBytesRequested = (depth * sizeof(LocklessSlot)) + alignof(LocklessSlot); | |
| 125 | 192 | FwSizeType slotBytesAllocated = slotBytesRequested; | |
| 126 |
1/1✓ Branch 4 taken 192 times.
|
192 | slotsAllocation = allocator.allocate(id, slotBytesAllocated, alignof(LocklessSlot)); |
| 127 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 192 times.
|
192 | if (slotsAllocation == nullptr) { |
| 128 | ✗ | status = QueueInterface::Status::ALLOCATION_FAILED; | |
| 129 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 192 times.
|
192 | } else if (slotBytesAllocated < slotBytesRequested) { |
| 130 | ✗ | allocator.deallocate(id, slotsAllocation); | |
| 131 | ✗ | status = QueueInterface::Status::ALLOCATION_FAILED; | |
| 132 | } else { | ||
| 133 | 192 | const PlatformPointerCastType base = reinterpret_cast<PlatformPointerCastType>(slotsAllocation); | |
| 134 | 192 | const PlatformPointerCastType aligned = | |
| 135 | 192 | (base + (alignof(LocklessSlot) - 1)) & ~static_cast<PlatformPointerCastType>(alignof(LocklessSlot) - 1); | |
| 136 | 192 | const FwSizeType offset = static_cast<FwSizeType>(aligned - base); | |
| 137 |
1/1✓ Branch 4 taken 192 times.
|
192 | slots = Fw::arrayPlacementNew<LocklessSlot>( |
| 138 | Fw::ByteArray(static_cast<U8*>(slotsAllocation) + offset, slotBytesAllocated - offset), depth); | ||
| 139 | } | ||
| 140 | |||
| 141 | // Allocate the message-data region. | ||
| 142 |
1/2✓ Branch 0 taken 192 times.
✗ Branch 1 not taken.
|
192 | if (status == QueueInterface::Status::OP_OK) { |
| 143 | 192 | FwSizeType dataBytesRequested = depth * messageSize; | |
| 144 | 192 | FwSizeType dataBytesAllocated = dataBytesRequested; | |
| 145 |
1/1✓ Branch 4 taken 192 times.
|
192 | void* dataAllocation = allocator.allocate(id, dataBytesAllocated, alignof(U8)); |
| 146 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 192 times.
|
192 | if (dataAllocation == nullptr) { |
| 147 | ✗ | Fw::arrayPlacementDestruct<LocklessSlot>(slots, depth); | |
| 148 | ✗ | allocator.deallocate(id, slotsAllocation); | |
| 149 | ✗ | status = QueueInterface::Status::ALLOCATION_FAILED; | |
| 150 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 192 times.
|
192 | } else if (dataBytesAllocated < dataBytesRequested) { |
| 151 | ✗ | Fw::arrayPlacementDestruct<LocklessSlot>(slots, depth); | |
| 152 | ✗ | allocator.deallocate(id, slotsAllocation); | |
| 153 | ✗ | allocator.deallocate(id, dataAllocation); | |
| 154 | ✗ | status = QueueInterface::Status::ALLOCATION_FAILED; | |
| 155 | } else { | ||
| 156 | 192 | data = static_cast<U8*>(dataAllocation); | |
| 157 | } | ||
| 158 | } | ||
| 159 | |||
| 160 | // Publish the configured handle once both allocations succeeded. | ||
| 161 |
1/2✓ Branch 0 taken 192 times.
✗ Branch 1 not taken.
|
192 | if (status == QueueInterface::Status::OP_OK) { |
| 162 | 192 | this->m_handle.m_id = id; | |
| 163 | 192 | this->m_handle.m_messageSize = messageSize; | |
| 164 | 192 | this->m_handle.m_depth = depth; | |
| 165 | 192 | this->m_handle.m_slots = slots; | |
| 166 | 192 | this->m_handle.m_slotsAllocation = slotsAllocation; | |
| 167 | 192 | this->m_handle.m_data = data; | |
| 168 | 192 | this->m_handle.m_sequence.store(0, std::memory_order_relaxed); | |
| 169 | 192 | this->m_handle.m_count.store(0, std::memory_order_relaxed); | |
| 170 | 192 | this->m_handle.m_available.store(0, std::memory_order_relaxed); | |
| 171 | 192 | this->m_handle.m_highMark.store(0, std::memory_order_relaxed); | |
| 172 | } | ||
| 173 | 384 | return status; | |
| 174 | } | ||
| 175 | |||
| 176 | 203 | void LocklessPriorityQueue::teardown() { | |
| 177 |
2/2✓ Branch 4 taken 192 times.
✓ Branch 5 taken 11 times.
|
203 | if (this->m_handle.m_slots != nullptr) { |
| 178 |
2/2✓ Branch 3 taken 192 times.
✓ Branch 6 taken 192 times.
|
384 | Fw::MemAllocator& allocator = Fw::MemAllocatorRegistry::getInstance().getAnAllocator( |
| 179 | 192 | Fw::MemoryAllocation::MemoryAllocatorType::OS_GENERIC_PRIORITY_QUEUE); | |
| 180 | 192 | Fw::arrayPlacementDestruct<LocklessSlot>(this->m_handle.m_slots, this->m_handle.m_depth); | |
| 181 | 192 | allocator.deallocate(this->m_handle.m_id, this->m_handle.m_slotsAllocation); | |
| 182 |
1/2✓ Branch 4 taken 192 times.
✗ Branch 5 not taken.
|
192 | if (this->m_handle.m_data != nullptr) { |
| 183 | 192 | allocator.deallocate(this->m_handle.m_id, this->m_handle.m_data); | |
| 184 | } | ||
| 185 | 192 | this->m_handle.m_slots = nullptr; | |
| 186 | 192 | this->m_handle.m_slotsAllocation = nullptr; | |
| 187 | 192 | this->m_handle.m_data = nullptr; | |
| 188 | 192 | this->m_handle.m_depth = 0; | |
| 189 | 192 | this->m_handle.m_messageSize = 0; | |
| 190 | 192 | this->m_handle.m_count.store(0, std::memory_order_relaxed); | |
| 191 | 192 | this->m_handle.m_available.store(0, std::memory_order_relaxed); | |
| 192 | 192 | this->m_handle.m_highMark.store(0, std::memory_order_relaxed); | |
| 193 | 192 | this->m_handle.m_sequence.store(0, std::memory_order_relaxed); | |
| 194 | } | ||
| 195 | 203 | } | |
| 196 | |||
| 197 | 2133115 | QueueInterface::Status LocklessPriorityQueue::send(const U8* buffer, | |
| 198 | FwSizeType size, | ||
| 199 | FwQueuePriorityType priority, | ||
| 200 | QueueInterface::BlockingType blockType) { | ||
| 201 | // Programming-error checks: queue must be created and inputs must be well-formed. These are | ||
| 202 | // preconditions, not untrusted-input checks. | ||
| 203 | 2133115 | FW_ASSERT(this->m_handle.m_slots != nullptr); | |
| 204 | 2136869 | FW_ASSERT(this->m_handle.m_data != nullptr); | |
| 205 | 2136898 | FW_ASSERT(buffer != nullptr); | |
| 206 | |||
| 207 | // Reject oversized messages without touching the queue. | ||
| 208 |
2/2✓ Branch 4 taken 1 times.
✓ Branch 5 taken 2136883 times.
|
2136898 | if (size > this->m_handle.m_messageSize) { |
| 209 | 1 | return QueueInterface::Status::SIZE_MISMATCH; | |
| 210 | } | ||
| 211 | |||
| 212 | 2136883 | const FwSizeType depth = this->m_handle.m_depth; | |
| 213 | 2132109 | const bool blocking = (blockType == QueueInterface::BlockingType::BLOCKING); | |
| 214 | |||
| 215 | // Bounded for non-blocking, unbounded for blocking (explicit user contract). | ||
| 216 |
4/4✓ Branch 0 taken 24155 times.
✓ Branch 1 taken 4546353 times.
✓ Branch 2 taken 4078763 times.
✓ Branch 3 taken 467590 times.
|
4570508 | for (FwSizeType pass = 0; blocking || (pass < MAX_RETRY_PASSES); pass++) { |
| 217 |
2/2✓ Branch 0 taken 55348640 times.
✓ Branch 1 taken 2594840 times.
|
57943480 | for (FwSizeType i = 0; i < depth; i++) { |
| 218 | 55348640 | LocklessSlot& slot = this->m_handle.m_slots[i]; | |
| 219 | 55361828 | LocklessStateTagType packed = slot.m_stateTag.load(std::memory_order_acquire); | |
| 220 |
2/2✓ Branch 1 taken 53749534 times.
✓ Branch 2 taken 1472545 times.
|
55279462 | if (stateOf(packed) != LOCKLESS_SLOT_FREE) { |
| 221 | 53749534 | continue; | |
| 222 | } | ||
| 223 | 1472545 | const LocklessStateTagType desired = packStateTag(LOCKLESS_SLOT_WRITING, tagOf(packed) + 1); | |
| 224 |
2/2✓ Branch 2 taken 1669560 times.
✓ Branch 3 taken 40312 times.
|
3420263 | if (slot.m_stateTag.compare_exchange_strong(packed, desired, std::memory_order_acq_rel, |
| 225 | std::memory_order_relaxed)) { | ||
| 226 |
2/2✓ Branch 0 taken 1668800 times.
✓ Branch 1 taken 760 times.
|
1669560 | if (size > 0) { |
| 227 | 1668800 | const FwSizeType offset = i * this->m_handle.m_messageSize; | |
| 228 |
2/4✗ Branch 5 not taken.
✓ Branch 6 taken 1668055 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 1668055 times.
|
1669212 | static_cast<void>(::memcpy(this->m_handle.m_data + offset, buffer, static_cast<size_t>(size))); |
| 229 | } | ||
| 230 | 1668815 | slot.m_size = size; | |
| 231 | 1668093 | slot.m_priority.store(priority, std::memory_order_relaxed); | |
| 232 | 3337236 | slot.m_sequence.store(this->m_handle.m_sequence.fetch_add(1, std::memory_order_relaxed), | |
| 233 | std::memory_order_relaxed); | ||
| 234 | |||
| 235 | // Increment the occupancy count *before* publishing READY. This guarantees a | ||
| 236 | // consumer's decrement (which can only follow a READY observation) never precedes | ||
| 237 | // this increment, so m_count cannot transiently underflow. | ||
| 238 | 1667359 | const U32 nextCount = this->m_handle.m_count.fetch_add(1, std::memory_order_acq_rel) + 1; | |
| 239 | |||
| 240 | 1669328 | slot.m_stateTag.store(packStateTag(LOCKLESS_SLOT_READY, tagOf(desired) + 1), std::memory_order_release); | |
| 241 | |||
| 242 | // Increment the receivable count only after READY is published, so a nonzero | ||
| 243 | // getMessagesAvailable() implies at least one message has been made receivable. | ||
| 244 | 1664867 | static_cast<void>(this->m_handle.m_available.fetch_add(1, std::memory_order_acq_rel)); | |
| 245 | |||
| 246 | // Raise the high-water mark after publication so the message is never invisible | ||
| 247 | // while the producer runs this loop. Each strong-CAS failure strictly raises | ||
| 248 | // prevMark (mark only increases, capped at depth), so at most depth iterations run. | ||
| 249 | 1669283 | U32 prevMark = this->m_handle.m_highMark.load(std::memory_order_relaxed); | |
| 250 |
4/4✓ Branch 0 taken 1664762 times.
✓ Branch 1 taken 1261 times.
✓ Branch 2 taken 1932 times.
✓ Branch 3 taken 1662830 times.
|
1666023 | for (FwSizeType markPass = 0; (markPass < depth) && (nextCount > prevMark); markPass++) { |
| 251 |
2/2✓ Branch 3 taken 1927 times.
✓ Branch 4 taken 5 times.
|
3864 | if (this->m_handle.m_highMark.compare_exchange_strong( |
| 252 | prevMark, nextCount, std::memory_order_relaxed, std::memory_order_relaxed)) { | ||
| 253 | 1927 | break; | |
| 254 | } | ||
| 255 | } | ||
| 256 | 1666018 | return QueueInterface::Status::OP_OK; | |
| 257 | } | ||
| 258 | } | ||
| 259 | // No free slot found this pass. Blocking callers back off; non-blocking callers retry. | ||
| 260 |
2/2✓ Branch 0 taken 1872 times.
✓ Branch 1 taken 2592968 times.
|
2594840 | if (blocking) { |
| 261 |
2/2✓ Branch 2 taken 1886 times.
✓ Branch 6 taken 1890 times.
|
1872 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, LOCKLESS_QUEUE_BLOCKING_BACKOFF_US))); |
| 262 | } | ||
| 263 | } | ||
| 264 | 467590 | return QueueInterface::Status::FULL; | |
| 265 | } | ||
| 266 | |||
| 267 | 2568841 | QueueInterface::Status LocklessPriorityQueue::receive(U8* destination, | |
| 268 | FwSizeType capacity, | ||
| 269 | QueueInterface::BlockingType blockType, | ||
| 270 | FwSizeType& actualSize, | ||
| 271 | FwQueuePriorityType& priority) { | ||
| 272 | // Programming-error checks. These mirror the assertions in send(). | ||
| 273 | 2568841 | FW_ASSERT(this->m_handle.m_slots != nullptr); | |
| 274 | 2594294 | FW_ASSERT(this->m_handle.m_data != nullptr); | |
| 275 | 2597045 | FW_ASSERT(destination != nullptr); | |
| 276 | |||
| 277 | 2597045 | const FwSizeType depth = this->m_handle.m_depth; | |
| 278 | 2596550 | const bool blocking = (blockType == QueueInterface::BlockingType::BLOCKING); | |
| 279 | |||
| 280 | // Bounded for non-blocking, unbounded for blocking (explicit user contract). | ||
| 281 |
4/4✓ Branch 0 taken 99257 times.
✓ Branch 1 taken 6788263 times.
✓ Branch 2 taken 5908647 times.
✓ Branch 3 taken 879616 times.
|
6887520 | for (FwSizeType pass = 0; blocking || (pass < MAX_RETRY_PASSES); pass++) { |
| 282 | // Fast path: skip the O(depth) scan while no message is receivable. | ||
| 283 |
2/2✓ Branch 3 taken 2638040 times.
✓ Branch 4 taken 3275082 times.
|
11921026 | if (this->m_handle.m_available.load(std::memory_order_acquire) == 0) { |
| 284 |
2/2✓ Branch 0 taken 1835 times.
✓ Branch 1 taken 2636205 times.
|
2638040 | if (blocking) { |
| 285 |
2/2✓ Branch 2 taken 1839 times.
✓ Branch 6 taken 1819 times.
|
1835 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, LOCKLESS_QUEUE_BLOCKING_BACKOFF_US))); |
| 286 | } | ||
| 287 | 2634521 | continue; | |
| 288 | } | ||
| 289 | 3275082 | FwSizeType bestIndex = depth; | |
| 290 | 3275082 | FwQueuePriorityType bestPriority = FwQueuePriorityType(); | |
| 291 | 3275082 | U32 bestSequence = 0; | |
| 292 | 3275082 | LocklessStateTagType bestPacked = 0; | |
| 293 | |||
| 294 |
2/2✓ Branch 0 taken 44495469 times.
✓ Branch 1 taken 6840451 times.
|
51335920 | for (FwSizeType i = 0; i < depth; i++) { |
| 295 | 44495469 | LocklessSlot& slot = this->m_handle.m_slots[i]; | |
| 296 | 45098105 | LocklessStateTagType packed = slot.m_stateTag.load(std::memory_order_acquire); | |
| 297 |
2/2✓ Branch 1 taken 12233853 times.
✓ Branch 2 taken 31769154 times.
|
43826529 | if (stateOf(packed) != LOCKLESS_SLOT_READY) { |
| 298 | 12233853 | continue; | |
| 299 | } | ||
| 300 | 31769154 | const FwQueuePriorityType candidatePriority = slot.m_priority.load(std::memory_order_relaxed); | |
| 301 | 35588809 | const U32 candidateSequence = slot.m_sequence.load(std::memory_order_relaxed); | |
| 302 | // Recheck: if the state-tag changed, the relaxed priority/sequence reads above may | ||
| 303 | // belong to a recycled slot; discard the candidate and keep scanning. | ||
| 304 | 35307746 | const LocklessStateTagType packedRecheck = slot.m_stateTag.load(std::memory_order_acquire); | |
| 305 |
2/2✓ Branch 0 taken 110985 times.
✓ Branch 1 taken 34783310 times.
|
34894295 | if (packed != packedRecheck) { |
| 306 | 110985 | continue; | |
| 307 | } | ||
| 308 |
6/6✓ Branch 0 taken 33548047 times.
✓ Branch 1 taken 1235263 times.
✓ Branch 2 taken 8247529 times.
✓ Branch 3 taken 26210955 times.
✓ Branch 4 taken 9494650 times.
✓ Branch 5 taken 26199097 times.
|
69241794 | if ((bestIndex == depth) || |
| 309 | 33548047 | isCandidatePreferred(candidatePriority, candidateSequence, bestPriority, bestSequence)) { | |
| 310 | 9494650 | bestIndex = i; | |
| 311 | 9494650 | bestPriority = candidatePriority; | |
| 312 | 9494650 | bestSequence = candidateSequence; | |
| 313 | 9494650 | bestPacked = packed; | |
| 314 | } | ||
| 315 | } | ||
| 316 | |||
| 317 |
2/2✓ Branch 0 taken 50645 times.
✓ Branch 1 taken 6789806 times.
|
6840451 | if (bestIndex == depth) { |
| 318 |
2/2✓ Branch 0 taken 19 times.
✓ Branch 1 taken 50626 times.
|
50645 | if (blocking) { |
| 319 |
2/2✓ Branch 2 taken 19 times.
✓ Branch 6 taken 19 times.
|
19 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, LOCKLESS_QUEUE_BLOCKING_BACKOFF_US))); |
| 320 | } | ||
| 321 | 50609 | continue; | |
| 322 | } | ||
| 323 | |||
| 324 | 6789806 | const LocklessStateTagType desired = packStateTag(LOCKLESS_SLOT_READING, tagOf(bestPacked) + 1); | |
| 325 |
2/2✓ Branch 7 taken 1669740 times.
✓ Branch 8 taken 1434953 times.
|
6204424 | if (this->m_handle.m_slots[bestIndex].m_stateTag.compare_exchange_strong( |
| 326 | bestPacked, desired, std::memory_order_acq_rel, std::memory_order_relaxed)) { | ||
| 327 | // Decrement the receivable count at the successful READY->READING claim: this | ||
| 328 | // message can no longer complete another receive. | ||
| 329 | 1669740 | static_cast<void>(this->m_handle.m_available.fetch_sub(1, std::memory_order_acq_rel)); | |
| 330 | 1669731 | LocklessSlot& slot = this->m_handle.m_slots[bestIndex]; | |
| 331 | 1669671 | const FwSizeType storedSize = slot.m_size; | |
| 332 | 1669675 | FW_ASSERT(storedSize <= capacity); | |
| 333 |
2/2✓ Branch 0 taken 1669463 times.
✓ Branch 1 taken 212 times.
|
1669675 | if (storedSize > 0) { |
| 334 | 1669463 | const FwSizeType offset = bestIndex * this->m_handle.m_messageSize; | |
| 335 | static_cast<void>( | ||
| 336 |
2/4✗ Branch 5 not taken.
✓ Branch 6 taken 1669156 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 1669156 times.
|
1669548 | ::memcpy(destination, this->m_handle.m_data + offset, static_cast<size_t>(storedSize))); |
| 337 | } | ||
| 338 | 1669368 | actualSize = storedSize; | |
| 339 | 3338721 | priority = slot.m_priority.load(std::memory_order_relaxed); | |
| 340 | // Decrement the occupancy count *before* releasing the slot to FREE. This keeps | ||
| 341 | // m_count (and therefore the high-water mark) at or below the queue depth: a producer | ||
| 342 | // can only re-claim and re-count this slot after observing FREE, which follows the | ||
| 343 | // decrement. | ||
| 344 | 1669236 | static_cast<void>(this->m_handle.m_count.fetch_sub(1, std::memory_order_acq_rel)); | |
| 345 | 1669472 | slot.m_stateTag.store(packStateTag(LOCKLESS_SLOT_FREE, tagOf(desired) + 1), std::memory_order_release); | |
| 346 | 1668583 | return QueueInterface::Status::OP_OK; | |
| 347 | } | ||
| 348 | // CAS failed — another consumer claimed this slot; yield and rescan. | ||
| 349 |
2/2✓ Branch 0 taken 111 times.
✓ Branch 1 taken 1434842 times.
|
1434953 | if (blocking) { |
| 350 |
2/2✓ Branch 2 taken 111 times.
✓ Branch 6 taken 111 times.
|
111 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, 0))); |
| 351 | } | ||
| 352 | } | ||
| 353 | 879616 | return QueueInterface::Status::EMPTY; | |
| 354 | } | ||
| 355 | |||
| 356 | 69921 | FwSizeType LocklessPriorityQueue::getMessagesAvailable() const { | |
| 357 | 139842 | return static_cast<FwSizeType>(this->m_handle.m_available.load(std::memory_order_acquire)); | |
| 358 | } | ||
| 359 | |||
| 360 | 67231 | FwSizeType LocklessPriorityQueue::getMessageHighWaterMark() const { | |
| 361 | 134462 | return static_cast<FwSizeType>(this->m_handle.m_highMark.load(std::memory_order_acquire)); | |
| 362 | } | ||
| 363 | |||
| 364 | ✗ | QueueHandle* LocklessPriorityQueue::getHandle() { | |
| 365 | ✗ | return &this->m_handle; | |
| 366 | } | ||
| 367 | |||
| 368 | } // namespace Generic | ||
| 369 | } // namespace Os | ||
| 370 |