Address SonarCloud PR 193 issues: code fixes and scoped suppressions

Fix the two BUG-severity destructor-throw findings (S1048/M23_201) by marking
Group::release_token, drain_one_to_pool, and notify_slow_path noexcept.

Apply stylistic refactors: remove continue/break from lock-free retry loops in
TaskQueue/MPSCQueue and Group; make Pool::drain_queues const; rename Binder
ctor param to avoid shadowing; drop redundant BLOCK_SIZE casts.

Add C++14 odr-definitions for template static constexpr BLOCK_SIZE members
used from Pool.cpp/Group.cpp instantiations.

Add sonar-project.properties to scope-ignore S8417 (memory_order) under
src/threading/** and src/extension/**, and S5025/S3630/S3432 (placement-new
queue idioms) under **/scheduler/queue/*.hpp.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: TrentHouliston

sonar-project.properties

@@ -0,0 +1,26 @@
+# SonarCloud issue suppressions for deliberate lock-free / placement-new code.
+# projectKey, organization, sources, tests and coverage settings are passed on
+# the scanner CLI in .github/workflows/sonarcloud.yaml; only the ignore rules
+# below are configured here.
+
+sonar.issue.ignore.multicriteria=e1,e2,e3,e4,e5
+
+# S8417: explicit memory_order arguments are intentional in this concurrency
+# framework; the carefully chosen relaxed/acquire/release/acq_rel orderings are
+# required for performance and must not be forced to seq_cst.
+sonar.issue.ignore.multicriteria.e1.ruleKey=cpp:S8417
+sonar.issue.ignore.multicriteria.e1.resourceKey=src/threading/**
+sonar.issue.ignore.multicriteria.e2.ruleKey=cpp:S8417
+sonar.issue.ignore.multicriteria.e2.resourceKey=src/extension/**
+
+# S5025 (manual new/delete), S3630 (reinterpret_cast) and S3432 (explicit
+# destructor call) are unavoidable in the lock-free queues: manual Block
+# lifetime is dictated by the graveyard reclamation scheme and the
+# reinterpret_cast + explicit ~T() are the placement-new idiom for the aligned
+# slot storage. Scope these to the queue files only.
+sonar.issue.ignore.multicriteria.e3.ruleKey=cpp:S5025
+sonar.issue.ignore.multicriteria.e3.resourceKey=**/scheduler/queue/*.hpp
+sonar.issue.ignore.multicriteria.e4.ruleKey=cpp:S3630
+sonar.issue.ignore.multicriteria.e4.resourceKey=**/scheduler/queue/*.hpp
+sonar.issue.ignore.multicriteria.e5.ruleKey=cpp:S3432
+sonar.issue.ignore.multicriteria.e5.resourceKey=**/scheduler/queue/*.hpp

src/Reactor.hpp

@@ -390,7 +390,7 @@ class Reactor {
 
     public:
         template <typename... Args>
-        Binder(Reactor& r, Args&&... args) : reactor(r), args(std::forward<Args>(args)...) {}
+        Binder(Reactor& r, Args&&... args_) : reactor(r), args(std::forward<Args>(args_)...) {}
 
         template <typename Label, typename Function>
         auto then(Label&& label, Function&& callback) {

src/threading/scheduler/Group.cpp

@@ -107,18 +107,19 @@ namespace threading {
             // Otherwise: no fast waiter was directly entitled. If slow_pending is now 0 and a
             // token is available, give it to any fast waiter we have so they don't get stranded.
             if (was_locked && group.slow_pending.load(std::memory_order_acquire) == 0) {
-                while (true) {
-                    int expected = group.tokens.load(std::memory_order_acquire);
-                    if (expected <= 0) {
-                        break;
-                    }
+                bool claimed = false;
+                int expected = group.tokens.load(std::memory_order_acquire);
+                while (!claimed && expected > 0) {
                     if (group.tokens.compare_exchange_weak(expected,
                                                            expected - 1,
                                                            std::memory_order_acq_rel)) {
                         if (!group.drain_one_to_pool()) {
                             group.tokens.fetch_add(1, std::memory_order_release);
                         }
-                        break;
+                        claimed = true;
+                    }
+                    else {
+                        expected = group.tokens.load(std::memory_order_acquire);
                     }
                 }
             }
@@ -194,15 +195,18 @@ namespace threading {
             // Don't jump ahead of multi-group waiters; if any exist, queue ourselves.
             if (slow_pending.load(std::memory_order_acquire) == 0) {
                 int expected = tokens.load(std::memory_order_acquire);
-                while (expected > 0) {
+                bool done    = false;
+                while (!done && expected > 0) {
                     if (tokens.compare_exchange_weak(expected, expected - 1, std::memory_order_acq_rel)) {
                         if (slow_pending.load(std::memory_order_acquire) > 0) {
                             // Restore the token and fall through to enqueueing.
                             release_token();
-                            break;
+                            done = true;
+                        }
+                        else {
+                            pool->submit({std::move(task), make_running_lock()}, clear_idle);
+                            return true;
                         }
-                        pool->submit({std::move(task), make_running_lock()}, clear_idle);
-                        return true;
                     }
                 }
             }
@@ -234,7 +238,7 @@ namespace threading {
             return false;
         }
 
-        void Group::release_token() {
+        void Group::release_token() noexcept {
             const int prev = tokens.fetch_add(1, std::memory_order_acq_rel);
 
             // If a slow-path waiter exists give them first chance.
@@ -249,7 +253,7 @@ namespace threading {
             }
         }
 
-        void Group::notify_slow_path() {
+        void Group::notify_slow_path() noexcept {
             std::vector<std::shared_ptr<LockHandle>> to_notify;
             /*mutex scope*/ {
                 const std::lock_guard<std::mutex> lock(mutex);
@@ -267,7 +271,7 @@ namespace threading {
             }
         }
 
