| 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 | 74708844 | constexpr LocklessStateTagType stateOf(LocklessStateTagType packed) { | |
| 24 | 74708844 | return packed & LocklessSlot::STATE_MASK; | |
| 25 | } | ||
| 26 | |||
| 27 | //! Extract the tag portion of a packed state-tag word. | ||
| 28 | 8034191 | constexpr LocklessStateTagType tagOf(LocklessStateTagType packed) { | |
| 29 | 8034191 | return packed >> LocklessSlot::STATE_BITS; | |
| 30 | } | ||
| 31 | |||
| 32 | //! Pack a (state, tag) pair into a single word. | ||
| 33 | 8055997 | constexpr LocklessStateTagType packStateTag(LocklessStateTagType state, LocklessStateTagType tag) { | |
| 34 | 8055997 | return (tag << LocklessSlot::STATE_BITS) | (state & LocklessSlot::STATE_MASK); | |
| 35 | } | ||
| 36 | |||
| 37 | } // namespace | ||
| 38 | |||
| 39 | 34753409 | bool LocklessPriorityQueue::isCandidatePreferred(FwQueuePriorityType candidatePriority, | |
| 40 | U32 candidateSequence, | ||
| 41 | FwQueuePriorityType bestPriority, | ||
| 42 | U32 bestSequence) { | ||
| 43 | 34753409 | bool preferred = false; | |
| 44 |
2/2✓ Branch 0 taken 4448759 times.
✓ Branch 1 taken 30304650 times.
|
34753409 | if (candidatePriority > bestPriority) { |
| 45 | 4448759 | preferred = true; | |
| 46 |
2/2✓ Branch 0 taken 10261556 times.
✓ Branch 1 taken 20043094 times.
|
30304650 | } 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 | 10261556 | const U32 difference = candidateSequence - bestSequence; | |
| 50 | 10261556 | const U32 topBit = static_cast<U32>(1) << (std::numeric_limits<U32>::digits - 1); | |
| 51 | 10261556 | preferred = (difference & topBit) != 0; | |
| 52 | } | ||
| 53 | 34753409 | return preferred; | |
| 54 | } | ||
| 55 | |||
| 56 | 2246 | LocklessSlot::LocklessSlot() | |
| 57 | 2246 | : 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 | 2153995 | 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 | 2153995 | FW_ASSERT(this->m_handle.m_slots != nullptr); | |
| 204 | 2157904 | FW_ASSERT(this->m_handle.m_data != nullptr); | |
| 205 | 2157925 | FW_ASSERT(buffer != nullptr); | |
| 206 | |||
| 207 | // Reject oversized messages without touching the queue. | ||
| 208 |
2/2✓ Branch 4 taken 1 times.
✓ Branch 5 taken 2157951 times.
|
2157925 | if (size > this->m_handle.m_messageSize) { |
| 209 | 1 | return QueueInterface::Status::SIZE_MISMATCH; | |
| 210 | } | ||
| 211 | |||
| 212 | 2157951 | const FwSizeType depth = this->m_handle.m_depth; | |
| 213 | 2152999 | 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 24753 times.
✓ Branch 1 taken 4479855 times.
✓ Branch 2 taken 4030809 times.
✓ Branch 3 taken 449046 times.
|
4504608 | for (FwSizeType pass = 0; blocking || (pass < MAX_RETRY_PASSES); pass++) { |
| 217 |
2/2✓ Branch 0 taken 54751778 times.
✓ Branch 1 taken 2489698 times.
|
57241476 | for (FwSizeType i = 0; i < depth; i++) { |
| 218 | 54751778 | LocklessSlot& slot = this->m_handle.m_slots[i]; | |
| 219 | 54752994 | LocklessStateTagType packed = slot.m_stateTag.load(std::memory_order_acquire); | |
| 220 |
2/2✓ Branch 1 taken 53118070 times.
✓ Branch 2 taken 1506892 times.
|
54667396 | if (stateOf(packed) != LOCKLESS_SLOT_FREE) { |
| 221 | 53118070 | continue; | |
| 222 | } | ||
| 223 | 1506892 | const LocklessStateTagType desired = packStateTag(LOCKLESS_SLOT_WRITING, tagOf(packed) + 1); | |
| 224 |
2/2✓ Branch 2 taken 1709103 times.
✓ Branch 3 taken 41327 times.
|
3501488 | 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 1708397 times.
✓ Branch 1 taken 706 times.
|
1709103 | if (size > 0) { |
| 227 | 1708397 | const FwSizeType offset = i * this->m_handle.m_messageSize; | |
| 228 |
2/4✗ Branch 5 not taken.
✓ Branch 6 taken 1707576 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 1707576 times.
|
1708799 | static_cast<void>(::memcpy(this->m_handle.m_data + offset, buffer, static_cast<size_t>(size))); |
| 229 | } | ||
| 230 | 1708282 | slot.m_size = size; | |
| 231 | 1707621 | slot.m_priority.store(priority, std::memory_order_relaxed); | |
| 232 | 3416476 | 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 | 1706911 | const U32 nextCount = this->m_handle.m_count.fetch_add(1, std::memory_order_acq_rel) + 1; | |
| 239 | |||
| 240 | 1709053 | 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 | 1704652 | 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 | 1708889 | U32 prevMark = this->m_handle.m_highMark.load(std::memory_order_relaxed); | |
| 250 |
4/4✓ Branch 0 taken 1704183 times.
✓ Branch 1 taken 1378 times.
✓ Branch 2 taken 1993 times.
✓ Branch 3 taken 1702190 times.
|
1705561 | for (FwSizeType markPass = 0; (markPass < depth) && (nextCount > prevMark); markPass++) { |
| 251 |
2/2✓ Branch 3 taken 1990 times.
✓ Branch 4 taken 3 times.
|
3986 | if (this->m_handle.m_highMark.compare_exchange_strong( |
| 252 | prevMark, nextCount, std::memory_order_relaxed, std::memory_order_relaxed)) { | ||
| 253 | 1990 | break; | |
| 254 | } | ||
| 255 | } | ||
| 256 | 1705558 | 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 1773 times.
✓ Branch 1 taken 2487925 times.
|
2489698 | if (blocking) { |
| 261 |
2/2✓ Branch 2 taken 1782 times.
✓ Branch 6 taken 1741 times.
|
1773 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, LOCKLESS_QUEUE_BLOCKING_BACKOFF_US))); |
| 262 | } | ||
| 263 | } | ||
| 264 | 449046 | return QueueInterface::Status::FULL; | |
| 265 | } | ||
| 266 | |||
| 267 | 2807778 | 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 | 2807778 | FW_ASSERT(this->m_handle.m_slots != nullptr); | |
| 274 | 2836151 | FW_ASSERT(this->m_handle.m_data != nullptr); | |
| 275 | 2857701 | FW_ASSERT(destination != nullptr); | |
| 276 | |||
| 277 | 2857701 | const FwSizeType depth = this->m_handle.m_depth; | |
| 278 | 2857715 | 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 131127 times.
✓ Branch 1 taken 7773431 times.
✓ Branch 2 taken 6694797 times.
✓ Branch 3 taken 1078634 times.
|
7904558 | 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 3406068 times.
✓ Branch 4 taken 3297485 times.
|
13529477 | if (this->m_handle.m_available.load(std::memory_order_acquire) == 0) { |
| 284 |
2/2✓ Branch 0 taken 1694 times.
✓ Branch 1 taken 3404374 times.
|
3406068 | if (blocking) { |
| 285 |
2/2✓ Branch 2 taken 1695 times.
✓ Branch 6 taken 1670 times.
|
1694 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, LOCKLESS_QUEUE_BLOCKING_BACKOFF_US))); |
| 286 | } | ||
| 287 | 3397559 | continue; | |
| 288 | } | ||
| 289 | 3297485 | FwSizeType bestIndex = depth; | |
| 290 | 3297485 | FwQueuePriorityType bestPriority = FwQueuePriorityType(); | |
| 291 | 3297485 | U32 bestSequence = 0; | |
| 292 | 3297485 | LocklessStateTagType bestPacked = 0; | |
| 293 | |||
| 294 |
2/2✓ Branch 0 taken 46507867 times.
✓ Branch 1 taken 6869704 times.
|
53377571 | for (FwSizeType i = 0; i < depth; i++) { |
| 295 | 46507867 | LocklessSlot& slot = this->m_handle.m_slots[i]; | |
| 296 | 47100901 | LocklessStateTagType packed = slot.m_stateTag.load(std::memory_order_acquire); | |
| 297 |
2/2✓ Branch 1 taken 13494050 times.
✓ Branch 2 taken 32568841 times.
|
45904680 | if (stateOf(packed) != LOCKLESS_SLOT_READY) { |
| 298 | 13494050 | continue; | |
| 299 | } | ||
| 300 | 32568841 | const FwQueuePriorityType candidatePriority = slot.m_priority.load(std::memory_order_relaxed); | |
| 301 | 36330478 | 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 | 36059885 | const LocklessStateTagType packedRecheck = slot.m_stateTag.load(std::memory_order_acquire); | |
| 305 |
2/2✓ Branch 0 taken 113107 times.
✓ Branch 1 taken 35578065 times.
|
35691172 | if (packed != packedRecheck) { |
| 306 | 113107 | continue; | |
| 307 | } | ||
| 308 |
6/6✓ Branch 0 taken 34329262 times.
✓ Branch 1 taken 1248803 times.
✓ Branch 2 taken 8272304 times.
✓ Branch 3 taken 26922337 times.
✓ Branch 4 taken 9532134 times.
✓ Branch 5 taken 26911310 times.
|
70772706 | if ((bestIndex == depth) || |
| 309 | 34329262 | isCandidatePreferred(candidatePriority, candidateSequence, bestPriority, bestSequence)) { | |
| 310 | 9532134 | bestIndex = i; | |
| 311 | 9532134 | bestPriority = candidatePriority; | |
| 312 | 9532134 | bestSequence = candidateSequence; | |
| 313 | 9532134 | bestPacked = packed; | |
| 314 | } | ||
| 315 | } | ||
| 316 | |||
| 317 |
2/2✓ Branch 0 taken 51596 times.
✓ Branch 1 taken 6818108 times.
|
6869704 | if (bestIndex == depth) { |
| 318 |
2/2✓ Branch 0 taken 15 times.
✓ Branch 1 taken 51581 times.
|
51596 | if (blocking) { |
| 319 |
2/2✓ Branch 2 taken 15 times.
✓ Branch 6 taken 15 times.
|
15 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, LOCKLESS_QUEUE_BLOCKING_BACKOFF_US))); |
| 320 | } | ||
| 321 | 51528 | continue; | |
| 322 | } | ||
| 323 | |||
| 324 | 6818108 | const LocklessStateTagType desired = packStateTag(LOCKLESS_SLOT_READING, tagOf(bestPacked) + 1); | |
| 325 |
2/2✓ Branch 7 taken 1709408 times.
✓ Branch 8 taken 1422048 times.
|
6256881 | 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 | 1709408 | static_cast<void>(this->m_handle.m_available.fetch_sub(1, std::memory_order_acq_rel)); | |
| 330 | 1709410 | LocklessSlot& slot = this->m_handle.m_slots[bestIndex]; | |
| 331 | 1709353 | const FwSizeType storedSize = slot.m_size; | |
| 332 | 1709352 | FW_ASSERT(storedSize <= capacity); | |
| 333 |
2/2✓ Branch 0 taken 1709081 times.
✓ Branch 1 taken 271 times.
|
1709352 | if (storedSize > 0) { |
| 334 | 1709081 | const FwSizeType offset = bestIndex * this->m_handle.m_messageSize; | |
| 335 | static_cast<void>( | ||
| 336 |
2/4✗ Branch 5 not taken.
✓ Branch 6 taken 1708767 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 1708767 times.
|
1709199 | ::memcpy(destination, this->m_handle.m_data + offset, static_cast<size_t>(storedSize))); |
| 337 | } | ||
| 338 | 1709038 | actualSize = storedSize; | |
| 339 | 3417991 | 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 | 1708893 | static_cast<void>(this->m_handle.m_count.fetch_sub(1, std::memory_order_acq_rel)); | |
| 345 | 1709152 | slot.m_stateTag.store(packStateTag(LOCKLESS_SLOT_FREE, tagOf(desired) + 1), std::memory_order_release); | |
| 346 | 1708144 | return QueueInterface::Status::OP_OK; | |
| 347 | } | ||
| 348 | // CAS failed — another consumer claimed this slot; yield and rescan. | ||
| 349 |
2/2✓ Branch 0 taken 115 times.
✓ Branch 1 taken 1421933 times.
|
1422048 | if (blocking) { |
| 350 |
2/2✓ Branch 2 taken 115 times.
✓ Branch 6 taken 115 times.
|
115 | static_cast<void>(Os::Task::delay(Fw::TimeInterval(0, 0))); |
| 351 | } | ||
| 352 | } | ||
| 353 | 1078634 | return QueueInterface::Status::EMPTY; | |
| 354 | } | ||
| 355 | |||
| 356 | 228709 | FwSizeType LocklessPriorityQueue::getMessagesAvailable() const { | |
| 357 | 457418 | return static_cast<FwSizeType>(this->m_handle.m_available.load(std::memory_order_acquire)); | |
| 358 | } | ||
| 359 | |||
| 360 | 225952 | FwSizeType LocklessPriorityQueue::getMessageHighWaterMark() const { | |
| 361 | 451904 | 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 |