AI/rocm-systems
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

clr: Implement dynamic stream to HWq logic (#1958)

Browse Source 
* clr: Implement dynamic stream to HW queue assignment

This change implements dynamic stream to hardware queue (HWq) mapping
with the following features:

* Queue depth heuristics with weights for optimal HWq assignment
* Make last used queue sticky for better locality
* Use pipe HWq to pipe mapping - gfx9 follows a round-robin queue to
  pipe mapping based on creation order (single process per device only,
  as pipe ID is statically assigned by runtime)
* More aggressive heuristic usage for better queue distribution
* Extend dynamic queues support for all stream priorities

Environment variables:
* DEBUG_HIP_DYNAMIC_QUEUE: 0 - disabled, 1 - Depth heuristics 2 -
  Depth+Pipe heuristics
* DEBUG_HIP_IGNORE_STREAM_PRIORITY=1: ignore priority stream creation

* clr: Clean up last_used_queue_
This commit is contained in:
SaleelK
2026-01-23 10:40:54 -08:00
committed by GitHub
parent 89170521f5
commit 340f3aa887
10 changed files with 341 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
projects/clr/rocclr/device/rocm/rocdevice.cpp
+191 -98
View File
@@ -136,18 +136,22 @@ Device::Device(hsa_agent_t bkendDevice)
alloc_granularity_(0),
xferQueue_(nullptr),
freeMem_(0),
vgpusAccess_(true) /* Virtual GPU List Ops Lock */
,
vgpusAccess_(true), /* Virtual GPU List Ops Lock */
hsa_exclusive_gpu_access_(false),
queuePool_(QueuePriority::Total),
coopHostcallBuffer_(nullptr),
queueWithCUMaskPool_(QueuePriority::Total),
numOfVgpus_(0),
preferred_numa_node_(0),
maxSdmaReadMask_(0),
maxSdmaWriteMask_(0),
sdma_engine_allocator_(*this),
cpu_agent_info_(nullptr) {
cpu_agent_info_(nullptr),
numHwPipes_(4) {
// Initialize queue pools with proper comparators (requires 'this' pointer)
for (uint i = 0; i < QueuePriority::Total; ++i) {
queuePool_.emplace_back(QueueCompare(this));
queueWithCUMaskPool_.emplace_back(QueueCompare(this));
}
group_segment_.handle = 0;
gpuvm_segment_.handle = 0;
gpu_fine_grained_segment_.handle = 0;
@@ -225,6 +229,11 @@ Device::~Device() {
glb_ctx_ = nullptr;
}
// Destroy transfer queue FIRST (before destroying queues in pool)
// because its destructor will call releaseQueue()
delete xferQueue_;
xferQueue_ = nullptr;
for (auto& it : queuePool_) {
for (auto qIter = it.begin(); qIter != it.end();) {
hsa_queue_t* queue = qIter->first;
@@ -244,9 +253,6 @@ Device::~Device() {
}
queuePool_.clear();
// Destroy transfer queue
delete xferQueue_;
delete blitProgram_;
if (context_ != nullptr) {
@@ -1695,6 +1701,7 @@ device::VirtualDevice* Device::createVirtualDevice(amd::CommandQueue* queue) {
bool profiling = (queue != nullptr) && queue->properties().test(CL_QUEUE_PROFILING_ENABLE);
bool cooperative = false;
bool dedicated_queue = (queue != nullptr) && queue->isDedicatedQueue();
// If amd command queue is null, then it's an internal device queue
if (queue == nullptr) {
@@ -1708,7 +1715,8 @@ device::VirtualDevice* Device::createVirtualDevice(amd::CommandQueue* queue) {
bool q = (queue != nullptr);
VirtualGPU* virtualDevice =
new VirtualGPU(*this, profiling, cooperative, q ? queue->cuMask() : defaultCuMask,
q ? queue->priority() : amd::CommandQueue::Priority::Normal);
q ? queue->priority() : amd::CommandQueue::Priority::Normal,
dedicated_queue);
if (!virtualDevice->create()) {
delete virtualDevice;
@@ -2811,7 +2819,7 @@ VirtualGPU* Device::xferQueue() const {
return nullptr;
}
if (xferQueue_->gpu_queue() == nullptr) {
xferQueue_->set_gpu_queue(thisDevice->AcquireActiveNormalQueue());
xferQueue_->set_gpu_queue(thisDevice->AcquireActiveQueue(amd::CommandQueue::Priority::Normal));
}
}
xferQueue_->enableSyncBlit();
@@ -2863,58 +2871,73 @@ void Device::getHwEventTime(const amd::Event& event, uint64_t* start, uint64_t*
}
// ================================================================================================
hsa_queue_t* Device::getQueueFromPool(const uint qIndex) {
// Check if queue with refCount 0 is available to use
if (queuePool_[qIndex].size() < GPU_MAX_HW_QUEUES) {
for (auto& it : queuePool_[qIndex]) {
if (it.second.refCount == 0) {
it.second.refCount++;
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE, "Selected queue refCount: %p (%d)",
it.first->base_address, it.second.refCount);
return it.first;
}
}
} else {
if (qIndex < QueuePriority::Total && queuePool_[qIndex].size() > 0) {
// Search through all available queues for the lowest counter.
// Note: the map is sorted in the allocation order for possible round-robin selection
typedef decltype(queuePool_)::value_type::const_reference PoolRef;
auto lowest = std::min_element(
queuePool_[qIndex].begin(), queuePool_[qIndex].end(),
[](PoolRef A, PoolRef B) { return A.second.refCount < B.second.refCount; });
lowest->second.refCount++;
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE, "Selected queue refCount: %p (%d)",
lowest->first->base_address, lowest->second.refCount);
return lowest->first;
}
hsa_queue_t* Device::getQueueFromPool(const uint qIndex, bool force_reuse) {
// Only reuse queues when we've reached the maximum limit, unless forced
// Below the limit, return nullptr to allow creating new queues
if (!force_reuse && queuePool_[qIndex].size() < settings().max_hw_queues_) {
return nullptr;
}
// We've hit the limit, must reuse - find the queue with lowest load metric
if (qIndex < QueuePriority::Total && queuePool_[qIndex].size() > 0) {
typedef decltype(queuePool_)::value_type::const_reference PoolRef;
// Select queue based on dynamic_queues_ mode
decltype(queuePool_[qIndex].begin()) lowest;
uint32_t mode = settings().dynamic_queues_;
// gfx9XX pipe distribution: queues map to pipes via queue_id % num_pipes
const bool pipe_dist = settings().queue_pipe_dist_;
const uint32_t num_pipes = numHwPipes_;
lowest = std::min_element(
queuePool_[qIndex].begin(), queuePool_[qIndex].end(),
[mode, pipe_dist, num_pipes](PoolRef A, PoolRef B) {
if (mode >= 1) {
// Mode 1+: Advanced weighted metric with dedicated queue penalty
// Metric = dedicated_queue_penalty + (depth << 4) + refCount
uint64_t metricA = A.second.GetLoadMetric(A.first, mode);
uint64_t metricB = B.second.GetLoadMetric(B.first, mode);
if (metricA == metricB && pipe_dist) {
// gfx9XX pipe distribution: prefer lower pipe IDs for consistent distribution
uint64_t pipeA = A.first->id % num_pipes;
uint64_t pipeB = B.first->id % num_pipes;
return pipeA < pipeB;
}
return metricA < metricB;
} else {
// Mode 0: Simple refCount-based selection
return A.second.refCount < B.second.refCount;
}
});
lowest->second.refCount++;
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE,
"Selected queue (mode=%u): %p refCount: %d, depth: %lu, metric: %lu, pipe: %d%s",
mode, lowest->first->base_address, lowest->second.refCount,
QueueInfo::GetHwQueueDepth(lowest->first),
lowest->second.GetLoadMetric(lowest->first, mode),
pipe_dist ? (lowest->first->id % num_pipes) : -1,
force_reuse ? " (forced)" : "");
return lowest->first;
}
return nullptr;
}
// ================================================================================================
hsa_queue_t* Device::AcquireActiveNormalQueue() {
hsa_queue_t* Device::AcquireActiveQueue(amd::CommandQueue::Priority priority) {
uint32_t queue_size = ROC_AQL_QUEUE_SIZE;
auto queue = acquireQueue(queue_size, false, std::vector<uint32_t>{},
amd::CommandQueue::Priority::Normal, true);
priority, true, false);
return queue;
}
// ================================================================================================
hsa_queue_t* Device::acquireQueue(uint32_t queue_size_hint, bool coop_queue,
const std::vector<uint32_t>& cuMask,
amd::CommandQueue::Priority priority, bool managed) {
amd::ScopedLock l(active_queue_access_);
assert(queuePool_[QueuePriority::Low].size() <= GPU_MAX_HW_QUEUES ||
queuePool_[QueuePriority::Normal].size() <= GPU_MAX_HW_QUEUES ||
queuePool_[QueuePriority::High].size() <= GPU_MAX_HW_QUEUES);
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE,
"Number of allocated hardware queues with low priority: %d,"
" with normal priority: %d, with high priority: %d, maximum per priority is: %d",
queuePool_[QueuePriority::Low].size(), queuePool_[QueuePriority::Normal].size(),
queuePool_[QueuePriority::High].size(), GPU_MAX_HW_QUEUES);
amd::CommandQueue::Priority priority, bool managed,
bool dedicated_queue) {
hsa_amd_queue_priority_t queue_priority;
uint qIndex;
switch (priority) {
@@ -2934,22 +2957,49 @@ hsa_queue_t* Device::acquireQueue(uint32_t queue_size_hint, bool coop_queue,
break;
}
// If we have reached the max number of queues, reuse an existing queue with the matching queue
// priority, choosing the one with the least number of users. Note: Don't attempt to reuse the
// cooperative queue, since it's single per device
if (!coop_queue && (cuMask.size() == 0) &&
((queuePool_[qIndex].size() == GPU_MAX_HW_QUEUES) || queuePool_[qIndex].size() > 0)) {
hsa_queue_t* queue = getQueueFromPool(qIndex);
if (queue != nullptr) {
if (!managed && (qIndex == QueuePriority::Normal)) {
num_normal_queues_++;
}
return queue;
}
// If flag set, force all streams to normal priority
// This means that GPU_MAX_HW_QUEUES may need to be incremented
// to account for the additional normal priority queues
if (DEBUG_HIP_IGNORE_STREAM_PRIORITY) {
queue_priority = HSA_AMD_QUEUE_PRIORITY_NORMAL;
qIndex = QueuePriority::Normal;
}
// Else create a new queue. This also includes the initial state where there
// is no queue.
{ // Lock
amd::ScopedLock l(active_queue_access_);
assert(queuePool_[QueuePriority::Low].size() <= settings().max_hw_queues_ ||
queuePool_[QueuePriority::Normal].size() <= settings().max_hw_queues_ ||
queuePool_[QueuePriority::High].size() <= settings().max_hw_queues_);
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE,
"Number of allocated hardware queues with low priority: %d,"
" with normal priority: %d, with high priority: %d, maximum per priority is: %d",
queuePool_[QueuePriority::Low].size(), queuePool_[QueuePriority::Normal].size(),
queuePool_[QueuePriority::High].size(), settings().max_hw_queues_);
// If we have reached the max number of queues, reuse an existing queue with the matching queue
// priority, choosing the one with the least number of users. Note: Don't attempt to reuse the
// cooperative queue, since it's single per device.
// num_queues_[qIndex] tracks persistent (non-managed) queues per priority.
// When the total queues (managed + non-managed) exceed max_hw_queues_, we must reuse existing
// queues. 'managed' streams do not increment num_queues_, allowing them to use the
// pool without permanently consuming slots. ReleaseActiveQueue() uses this counter to
// decide when to start reclaiming queues.
if (!coop_queue && (cuMask.size() == 0) &&
(queuePool_[qIndex].size() >= settings().max_hw_queues_)) {
hsa_queue_t* queue = getQueueFromPool(qIndex, false);
if (queue != nullptr) {
if (!managed) {
num_queues_[qIndex]++;
}
return queue;
}
}
} // Lock release
// Create a new queue.
uint32_t queue_max_packets = 0;
if (HSA_STATUS_SUCCESS !=
Hsa::agent_get_info(bkendDevice_, HSA_AGENT_INFO_QUEUE_MAX_SIZE, &queue_max_packets)) {
@@ -2971,9 +3021,14 @@ hsa_queue_t* Device::acquireQueue(uint32_t queue_size_hint, bool coop_queue,
&queue) != HSA_STATUS_SUCCESS) {
queue_size >>= 1;
if (queue_size < 64) {
// if a queue with the same requested priority available from the pool, returns it here
if (!coop_queue && (cuMask.size() == 0) && (queuePool_[qIndex].size() > 0)) {
return getQueueFromPool(qIndex);
LogError("Device::acquireQueue: hsa_queue_create failed!");
// If we can't create even a small queue, try to reuse any existing queue
if (!coop_queue && (cuMask.size() == 0)) {
amd::ScopedLock l(active_queue_access_);
if (queuePool_[qIndex].size() > 0) {
bool kForceReuse = true;
return getQueueFromPool(qIndex, kForceReuse);
}
}
ClPrint(amd::LOG_DETAIL_DEBUG, amd::LOG_QUEUE,
"Device::acquireQueue: hsa_queue_create failed!");
@@ -3067,12 +3122,14 @@ hsa_queue_t* Device::acquireQueue(uint32_t queue_size_hint, bool coop_queue,
return nullptr;
}
if (cuMask.size() != 0) {
amd::ScopedLock l(active_queue_access_);
// add queues with custom CU mask into their special pool to keep track
// of mapping of these queues to their associated queueInfo (i.e., hostcall buffers)
auto result = queueWithCUMaskPool_[qIndex].emplace(std::make_pair(queue, QueueInfo()));
assert(result.second && "QueueInfo already exists");
auto& qInfo = result.first->second;
qInfo.refCount = 1;
qInfo.hasDedicatedQueue_ = dedicated_queue; // Track if this is a dedicated queue
return queue;
}
@@ -3083,22 +3140,41 @@ hsa_queue_t* Device::acquireQueue(uint32_t queue_size_hint, bool coop_queue,
// per device.
return queue;
}
// Add queue to the pool (including dedicated queues)
amd::ScopedLock l(active_queue_access_);
auto result = queuePool_[qIndex].emplace(std::make_pair(queue, QueueInfo()));
assert(result.second && "QueueInfo already exists");
auto& qInfo = result.first->second;
qInfo.refCount = 1;
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE, "acquireQueue refCount: %p (%d)",
result.first->first->base_address, result.first->second.refCount);
if (!managed && (cuMask.size() == 0) && (qIndex = QueuePriority::Normal)) {
num_normal_queues_++;
qInfo.hasDedicatedQueue_ = dedicated_queue; // Track if this is a dedicated queue
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE, "acquireQueue refCount: %p (%d) %s",
result.first->first->base_address, result.first->second.refCount,
dedicated_queue ? "(dedicated)" : "");
if (!managed && (cuMask.size() == 0)) {
num_queues_[qIndex]++;
}
return queue;
}
// ================================================================================================
bool Device::ReleaseActiveNormalQueue(hsa_queue_t* queue) {
bool Device::ReleaseActiveQueue(hsa_queue_t* queue, amd::CommandQueue::Priority priority) {
uint qIndex;
switch (priority) {
case amd::CommandQueue::Priority::Low:
qIndex = QueuePriority::Low;
break;
case amd::CommandQueue::Priority::High:
qIndex = QueuePriority::High;
break;
case amd::CommandQueue::Priority::Normal:
case amd::CommandQueue::Priority::Medium:
default:
qIndex = QueuePriority::Normal;
break;
}
// Release a queue if the total number of allocated queues exceeds the max possible
if (num_normal_queues_.load() > GPU_MAX_HW_QUEUES) {
if (num_queues_[qIndex].load() > settings().max_hw_queues_) {
releaseQueue(queue, std::vector<uint32_t>{}, false, true);
return true;
} else {
@@ -3109,36 +3185,52 @@ bool Device::ReleaseActiveNormalQueue(hsa_queue_t* queue) {
// ================================================================================================
void Device::releaseQueue(hsa_queue_t* queue, const std::vector<uint32_t>& cuMask, bool coop_queue,
bool managed) {
amd::ScopedLock l(active_queue_access_);
for (auto& it : cuMask.size() == 0 ? queuePool_ : queueWithCUMaskPool_) {
auto qIter = it.find(queue);
if (qIter != it.end()) {
if (!managed && (cuMask.size() == 0) && (&it == &queuePool_[QueuePriority::Normal])) {
num_normal_queues_--;
}
auto& qInfo = qIter->second;
assert(qInfo.refCount > 0);
qInfo.refCount--;
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE, "releaseQueue refCount:%p (%d)",
qIter->first->base_address, qIter->second.refCount);
// hsa queues with cumask set are not being reused. Hence, if the app uses multiple
// such queues it can cause memory leak and those must be destroyed here once the
// refcount reaches 0.
if ((!cuMask.empty()) && (qInfo.refCount == 0)) {
if (qInfo.hostcallBuffer_) {
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE,
"Deleting hostcall buffer %p for hardware queue %p", qInfo.hostcallBuffer_,
qIter->first->base_address);
amd::disableHostcalls(qInfo.hostcallBuffer_);
context().svmFree(qInfo.hostcallBuffer_);
// Defer cleanup operations outside the lock
void* hostcallBufferToFree = nullptr;
bool shouldDestroyQueue = false;
{ // Lock
amd::ScopedLock l(active_queue_access_);
auto& pools = cuMask.size() == 0 ? queuePool_ : queueWithCUMaskPool_;
for (uint qIndex = 0; qIndex < pools.size(); ++qIndex) {
auto& it = pools[qIndex];
auto qIter = it.find(queue);
if (qIter != it.end()) {
if (!managed && (cuMask.size() == 0)) {
num_queues_[qIndex]--;
}
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE, "Deleting hardware queue %p with refCount 0",
queue->base_address);
qIter = it.erase(qIter);
Hsa::queue_destroy(queue);
auto& qInfo = qIter->second;
assert(qInfo.refCount > 0);
qInfo.refCount--;
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE, "releaseQueue refCount:%p (%d)",
qIter->first->base_address, qIter->second.refCount);
// hsa queues with cumask set are not being reused. Hence, if the app uses multiple
// such queues it can cause memory leak and those must be destroyed here once the
// refcount reaches 0.
if ((!cuMask.empty()) && (qInfo.refCount == 0)) {
hostcallBufferToFree = qInfo.hostcallBuffer_;
shouldDestroyQueue = true;
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE, "Deleting hardware queue %p with refCount 0",
queue->base_address);
it.erase(qIter);
}
break; // Found and processed the queue
}
}
} // Lock release
// Perform expensive cleanup operations outside the lock
if (shouldDestroyQueue) {
if (hostcallBufferToFree) {
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE,
"Deleting hostcall buffer %p for hardware queue %p", hostcallBufferToFree,
queue->base_address);
amd::disableHostcalls(hostcallBufferToFree);
context().svmFree(hostcallBufferToFree);
}
Hsa::queue_destroy(queue);
}
if (coop_queue) { // cooperative queue
ClPrint(amd::LOG_INFO, amd::LOG_QUEUE, "Deleting CG enabled hardware queue %p ",
queue->base_address);
@@ -3150,6 +3242,7 @@ void* Device::getOrCreateHostcallBuffer(hsa_queue_t* queue, bool coop_queue,
const std::vector<uint32_t>& cuMask) {
decltype(queuePool_)::value_type::iterator qIter;
bool found = false;
if (!coop_queue) {
for (auto& it : cuMask.size() == 0 ? queuePool_ : queueWithCUMaskPool_) {
qIter = it.find(queue);