SWDEV-555347 - Remove lock contention in async events loop (#878)

* SWDEV-555347 - Remove lock contention in async events loop * SWDEV-555347 - Introduce Pool of AsyncEventItems * create generic mempool for AsyncEventItem * Use BaseShared allocate and free for async event pool --------- Co-authored-by: Rahul Manocha <rmanocha@amd.com>
2025-10-24 08:43:00 -07:00
parent 7e20e8ec13
commit 4f075902fc
5 changed files with 451 additions and 22 deletions
@@ -168,6 +168,7 @@ void HostQueue::finish(bool cpu_wait) {

    command = getLastQueuedCommand(true);
    if (command == nullptr) {
+      ClPrint(LOG_DEBUG, LOG_CMD, "No command awaiting completion on host");
      return;
    }
    // Force blocking wait if requested. That allows to avoid a build up of unreleased CPU commands
@@ -75,6 +75,7 @@
 #include "core/util/locks.h"
 #include "core/util/os.h"
 #include "core/util/utils.h"
+#include "core/util/mpsc_queue.hpp"

 #include "core/inc/amd_loader_context.hpp"
 #include "core/inc/amd_hsa_code.hpp"
@@ -595,6 +596,89 @@ class Runtime {
    std::vector<void*> arg_;
  };

+  // Event item structure to hold all signal information
+  struct AsyncEventItem {
+    hsa_signal_t signal;
+    hsa_signal_condition_t cond;
+    hsa_signal_value_t value;
+    hsa_amd_signal_handler handler;
+    void* arg;
+    HsaEvent* hsa_event;  //!< A list of HSA events for KFD wait
+    uint64_t age;         //!< The age list for KFD wait
+
+    AsyncEventItem() : signal{0}, cond(HSA_SIGNAL_CONDITION_EQ), value(0),
+                      handler(nullptr), arg(nullptr), hsa_event(nullptr), age(0) {}
+
+    AsyncEventItem(hsa_signal_t sig, hsa_signal_condition_t c, hsa_signal_value_t val,
+                    hsa_amd_signal_handler h, void* a)
+        : signal(sig), cond(c), value(val), handler(h), arg(a),
+          hsa_event(nullptr), age(0) {}
+
+    AsyncEventItem(const AsyncEventItem& other)
+        : signal(other.signal), cond(other.cond), value(other.value),
+          handler(other.handler), arg(other.arg), hsa_event(other.hsa_event), age(other.age) {}
+
+    void init(hsa_signal_t sig, hsa_signal_condition_t c, hsa_signal_value_t v, hsa_amd_signal_handler h, void* a) {
+        signal = sig;
+        cond = c;
+        value = v;
+        handler = h;
+        arg = a;
+    }
+    // Helper operator to convert signal to Signal* for easier access
+    Signal* operator->() {
+      if (signal.handle == 0) {
+        return nullptr;
+      }
+      return core::Signal::Convert(signal);
+    }
+  };
+
+  class AsyncEventsPool : private BaseShared {
+    public:
+      AsyncEventsPool() : block_size_(preallocblocks_ * minblock_) {}
+      ~AsyncEventsPool() { clear(); }
+
+      AsyncEventItem* alloc();
+      void free(AsyncEventItem* item);
+      void clear();
+
+    private:
+      static const size_t minblock_ = 4096 / sizeof(AsyncEventItem);
+      static const size_t preallocblocks_ = 512;
+      static const size_t maxblocksize_ = 1ULL << 28;
+      HybridMutex lock_;
+      std::vector<AsyncEventItem*> free_list_;
+      std::vector<std::pair<void*, size_t>> block_list_;
+      size_t block_size_;
+  };
+  // New concurrent events structure using lock-free queue
+  struct ConcurrentAsyncEvents {
+    ConcurrentAsyncEvents() {}
+
+    void PushBack(hsa_signal_t signal, hsa_signal_condition_t cond,
+                  hsa_signal_value_t value, hsa_amd_signal_handler handler, void* arg);
+
+    void Clear();
+
+    size_t Size();
+
+    bool empty() { return event_queue_.empty(); }
+
+    //! Get all events for processing
+    bool GetAllEvents(std::vector<AsyncEventItem>& all_events);
+
+    //! Get single event for processing
+    bool GetEvent(AsyncEventItem& event);
+
+    //! Add events back to queue (for events that need to be kept)
+    void AddEventsBack(const std::vector<AsyncEventItem>& events);
+  private:
+    //AsyncEventItem Queue
+    ::rocr::MPSCQueue<AsyncEventItem*> event_queue_;
+    AsyncEventsPool asyncEventPool_;
+  };
+
  struct PrefetchRange;
  typedef std::map<uintptr_t, PrefetchRange> prefetch_map_t;

@@ -747,8 +831,10 @@ class Runtime {
  struct AsyncEventsInfo {
    AsyncEventsControl control;
    AsyncEvents events;
-    AsyncEvents new_events;
+    ConcurrentAsyncEvents new_events;
    bool monitor_exceptions;
+
+    AsyncEventsInfo() : control(), events(), new_events(), monitor_exceptions(false) {}
  };

  struct AsyncEventsInfo asyncSignals_;
@@ -372,7 +372,6 @@ hsa_signal_value_t InterruptSignal::CasAcqRel(hsa_signal_value_t expected,
 }
  /// @brief Notify driver of signal value change if necessary.
  void InterruptSignal::SetEvent() {
-    std::atomic_signal_fence(std::memory_order_seq_cst);
    if (InWaiting()) HSAKMT_CALL(hsaKmtSetEvent(event_));
  }

@@ -832,7 +832,6 @@ hsa_status_t Runtime::SetAsyncSignalHandler(hsa_signal_t signal,
                                            hsa_signal_value_t value,
                                            hsa_amd_signal_handler handler,
                                            void* arg) {
-
  struct AsyncEventsInfo* asyncInfo = &asyncSignals_;
  int priority = runtime_singleton_->flag().async_events_thread_priority();

@@ -847,7 +846,6 @@ hsa_status_t Runtime::SetAsyncSignalHandler(hsa_signal_t signal,
    }
  }

-  ScopedAcquire<HybridMutex> scope_lock(&asyncInfo->control.lock);

  // Lazy initializer
  if (asyncInfo->control.async_events_thread_ == NULL) {
@@ -1886,26 +1884,19 @@ void Runtime::AsyncEventsLoop(void* _eventsInfo) {
    // Insert new signals and find plain functions
    typedef std::pair<void (*)(void*), void*> func_arg_t;
    std::vector<func_arg_t> functions;
-    {
-      ScopedAcquire<HybridMutex> scope_lock(&async_events_control_.lock);
-      for (size_t i = 0; i < new_async_events_.Size(); i++) {
-        if (new_async_events_.signal_[i].handle == 0) {
-          functions.push_back(
-              func_arg_t((void (*)(void*))new_async_events_.handler_[i],
-                         new_async_events_.arg_[i]));
-          continue;
-        }
-        async_events_.PushBack(
-            new_async_events_.signal_[i], new_async_events_.cond_[i],
-            new_async_events_.value_[i], new_async_events_.handler_[i],
-            new_async_events_.arg_[i]);
+    std::vector<AsyncEventItem> new_events;
+    new_async_events_.GetAllEvents(new_events);
+    for (const auto& event : new_events) {
+      if (event.signal.handle == 0) {
+        functions.push_back(func_arg_t((void (*)(void*))event.handler, event.arg));
+        continue;
      }
-      new_async_events_.Clear();
+      async_events_.PushBack(event.signal, event.cond, event.value, event.handler,event.arg);
    }
-
    // Call plain functions
-    for (size_t i = 0; i < functions.size(); i++)
+    for (size_t i = 0; i < functions.size(); i++) {
      functions[i].first(functions[i].second);
+    }
    functions.clear();
  }

@@ -1914,11 +1905,131 @@ void Runtime::AsyncEventsLoop(void* _eventsInfo) {
    hsa_signal_handle(async_events_.signal_[i])->Release();
  async_events_.Clear();

-  for (size_t i = 0; i < new_async_events_.Size(); i++)
-    hsa_signal_handle(new_async_events_.signal_[i])->Release();
+  std::vector<AsyncEventItem> remaining_events;
+  new_async_events_.GetAllEvents(remaining_events);
+  for (const auto& event : remaining_events) {
+    if (event.signal.handle != 0) {
+      hsa_signal_handle(event.signal)->Release();
+    }
+  }
  new_async_events_.Clear();
 }

+void Runtime::AsyncEventsPool::clear() {
+  ifdebug {
+    size_t capacity = 0;
+    for (auto& block : block_list_) capacity += block.second;
+    if (capacity != free_list_.size())
+      debug_print("Warning: Resource leak detected by AsyncEventsPool, %ld items leaked.\n",
+                  capacity - free_list_.size());
+  }
+
+  for (auto& block : block_list_) free_()(block.first);
+  block_list_.clear();
+  free_list_.clear();
+}
+
+Runtime::AsyncEventItem* Runtime::AsyncEventsPool::alloc() {
+  ScopedAcquire<HybridMutex> lock(&lock_);
+  if (free_list_.empty()) {
+    AsyncEventItem* block = reinterpret_cast<AsyncEventItem*>(
+        allocate_()(block_size_ * sizeof(AsyncEventItem), __alignof(AsyncEventItem), core::MemoryRegion::AllocateNonPaged, 0));
+    if (block == nullptr) {
+      block_size_ = minblock_;
+      block = reinterpret_cast<AsyncEventItem*>(
+          allocate_()(block_size_ * sizeof(AsyncEventItem), __alignof(AsyncEventItem), core::MemoryRegion::AllocateNonPaged, 0));
+      if (block == nullptr) throw std::bad_alloc();
+    }
+
+    MAKE_NAMED_SCOPE_GUARD(throwGuard, [&]() { free_()(block); });
+    block_list_.push_back(std::make_pair(block, block_size_));
+    throwGuard.Dismiss();
+
+    for (int i = 0; i < block_size_; i++) {
+      free_list_.push_back(&block[i]);
+    }
+    if (block_size_ > maxblocksize_)
+      block_size_ *= 2;
+  }
+  AsyncEventItem* ret = free_list_.back();
+  new (ret) AsyncEventItem();
+  free_list_.pop_back();
+  return ret;
+}
+
+void Runtime::AsyncEventsPool::free(AsyncEventItem* ptr) {
+  if (ptr == nullptr) return;
+
+  ptr->~AsyncEventItem();
+  ScopedAcquire<HybridMutex> lock(&lock_);
+
+  ifdebug {
+    bool valid = false;
+    for (auto& block : block_list_) {
+      if ((block.first <= ptr) &&
+          (uintptr_t(ptr) < uintptr_t(block.first) + block.second * sizeof(AsyncEventItem))) {
+        valid = true;
+        break;
+      }
+    }
+    assert(valid && "Object does not belong to pool.");
+  }
+  free_list_.push_back(ptr);
+}
+void Runtime::ConcurrentAsyncEvents::PushBack(hsa_signal_t signal,
+                                             hsa_signal_condition_t cond,
+                                             hsa_signal_value_t value,
+                                             hsa_amd_signal_handler handler, void* arg) {
+  // Allocate memory for the new event item
+  AsyncEventItem* item = asyncEventPool_.alloc();
+  item->init(signal, cond, value, handler, arg);
+  event_queue_.enqueue(item);
+}
+
+void Runtime::ConcurrentAsyncEvents::Clear() {
+  // Dequeue all items to clear the queue
+  while (!event_queue_.empty()) {
+    AsyncEventItem* item = event_queue_.dequeue();
+    asyncEventPool_.free(item);
+  }
+  asyncEventPool_.clear();
+}
+
+bool Runtime::ConcurrentAsyncEvents::GetEvent(AsyncEventItem& event) {
+  AsyncEventItem* item = event_queue_.dequeue();
+  if (item != nullptr) {
+    event = *item;
+    asyncEventPool_.free(item);
+    return true;
+  }
+  return false;
+}
+
+bool Runtime::ConcurrentAsyncEvents::GetAllEvents(std::vector<AsyncEventItem>& all_events) {
+  AsyncEventItem* item = nullptr;
+  while (!event_queue_.empty()) {
+    item = event_queue_.dequeue();
+    if (item == nullptr) {
+      return false;
+    }
+    all_events.emplace_back(*item);
+    asyncEventPool_.free(item);
+  }
+  return true;
+}
+
+void Runtime::ConcurrentAsyncEvents::AddEventsBack(const std::vector<AsyncEventItem>& events) {
+  for (const auto& event : events) {
+    AsyncEventItem* item = asyncEventPool_.alloc();
+    *item = event;
+    event_queue_.enqueue(item);
+  }
+}
+
+size_t Runtime::ConcurrentAsyncEvents::Size() {
+  return event_queue_.size();
+}
+
 void Runtime::BindErrorHandlers() {
  if (!core::g_use_interrupt_wait || gpu_agents_.empty()) return;

@@ -0,0 +1,232 @@
+/*
+************************************************************************************************************************
+*
+*  Copyright (C) 2007-2022 Advanced Micro Devices, Inc.  All rights reserved.
+*  SPDX-License-Identifier: MIT
+*
+***********************************************************************************************************************/
+
+#ifndef HSA_RUNTIME_CORE_UTIL_MPSC_QUEUE_HPP_
+#define HSA_RUNTIME_CORE_UTIL_MPSC_QUEUE_HPP_
+
+#include <atomic>
+#include <cstddef>
+#include <memory>
+#include <type_traits>
+#include <vector>
+#include <cassert>
+
+/*
+ * Lock-free Multi-Producer Single-Consumer (MPSC) queue.
+ *
+ * Algorithm:
+ *  - Based on a classic singly-linked list with a dummy stub node.
+ *  - Producers:
+ *      1. Allocate & construct a node (value constructed in-place).
+ *      2. Set node->next = nullptr.
+ *      3. prev = tail_.exchange(node, acq_rel).
+ *      4. prev->next.store(node, release).  (Publishes the node to the consumer.)
+ *  - Consumer (single thread):
+ *      1. Reads head_->next (acquire).
+ *      2. If null -> empty.
+ *      3. Else claim next, move its value out, advance head_, delete old dummy.
+ *
+ * Progress:
+ *  - Enqueue is wait-free for producers (bounded steps, no loops).
+ *  - Dequeue is lock-free (O(1); waits only if empty).
+ *
+ * Thread-safety:
+ *  - Multiple threads may call enqueue() concurrently.
+ *  - Exactly ONE thread may call any of: dequeue(), dequeue_batch(), clear(), destructor.
+ *
+ * Memory ordering rationale:
+ *  - tail_.exchange(..., acq_rel) pairs with consumer's acquire load of next pointer.
+ *  - The prev->next.store(..., release) publishes the new node after its value is fully constructed.
+ *  - Consumer's next.load(acquire) ensures it sees a fully initialized node.
+ *
+ * Custom Allocator:
+ *  - Alloc template parameter (defaults std::allocator<T>).
+ *  - Rebound to internal Node type.
+ *
+ */
+
+namespace rocr {
+
+template<typename T, typename Alloc = std::allocator<T>>
+class MPSCQueue {
+private:
+    struct Node {
+        std::atomic<Node*> next;
+        T value;
+
+        template<typename... Args>
+        explicit Node(Args&&... args)
+          : next(nullptr), value(std::forward<Args>(args)...) {}
+        // Dummy node ctor (no value)
+        Node() : next(nullptr), value() {}
+    };
+
+    using NodeAlloc = typename std::allocator_traits<Alloc>::template rebind_alloc<Node>;
+    using NodeAllocTraits = std::allocator_traits<NodeAlloc>;
+
+    // Cache line padding to reduce false-sharing between head & tail hot fields.
+    struct alignas(64) PaddedAtomicPtr {
+        std::atomic<Node*> ptr;
+        char pad[64 - sizeof(std::atomic<Node*>) > 0 ? 64 - sizeof(std::atomic<Node*>) : 1];
+    };
+
+public:
+    MPSCQueue()
+      : alloc_()
+    {
+        Node* stub = create_node_stub();
+        head_.ptr.store(stub, std::memory_order_relaxed);
+        tail_.ptr.store(stub, std::memory_order_relaxed);
+        qsize_.store(0, std::memory_order_relaxed);
+    }
+
+    explicit MPSCQueue(const Alloc& alloc)
+      : alloc_(alloc)
+    {
+        Node* stub = create_node_stub();
+        head_.ptr.store(stub, std::memory_order_relaxed);
+        tail_.ptr.store(stub, std::memory_order_relaxed);
+        qsize_.store(0, std::memory_order_relaxed);
+    }
+
+    // Non-copyable / non-movable (simplify invariants)
+    MPSCQueue(const MPSCQueue&) = delete;
+    MPSCQueue& operator=(const MPSCQueue&) = delete;
+    MPSCQueue(MPSCQueue&&) = delete;
+    MPSCQueue& operator=(MPSCQueue&&) = delete;
+
+    ~MPSCQueue() {
+        clear(); // drains and deletes any remaining nodes
+        Node* stub = head_.ptr.load(std::memory_order_relaxed);
+        if (stub) {
+            destroy_node(stub);
+        }
+        qsize_.store(0, std::memory_order_relaxed);
+    }
+
+    // Enqueue by const reference
+    inline void enqueue(const T& value) {
+        emplace(value);
+    }
+
+    // Enqueue by rvalue
+    inline void enqueue(T&& value) {
+        emplace(std::move(value));
+    }
+
+    // Perfect-forwarding construction
+    template<typename... Args>
+    inline void emplace(Args&&... args) {
+        Node* n = allocate_node(std::forward<Args>(args)...);
+        // Publish:
+        Node* prev = tail_.ptr.exchange(n, std::memory_order_acq_rel);
+        // Link from previous tail
+        prev->next.store(n, std::memory_order_release);
+        qsize_.fetch_add(1, std::memory_order_release);
+    }
+
+    // Try dequeue single element; returns false if empty
+    inline T dequeue() {
+        Node* head = head_.ptr.load(std::memory_order_relaxed);
+        Node* next = head->next.load(std::memory_order_acquire);
+        if (!next) {
+            return nullptr; // empty
+        }
+        // Move value out
+        T out = std::move(next->value);
+        // Advance head
+        head_.ptr.store(next, std::memory_order_relaxed);
+        // Old head was dummy
+        destroy_node(head);
+        qsize_.fetch_sub(1, std::memory_order_acq_rel);
+        return out;
+    }
+
+    // Non-blocking peek (observes front without removing). Not linearizable with concurrent enqueues,
+    // but safe for single-consumer inspection.
+    inline bool peek(T& out) const {
+        Node* head = head_.ptr.load(std::memory_order_relaxed);
+        Node* next = head->next.load(std::memory_order_acquire);
+        if (!next) return false;
+        out = next->value; 
+        return true;
+    }
+
+    // Batch dequeue up to max_items. Returns actual count.
+    inline size_t dequeue_batch(std::vector<T>& out) {
+        size_t count = 0;
+        while (1) {
+            Node* head = head_.ptr.load(std::memory_order_relaxed);
+            Node* next = head->next.load(std::memory_order_acquire);
+            if (!next) break;
+            out.emplace_back(std::move(next->value));
+            head_.ptr.store(next, std::memory_order_relaxed);
+            destroy_node(head);
+            ++count;
+        }
+        if (count) {
+            qsize_.fetch_sub(count, std::memory_order_acq_rel);
+        }
+        return count;
+    }
+
+    // empty check (safe for single consumer scenario)
+    inline bool empty() const {
+        return qsize_.load(std::memory_order_acquire) == 0;
+    }
+
+    // Drain everything
+    inline void clear() {
+        T dummy;
+        do  {
+            dummy = dequeue();
+        } while (dummy != nullptr);
+    }
+
+    inline size_t size() const {
+        return qsize_.load(std::memory_order_acquire);
+    }
+
+private:
+    NodeAlloc alloc_;
+    PaddedAtomicPtr head_;
+    PaddedAtomicPtr tail_;
+    alignas(64) std::atomic<size_t> qsize_;
+
+    Node* create_node_stub() {
+        Node* n = NodeAllocTraits::allocate(alloc_, 1);
+        try {
+            NodeAllocTraits::construct(alloc_, n);
+        } catch (...) {
+            NodeAllocTraits::deallocate(alloc_, n, 1);
+            throw;
+        }
+        return n;
+    }
+
+    template<typename... Args>
+    Node* allocate_node(Args&&... args) {
+        Node* n = NodeAllocTraits::allocate(alloc_, 1);
+        try {
+            NodeAllocTraits::construct(alloc_, n, std::forward<Args>(args)...);
+        } catch (...) {
+            NodeAllocTraits::deallocate(alloc_, n, 1);
+            throw;
+        }
+        n->next.store(nullptr, std::memory_order_relaxed);
+        return n;
+    }
+
+    void destroy_node(Node* n) {
+        NodeAllocTraits::destroy(alloc_, n);
+        NodeAllocTraits::deallocate(alloc_, n, 1);
+    }
+};
+
+}
+#endif // HSA_RUNTIME_CORE_UTIL_MPSC_QUEUE_HPP_