-        bool Group::drain_one_to_pool() {
+        bool Group::drain_one_to_pool() noexcept {
             WaitEntry entry;
             for (std::size_t bucket = 0; bucket < queue::PRIORITY_BUCKETS; ++bucket) {
                 if (wait_buckets[bucket].try_dequeue(entry)) {

src/threading/scheduler/Group.hpp

@@ -236,9 +236,9 @@ namespace threading {
             const std::shared_ptr<const util::GroupDescriptor> descriptor;
 
         private:
-            void release_token();
-            void notify_slow_path();
-            bool drain_one_to_pool();
+            void release_token() noexcept;
+            void notify_slow_path() noexcept;
+            bool drain_one_to_pool() noexcept;
             std::unique_ptr<Lock> make_running_lock();
 
             /// Available group tokens (signed when waiters are queued on the fast path)

src/threading/scheduler/Pool.cpp

@@ -242,10 +242,10 @@ namespace threading {
             return false;
         }
 
-        void Pool::drain_queues() {
+        void Pool::drain_queues() const {
             Task discarded;
-            for (auto& bucket : buckets) {
-                while (bucket->try_dequeue(discarded)) {
+            for (const auto& bucket : buckets) {
+                while (bucket->try_dequeue(discarded)) { /* discard all queued tasks */
                 }
             }
         }

src/threading/scheduler/Pool.hpp

@@ -231,7 +231,7 @@ namespace threading {
             /**
              * Drain all tasks from the priority buckets.
              */
-            void drain_queues();
+            void drain_queues() const;
 
             /**
              * Get an idle task to execute or hold.

src/threading/scheduler/queue/MPSCQueue.hpp

@@ -195,7 +195,7 @@ namespace threading {
                 bool try_dequeue(T& out) override {
                     while (true) {
                         const std::size_t write_observed = head_block->write.load(std::memory_order_acquire);
-                        const std::size_t published      = std::min(write_observed, static_cast<std::size_t>(BLOCK_SIZE));
+                        const std::size_t published      = std::min(write_observed, BLOCK_SIZE);
 
                         if (head_block->read < published) {
                             Slot& slot = head_block->slots[head_block->read];
@@ -218,21 +218,26 @@ namespace threading {
                             // mid-way through linking the next block; wait briefly for it to appear.
                             if (write_observed > BLOCK_SIZE) {
                                 std::this_thread::yield();
-                                continue;
                             }
-                            return false;
+                            else {
+                                return false;
+                            }
+                        }
+                        else {
+                            // We're the sole consumer so advancing head_block is a plain store. The old
+                            // block goes to the graveyard so any producer that still holds a pointer to
+                            // it (e.g. one mid-way through link_next_block) doesn't touch freed memory.
+                            Block* old = head_block;
+                            head_block = next;
+                            retire_block(old);
                         }
-
-                        // We're the sole consumer so advancing head_block is a plain store. The old
-                        // block goes to the graveyard so any producer that still holds a pointer to
-                        // it (e.g. one mid-way through link_next_block) doesn't touch freed memory.
-                        Block* old = head_block;
-                        head_block = next;
-                        retire_block(old);
                     }
                 }
             };
 
+            template <typename T>
+            constexpr std::size_t MPSCQueue<T>::BLOCK_SIZE;
+
         }  // namespace queue
     }  // namespace scheduler
 }  // namespace threading

src/threading/scheduler/queue/TaskQueue.hpp

@@ -209,58 +209,59 @@ namespace threading {
                         Block* block = head.load(std::memory_order_acquire);
 
                         const std::size_t published =
-                            std::min(block->write.load(std::memory_order_acquire),
-                                     static_cast<std::size_t>(BLOCK_SIZE));
+                            std::min(block->write.load(std::memory_order_acquire), BLOCK_SIZE);
                         std::size_t read_index = block->read.load(std::memory_order_relaxed);
 
                         if (read_index >= published) {
                             if (block->consumed.load(std::memory_order_acquire) < published) {
+                                // Consumers are still finishing slots in this block; let them progress.
                                 std::this_thread::yield();
-                                continue;
                             }
-
-                            Block* next = block->next.load(std::memory_order_acquire);
-                            if (next == nullptr) {
-                                // Producer may still be writing the first slot of an empty-looking block.
-                                if (published == 0 && block->write.load(std::memory_order_acquire) > 0) {
-                                    std::this_thread::yield();
-                                    continue;
+                            else {
+                                Block* next = block->next.load(std::memory_order_acquire);
+                                if (next == nullptr) {
+                                    // Producer may still be writing the first slot of an empty-looking block.
+                                    if (published == 0 && block->write.load(std::memory_order_acquire) > 0) {
+                                        std::this_thread::yield();
+                                    }
+                                    else {
+                                        return false;
+                                    }
+                                }
+                                else {
+                                    head.compare_exchange_strong(block,
+                                                                 next,
+                                                                 std::memory_order_release,
+                                                                 std::memory_order_relaxed);
                                 }
-                                return false;
                             }
-
-                            head.compare_exchange_strong(block, next, std::memory_order_release, std::memory_order_relaxed);
-                            continue;
-                        }
-
-                        if (!block->read.compare_exchange_weak(read_index,
-                                                               read_index + 1,
-                                                               std::memory_order_acq_rel,
-                                                               std::memory_order_relaxed)) {
-                            continue;
                         }
+                        else if (block->read.compare_exchange_weak(read_index,
+                                                                   read_index + 1,
+                                                                   std::memory_order_acq_rel,
+                                                                   std::memory_order_relaxed)) {
+                            Slot& slot = block->slots[read_index];
+                            while (!slot.committed.load(std::memory_order_acquire)) {
+                                std::this_thread::yield();
+                            }
 
-                        Slot& slot = block->slots[read_index];
-                        while (!slot.committed.load(std::memory_order_acquire)) {
-                            std::this_thread::yield();
-                        }
+                            out = std::move(*slot_ptr(slot));
+                            destroy_slot(slot);
 
-                        out = std::move(*slot_ptr(slot));
-                        destroy_slot(slot);
+                            if (block->consumed.fetch_add(1, std::memory_order_acq_rel) + 1 == BLOCK_SIZE) {
+                                try_reclaim_block(block);
+                            }
 
-                        if (block->consumed.fetch_add(1, std::memory_order_acq_rel) + 1 == BLOCK_SIZE) {
-                            try_reclaim_block(block);
+                            return true;
                         }
-
-                        return true;
                     }
                 }
 
                 bool empty() const {
                     Block* block = head.load(std::memory_order_acquire);
                     while (block != nullptr) {
-                        const std::size_t published = std::min(block->write.load(std::memory_order_acquire),
-                                                               static_cast<std::size_t>(BLOCK_SIZE));
+                        const std::size_t published =
+                            std::min(block->write.load(std::memory_order_acquire), BLOCK_SIZE);
                         if (block->read.load(std::memory_order_relaxed) < published) {
                             return false;
                         }
@@ -270,6 +271,9 @@ namespace threading {
                 }
             };
 
+            template <typename T>
+            constexpr std::size_t TaskQueue<T>::BLOCK_SIZE;
+
         }  // namespace queue
     }  // namespace scheduler
 }  // namespace threading