| 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 | 108307256 | constexpr LocklessStateTagType stateOf(LocklessStateTagType packed) { | |
| 24 | 108307256 | return packed & LocklessSlot::STATE_MASK; | |
| 25 | } | ||
| 26 | |||
| 27 | //! Extract the tag portion of a packed state-tag word. | ||
| 28 | 8325760 | constexpr LocklessStateTagType tagOf(LocklessStateTagType packed) { | |
| 29 | 8325760 | return packed >> LocklessSlot::STATE_BITS; | |
| 30 | } | ||
| 31 | |||
| 32 | //! Pack a (state, tag) pair into a single word. | ||
| 33 | 8315912 | constexpr LocklessStateTagType packStateTag(LocklessStateTagType state, LocklessStateTagType tag) { | |
| 34 | 8315912 | return (tag << LocklessSlot::STATE_BITS) | (state & LocklessSlot::STATE_MASK); | |
| 35 | } | ||
| 36 | |||
| 37 | } // namespace | ||
| 38 | |||
| 39 | 49733250 | bool LocklessPriorityQueue::isCandidatePreferred(FwQueuePriorityType candidatePriority, | |
| 40 | U32 candidateSequence, | ||
| 41 | FwQueuePriorityType bestPriority, | ||
| 42 | U32 bestSequence) { | ||
| 43 | 49733250 | bool preferred = false; | |
| 44 |
2/2✓ Branch 0 taken 4007747 times.
✓ Branch 1 taken 45725503 times.
|
49733250 | if (candidatePriority > bestPriority) { |
| 45 | 4007747 | preferred = true; | |
| 46 |
2/2✓ Branch 0 taken 12680734 times.
✓ Branch 1 taken 33044769 times.
|
45725503 | } 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 | 12680734 | const U32 difference = candidateSequence - bestSequence; | |
| 50 | 12680734 | const U32 topBit = static_cast<U32>(1) << (std::numeric_limits<U32>::digits - 1); | |
| 51 | 12680734 | preferred = (difference & topBit) != 0; | |
| 52 | } | ||
| 53 | 49733250 | return preferred; | |
| 54 | } | ||
| 55 | |||
| 56 | 2343 | LocklessSlot::LocklessSlot() | |
| 57 | 2343 | : m_stateTag(packStateTag(LOCKLESS_SLOT_FREE, 0)), m_sequence(0), m_size(0), m_priority(0) {} | |
| 58 | |||
| 59 | 205 | LocklessPriorityQueueHandle::LocklessPriorityQueueHandle() | |
| 60 | : QueueHandle(), | ||
| 61 | 205 | m_slots(nullptr), | |
| 62 | 205 | m_slotsAllocation(nullptr), | |
| 63 | 205 | m_data(nullptr), | |
| 64 | 205 | m_depth(0), | |
| 65 | 205 | m_messageSize(0), | |
| 66 | 205 | m_sequence(0), | |
| 67 | 205 | m_count(0), | |
| 68 | 205 | m_available(0), | |
| 69 | 205 | m_highMark(0), | |
| 70 | 205 | m_id(0) {} | |
| 71 | |||
| 72 | 410 | 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 | 410 | } | |
| 78 | |||
| 79 | 194 | 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 | 194 | FW_ASSERT(this->m_handle.m_slots == nullptr); | |
| 88 | 194 | 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 | 194 | std::atomic<LocklessStateTagType> probe(0); | |
| 95 | 194 | FW_ASSERT(probe.is_lock_free()); | |
| 96 | 194 | std::atomic<FwQueuePriorityType> priorityProbe(0); | |
| 97 | 194 | FW_ASSERT(priorityProbe.is_lock_free()); | |
| 98 | 194 | std::atomic<U32> counterProbe(0); | |
| 99 | 194 | FW_ASSERT(counterProbe.is_lock_free()); | |
| 100 | 194 | FW_ASSERT(depth > 0); | |
| 101 | 194 | FW_ASSERT(messageSize > 0); | |
| 102 | |||
| 103 | // Guard the multiplications used to compute allocation sizes against overflow. | ||
| 104 | 194 | const FwSizeType maxSize = std::numeric_limits<FwSizeType>::max(); | |
| 105 | 194 | FW_ASSERT(depth <= (maxSize / sizeof(LocklessSlot))); | |
| 106 | 194 | 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 | 194 | FW_ASSERT(depth < (std::numeric_limits<U32>::max() / 2)); | |
| 112 | |||
| 113 |
3/3✓ Branch 1 taken 194 times.
✓ Branch 5 taken 194 times.
✓ Branch 8 taken 194 times.
|
388 | Fw::MemAllocator& allocator = Fw::MemAllocatorRegistry::getInstance().getAnAllocator( |
| 114 | 194 | Fw::MemoryAllocation::MemoryAllocatorType::OS_GENERIC_PRIORITY_QUEUE); | |
| 115 | |||
| 116 | 194 | LocklessSlot* slots = nullptr; | |
| 117 | 194 | void* slotsAllocation = nullptr; | |
| 118 | 194 | U8* data = nullptr; | |
| 119 | 194 | 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 | 194 | FW_ASSERT(depth * sizeof(LocklessSlot) <= (maxSize - alignof(LocklessSlot))); | |
| 124 | 194 | FwSizeType slotBytesRequested = (depth * sizeof(LocklessSlot)) + alignof(LocklessSlot); | |
| 125 | 194 | FwSizeType slotBytesAllocated = slotBytesRequested; | |
| 126 |
1/1✓ Branch 4 taken 194 times.
|
194 | slotsAllocation = allocator.allocate(id, slotBytesAllocated, alignof(LocklessSlot)); |
| 127 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 194 times.
|
194 | if (slotsAllocation == nullptr) { |
| 128 | ✗ | status = QueueInterface::Status::ALLOCATION_FAILED; | |
| 129 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 194 times.
|
194 | } else if (slotBytesAllocated < slotBytesRequested) { |
| 130 | ✗ | allocator.deallocate(id, slotsAllocation); | |
| 131 | ✗ | status = QueueInterface::Status::ALLOCATION_FAILED; | |
| 132 | } else { | ||
| 133 | 194 | const PlatformPointerCastType base = reinterpret_cast<PlatformPointerCastType>(slotsAllocation); | |
| 134 | 194 | const PlatformPointerCastType aligned = | |
| 135 | 194 | (base + (alignof(LocklessSlot) - 1)) & ~static_cast<PlatformPointerCastType>(alignof(LocklessSlot) - 1); | |
| 136 | 194 | const FwSizeType offset = static_cast<FwSizeType>(aligned - base); | |
| 137 |
1/1✓ Branch 4 taken 194 times.
|
194 | 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 194 times.
✗ Branch 1 not taken.
|
194 | if (status == QueueInterface::Status::OP_OK) { |
| 143 | 194 | FwSizeType dataBytesRequested = depth * messageSize; | |
| 144 | 194 | FwSizeType dataBytesAllocated = dataBytesRequested; | |
| 145 |
1/1✓ Branch 4 taken 194 times.
|
194 | void* dataAllocation = allocator.allocate(id, dataBytesAllocated, alignof(U8)); |
| 146 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 194 times.
|
194 | 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 194 times.
|
194 | } 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 | 194 | data = static_cast<U8*>(dataAllocation); | |
| 157 | } | ||
| 158 | } | ||
| 159 | |||
| 160 | // Publish the configured handle once both allocations succeeded. | ||
| 161 |
1/2✓ Branch 0 taken 194 times.
✗ Branch 1 not taken.
|
194 | if (status == QueueInterface::Status::OP_OK) { |
| 162 | 194 | this->m_handle.m_id = id; | |
| 163 | 194 | this->m_handle.m_messageSize = messageSize; | |
| 164 | 194 | this->m_handle.m_depth = depth; | |
| 165 | 194 | this->m_handle.m_slots = slots; | |
| 166 | 194 | this->m_handle.m_slotsAllocation = slotsAllocation; | |
| 167 | 194 | this->m_handle.m_data = data; | |
| 168 | 194 | this->m_handle.m_sequence.store(0, std::memory_order_relaxed); | |
| 169 | 194 | this->m_handle.m_count.store(0, std::memory_order_relaxed); | |
| 170 | 194 | this->m_handle.m_available.store(0, std::memory_order_relaxed); | |
| 171 | 194 | this->m_handle.m_highMark.store(0, std::memory_order_relaxed); | |
| 172 | } | ||
| 173 | 388 | return status; | |
| 174 | } | ||
| 175 | |||
| 176 | 205 | void LocklessPriorityQueue::teardown() { | |
| 177 |
2/2✓ Branch 4 taken 194 times.
✓ Branch 5 taken 11 times.
|
205 | if (this->m_handle.m_slots != nullptr) { |
| 178 |
2/2✓ Branch 3 taken 194 times.
✓ Branch 6 taken 194 times.
|
388 | Fw::MemAllocator& allocator = Fw::MemAllocatorRegistry::getInstance().getAnAllocator( |
| 179 | 194 | Fw::MemoryAllocation::MemoryAllocatorType::OS_GENERIC_PRIORITY_QUEUE); | |
| 180 | 194 | Fw::arrayPlacementDestruct<LocklessSlot>(this->m_handle.m_slots, this->m_handle.m_depth); | |
| 181 | 194 | allocator.deallocate(this->m_handle.m_id, this->m_handle.m_slotsAllocation); | |
| 182 |
1/2✓ Branch 4 taken 194 times.
✗ Branch 5 not taken.
|
194 | if (this->m_handle.m_data != nullptr) { |
| 183 | 194 | allocator.deallocate(this->m_handle.m_id, this->m_handle.m_data); | |
| 184 | } | ||
| 185 | 194 | this->m_handle.m_slots = nullptr; | |
| 186 | 194 | this->m_handle.m_slotsAllocation = nullptr; | |
| 187 | 194 | this->m_handle.m_data = nullptr; | |
| 188 | 194 | this->m_handle.m_depth = 0; | |
| 189 | 194 | this->m_handle.m_messageSize = 0; | |
| 190 | 194 | this->m_handle.m_count.store(0, std::memory_order_relaxed); | |
| 191 | 194 | this->m_handle.m_available.store(0, std::memory_order_relaxed); | |
| 192 | 194 | this->m_handle.m_highMark.store(0, std::memory_order_relaxed); | |
| 193 | 194 | this->m_handle.m_sequence.store(0, std::memory_order_relaxed); | |
| 194 | } | ||
| 195 | 205 | } | |
| 196 | |||
| 197 | 2110528 | 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 | 2110528 | FW_ASSERT(this->m_handle.m_slots != nullptr); | |
| 204 | 2108901 | FW_ASSERT(this->m_handle.m_data != nullptr); | |
| 205 | 2108644 | FW_ASSERT(buffer != nullptr); | |
| 206 | |||
| 207 | // Reject oversized messages without touching the queue. | ||
| 208 |
2/2✓ Branch 4 taken 1 times.
✓ Branch 5 taken 2110763 times.
|
2108644 | if (size > this->m_handle.m_messageSize) { |
| 209 | 1 | return QueueInterface::Status::SIZE_MISMATCH; | |
| 210 | } | ||
| 211 | |||
| 212 | 2110763 | const FwSizeType depth = this->m_handle.m_depth; | |
| 213 | 2103137 | 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 9671 times.
✓ Branch 1 taken 4322233 times.
✓ Branch 2 taken 3955733 times.
✓ Branch 3 taken 366500 times.
|
4331904 | for (FwSizeType pass = 0; blocking || (pass < MAX_RETRY_PASSES); pass++) { |
| 217 |
2/2✓ Branch 0 taken 58286217 times.
✓ Branch 1 taken 2231997 times.
|
60518214 | for (FwSizeType i = 0; i < depth; i++) { |
| 218 | 58286217 | LocklessSlot& slot = this->m_handle.m_slots[i]; | |
| 219 | 58308407 | LocklessStateTagType packed = slot.m_stateTag.load(std::memory_order_acquire); | |
| 220 |
2/2✓ Branch 1 taken 56609497 times.
✓ Branch 2 taken 1693378 times.
|
58586996 | if (stateOf(packed) != LOCKLESS_SLOT_FREE) { |
| 221 | 56609497 | continue; | |
| 222 | } | ||
| 223 | 1693378 | const LocklessStateTagType desired = packStateTag(LOCKLESS_SLOT_WRITING, tagOf(packed) + 1); | |
| 224 |
2/2✓ Branch 2 taken 1749285 times.
✓ Branch 3 taken 28433 times.
|
3557179 | if (slot.m_stateTag.compare_exchange_strong(packed, desired, std::memory_order_acq_rel, |
| 225 | std::memory_order_relaxed)) { | ||
| 226 |
1/2✓ Branch 0 taken 1749325 times.
✗ Branch 1 not taken.
|
1749285 | if (size > 0) { |
| 227 | 1749325 | const FwSizeType offset = i * this->m_handle.m_messageSize; | |
| 228 |
2/4✗ Branch 5 not taken.
✓ Branch 6 taken 1748740 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 1748740 times.
|
1749078 | static_cast<void>(::memcpy(this->m_handle.m_data + offset, buffer, static_cast<size_t>(size))); |
| 229 | } | ||
| 230 | 1748700 | slot.m_size = size; | |
| 231 | 1748864 | slot.m_priority.store(priority, std::memory_order_relaxed); | |
| 232 | 3497456 | slot.m_sequence.store(this->m_handle.m_sequence.fetch_add(1, std::memory_order_relaxed), | |
| 233 | std::memory_order_relaxed); | ||
| 234 | |||
| 235 | // Increment both counters *before* publishing READY. This guarantees that a | ||
| 236 | // consumer's decrement (which can only follow a READY observation) never races | ||
| 237 | // ahead of the matching increment, so neither m_count nor m_available can | ||
| 238 | // transiently underflow or wrap to a huge value (fixes GitHub issue #6055). | ||
| 239 | 1747744 | const U32 nextCount = this->m_handle.m_count.fetch_add(1, std::memory_order_acq_rel) + 1; | |
| 240 | 1748659 | static_cast<void>(this->m_handle.m_available.fetch_add(1, std::memory_order_acq_rel)); | |
| 241 | |||
| 242 | 1748624 | slot.m_stateTag.store(packStateTag(LOCKLESS_SLOT_READY, tagOf(desired) + 1), std::memory_order_release); | |
| 243 | |||
| 244 | // Raise the high-water mark after publication so the message is never invisible | ||
| 245 | // while the producer runs this loop. Each strong-CAS failure strictly raises | ||
| 246 | // prevMark (mark only increases, capped at depth), so at most depth iterations run. | ||
| 247 | 1746742 | U32 prevMark = this->m_handle.m_highMark.load(std::memory_order_relaxed); | |
| 248 |
4/4✓ Branch 0 taken 1746612 times.
✓ Branch 1 taken 76 times.
✓ Branch 2 taken 2076 times.
✓ Branch 3 taken 1744536 times.
|
1746688 | for (FwSizeType markPass = 0; (markPass < depth) && (nextCount > prevMark); markPass++) { |
| 249 |
2/2✓ Branch 3 taken 2075 times.
✓ Branch 4 taken 1 times.
|
4152 | if (this->m_handle.m_highMark.compare_exchange_strong( |
| 250 | prevMark, nextCount, std::memory_order_relaxed, std::memory_order_relaxed)) { | ||
| 251 | 2075 | break; | |
| 252 | } | ||
| 253 | } | ||
| 254 | 1746687 | return QueueInterface::Status::OP_OK; | |
| 255 | } | ||
| 256 | } | ||
| 257 | // No free slot found this pass. Blocking callers back off; non-blocking callers retry. | ||
| 258 |
2/2✓ Branch 0 taken 1527 times.
✓ Branch 1 taken 2230470 times.
|
2231997 | if (blocking) { |
| 259 |
2/2✓ Branch 2 taken 1521 times.
✓ Branch 6 taken 1536 times.
|
1527 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, LOCKLESS_QUEUE_BLOCKING_BACKOFF_US))); |
| 260 | } | ||
| 261 | } | ||
| 262 | 366500 | return QueueInterface::Status::FULL; | |
| 263 | } | ||
| 264 | |||
| 265 | 2610699 | QueueInterface::Status LocklessPriorityQueue::receive(U8* destination, | |
| 266 | FwSizeType capacity, | ||
| 267 | QueueInterface::BlockingType blockType, | ||
| 268 | FwSizeType& actualSize, | ||
| 269 | FwQueuePriorityType& priority) { | ||
| 270 | // Programming-error checks. These mirror the assertions in send(). | ||
| 271 | 2610699 | FW_ASSERT(this->m_handle.m_slots != nullptr); | |
| 272 | 2604173 | FW_ASSERT(this->m_handle.m_data != nullptr); | |
| 273 | 2596860 | FW_ASSERT(destination != nullptr); | |
| 274 | |||
| 275 | // Reject undersized buffers before claiming a slot so the message stays queued. | ||
| 276 |
2/2✓ Branch 4 taken 1 times.
✓ Branch 5 taken 2625755 times.
|
2596860 | if (capacity < this->m_handle.m_messageSize) { |
| 277 | 1 | return QueueInterface::Status::SIZE_MISMATCH; | |
| 278 | } | ||
| 279 | |||
| 280 | 2625755 | const FwSizeType depth = this->m_handle.m_depth; | |
| 281 | 2605616 | const bool blocking = (blockType == QueueInterface::BlockingType::BLOCKING); | |
| 282 | |||
| 283 | // Bounded for non-blocking, unbounded for blocking (explicit user contract). | ||
| 284 |
4/4✓ Branch 0 taken 91710 times.
✓ Branch 1 taken 6730406 times.
✓ Branch 2 taken 5834016 times.
✓ Branch 3 taken 896390 times.
|
6822116 | for (FwSizeType pass = 0; blocking || (pass < MAX_RETRY_PASSES); pass++) { |
| 285 | // Fast path: skip the O(depth) scan while no message is receivable. | ||
| 286 |
2/2✓ Branch 3 taken 2774005 times.
✓ Branch 4 taken 3149496 times.
|
11849227 | if (this->m_handle.m_available.load(std::memory_order_acquire) == 0) { |
| 287 |
2/2✓ Branch 0 taken 1394 times.
✓ Branch 1 taken 2772611 times.
|
2774005 | if (blocking) { |
| 288 |
2/2✓ Branch 2 taken 1392 times.
✓ Branch 6 taken 1392 times.
|
1394 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, LOCKLESS_QUEUE_BLOCKING_BACKOFF_US))); |
| 289 | } | ||
| 290 | 2757735 | continue; | |
| 291 | } | ||
| 292 | 3149496 | FwSizeType bestIndex = depth; | |
| 293 | 3149496 | FwQueuePriorityType bestPriority = FwQueuePriorityType(); | |
| 294 | 3149496 | U32 bestSequence = 0; | |
| 295 | 3149496 | LocklessStateTagType bestPacked = 0; | |
| 296 | |||
| 297 |
2/2✓ Branch 0 taken 66438112 times.
✓ Branch 1 taken 4718148 times.
|
71156260 | for (FwSizeType i = 0; i < depth; i++) { |
| 298 | 66438112 | LocklessSlot& slot = this->m_handle.m_slots[i]; | |
| 299 | 66709556 | LocklessStateTagType packed = slot.m_stateTag.load(std::memory_order_acquire); | |
| 300 |
2/2✓ Branch 1 taken 15205238 times.
✓ Branch 2 taken 50746076 times.
|
66932148 | if (stateOf(packed) != LOCKLESS_SLOT_READY) { |
| 301 | 15205238 | continue; | |
| 302 | } | ||
| 303 | 50746076 | const FwQueuePriorityType candidatePriority = slot.m_priority.load(std::memory_order_relaxed); | |
| 304 | 54138140 | const U32 candidateSequence = slot.m_sequence.load(std::memory_order_relaxed); | |
| 305 | // Recheck: if the state-tag changed, the relaxed priority/sequence reads above may | ||
| 306 | // belong to a recycled slot; discard the candidate and keep scanning. | ||
| 307 | 52313224 | const LocklessStateTagType packedRecheck = slot.m_stateTag.load(std::memory_order_acquire); | |
| 308 |
2/2✓ Branch 0 taken 88773 times.
✓ Branch 1 taken 53339607 times.
|
53428380 | if (packed != packedRecheck) { |
| 309 | 88773 | continue; | |
| 310 | } | ||
| 311 |
6/6✓ Branch 0 taken 50558628 times.
✓ Branch 1 taken 2780979 times.
✓ Branch 2 taken 6859734 times.
✓ Branch 3 taken 43055099 times.
✓ Branch 4 taken 9657552 times.
✓ Branch 5 taken 43038260 times.
|
103254440 | if ((bestIndex == depth) || |
| 312 | 50558628 | isCandidatePreferred(candidatePriority, candidateSequence, bestPriority, bestSequence)) { | |
| 313 | 9657552 | bestIndex = i; | |
| 314 | 9657552 | bestPriority = candidatePriority; | |
| 315 | 9657552 | bestSequence = candidateSequence; | |
| 316 | 9657552 | bestPacked = packed; | |
| 317 | } | ||
| 318 | } | ||
| 319 | |||
| 320 |
2/2✓ Branch 0 taken 42755 times.
✓ Branch 1 taken 4675393 times.
|
4718148 | if (bestIndex == depth) { |
| 321 |
2/2✓ Branch 0 taken 33 times.
✓ Branch 1 taken 42722 times.
|
42755 | if (blocking) { |
| 322 |
2/2✓ Branch 2 taken 33 times.
✓ Branch 6 taken 33 times.
|
33 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, LOCKLESS_QUEUE_BLOCKING_BACKOFF_US))); |
| 323 | } | ||
| 324 | 42553 | continue; | |
| 325 | } | ||
| 326 | |||
| 327 | 4675393 | const LocklessStateTagType desired = packStateTag(LOCKLESS_SLOT_READING, tagOf(bestPacked) + 1); | |
| 328 |
2/2✓ Branch 7 taken 1749261 times.
✓ Branch 8 taken 1374011 times.
|
6243162 | if (this->m_handle.m_slots[bestIndex].m_stateTag.compare_exchange_strong( |
| 329 | bestPacked, desired, std::memory_order_acq_rel, std::memory_order_relaxed)) { | ||
| 330 | // Decrement the receivable count at the successful READY->READING claim. Because | ||
| 331 | // m_available is incremented before READY is published, it is always >= the number | ||
| 332 | // of genuinely READY slots, so this decrement can never underflow. | ||
| 333 | 1749261 | static_cast<void>(this->m_handle.m_available.fetch_sub(1, std::memory_order_acq_rel)); | |
| 334 | 1749240 | LocklessSlot& slot = this->m_handle.m_slots[bestIndex]; | |
| 335 | 1749190 | const FwSizeType storedSize = slot.m_size; | |
| 336 | 1749198 | FW_ASSERT(storedSize <= capacity); | |
| 337 |
1/2✓ Branch 0 taken 1749207 times.
✗ Branch 1 not taken.
|
1749198 | if (storedSize > 0) { |
| 338 | 1749207 | const FwSizeType offset = bestIndex * this->m_handle.m_messageSize; | |
| 339 | static_cast<void>( | ||
| 340 |
2/4✗ Branch 5 not taken.
✓ Branch 6 taken 1749140 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 1749140 times.
|
1749201 | ::memcpy(destination, this->m_handle.m_data + offset, static_cast<size_t>(storedSize))); |
| 341 | } | ||
| 342 | 1749131 | actualSize = storedSize; | |
| 343 | 3498011 | priority = slot.m_priority.load(std::memory_order_relaxed); | |
| 344 | // Decrement the occupancy count *before* releasing the slot to FREE. This keeps | ||
| 345 | // m_count (and therefore the high-water mark) at or below the queue depth: a producer | ||
| 346 | // can only re-claim and re-count this slot after observing FREE, which follows the | ||
| 347 | // decrement. | ||
| 348 | 1748890 | static_cast<void>(this->m_handle.m_count.fetch_sub(1, std::memory_order_acq_rel)); | |
| 349 | 1749087 | slot.m_stateTag.store(packStateTag(LOCKLESS_SLOT_FREE, tagOf(desired) + 1), std::memory_order_release); | |
| 350 | 1748567 | return QueueInterface::Status::OP_OK; | |
| 351 | } | ||
| 352 | // CAS failed — another consumer claimed this slot; yield and rescan. | ||
| 353 |
2/2✓ Branch 0 taken 137 times.
✓ Branch 1 taken 1373874 times.
|
1374011 | if (blocking) { |
| 354 |
2/2✓ Branch 2 taken 138 times.
✓ Branch 6 taken 138 times.
|
137 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, 0))); |
| 355 | } | ||
| 356 | } | ||
| 357 | 896390 | return QueueInterface::Status::EMPTY; | |
| 358 | } | ||
| 359 | |||
| 360 | 379019 | FwSizeType LocklessPriorityQueue::getMessagesAvailable() const { | |
| 361 | 758038 | return static_cast<FwSizeType>(this->m_handle.m_available.load(std::memory_order_acquire)); | |
| 362 | } | ||
| 363 | |||
| 364 | 352923 | FwSizeType LocklessPriorityQueue::getMessageHighWaterMark() const { | |
| 365 | 705846 | return static_cast<FwSizeType>(this->m_handle.m_highMark.load(std::memory_order_acquire)); | |
| 366 | } | ||
| 367 | |||
| 368 | ✗ | QueueHandle* LocklessPriorityQueue::getHandle() { | |
| 369 | ✗ | return &this->m_handle; | |
| 370 | } | ||
| 371 | |||
| 372 | } // namespace Generic | ||
| 373 | } // namespace Os | ||
| 374 |