From 4f075902fcac4e1e6c996210193f338d21c66130 Mon Sep 17 00:00:00 2001 From: Rahul Manocha <153310294+manocharahul@users.noreply.github.com> Date: Fri, 24 Oct 2025 08:43:00 -0700 Subject: [PATCH] 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 --- projects/clr/rocclr/platform/commandqueue.cpp | 1 + .../runtime/hsa-runtime/core/inc/runtime.h | 88 ++++++- .../core/runtime/interrupt_signal.cpp | 1 - .../hsa-runtime/core/runtime/runtime.cpp | 151 ++++++++++-- .../hsa-runtime/core/util/mpsc_queue.hpp | 232 ++++++++++++++++++ 5 files changed, 451 insertions(+), 22 deletions(-) create mode 100644 projects/rocr-runtime/runtime/hsa-runtime/core/util/mpsc_queue.hpp diff --git a/projects/clr/rocclr/platform/commandqueue.cpp b/projects/clr/rocclr/platform/commandqueue.cpp index c36fb001ad..34ccc0ebb8 100644 --- a/projects/clr/rocclr/platform/commandqueue.cpp +++ b/projects/clr/rocclr/platform/commandqueue.cpp @@ -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 diff --git a/projects/rocr-runtime/runtime/hsa-runtime/core/inc/runtime.h b/projects/rocr-runtime/runtime/hsa-runtime/core/inc/runtime.h index 4df2d2050a..07b461b820 100644 --- a/projects/rocr-runtime/runtime/hsa-runtime/core/inc/runtime.h +++ b/projects/rocr-runtime/runtime/hsa-runtime/core/inc/runtime.h @@ -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 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 free_list_; + std::vector> 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& 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& events); + private: + //AsyncEventItem Queue + ::rocr::MPSCQueue event_queue_; + AsyncEventsPool asyncEventPool_; + }; + struct PrefetchRange; typedef std::map 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_; diff --git a/projects/rocr-runtime/runtime/hsa-runtime/core/runtime/interrupt_signal.cpp b/projects/rocr-runtime/runtime/hsa-runtime/core/runtime/interrupt_signal.cpp index 8708ceaf96..bf0d7179d7 100644 --- a/projects/rocr-runtime/runtime/hsa-runtime/core/runtime/interrupt_signal.cpp +++ b/projects/rocr-runtime/runtime/hsa-runtime/core/runtime/interrupt_signal.cpp @@ -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_)); } diff --git a/projects/rocr-runtime/runtime/hsa-runtime/core/runtime/runtime.cpp b/projects/rocr-runtime/runtime/hsa-runtime/core/runtime/runtime.cpp index 263c2cbbd1..feb4682028 100644 --- a/projects/rocr-runtime/runtime/hsa-runtime/core/runtime/runtime.cpp +++ b/projects/rocr-runtime/runtime/hsa-runtime/core/runtime/runtime.cpp @@ -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 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 func_arg_t; std::vector functions; - { - ScopedAcquire 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 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 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 lock(&lock_); + if (free_list_.empty()) { + AsyncEventItem* block = reinterpret_cast( + allocate_()(block_size_ * sizeof(AsyncEventItem), __alignof(AsyncEventItem), core::MemoryRegion::AllocateNonPaged, 0)); + if (block == nullptr) { + block_size_ = minblock_; + block = reinterpret_cast( + 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 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& 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& 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; diff --git a/projects/rocr-runtime/runtime/hsa-runtime/core/util/mpsc_queue.hpp b/projects/rocr-runtime/runtime/hsa-runtime/core/util/mpsc_queue.hpp new file mode 100644 index 0000000000..58929cc4aa --- /dev/null +++ b/projects/rocr-runtime/runtime/hsa-runtime/core/util/mpsc_queue.hpp @@ -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 +#include +#include +#include +#include +#include + +/* + * 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). + * - Rebound to internal Node type. + * + */ + +namespace rocr { + +template> +class MPSCQueue { +private: + struct Node { + std::atomic next; + T value; + + template + explicit Node(Args&&... args) + : next(nullptr), value(std::forward(args)...) {} + // Dummy node ctor (no value) + Node() : next(nullptr), value() {} + }; + + using NodeAlloc = typename std::allocator_traits::template rebind_alloc; + using NodeAllocTraits = std::allocator_traits; + + // Cache line padding to reduce false-sharing between head & tail hot fields. + struct alignas(64) PaddedAtomicPtr { + std::atomic ptr; + char pad[64 - sizeof(std::atomic) > 0 ? 64 - sizeof(std::atomic) : 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 + inline void emplace(Args&&... args) { + Node* n = allocate_node(std::forward(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& 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 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 + Node* allocate_node(Args&&... args) { + Node* n = NodeAllocTraits::allocate(alloc_, 1); + try { + NodeAllocTraits::construct(alloc_, n, std::forward(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_ \ No newline at end of file