[OpenMP][AMDGPU] Single eager resource init + HSA queue utilization tracking

[mhalk]

Summary

This patch lazily initializes queues/streams/events since their initialization
might come at a cost even if we do not use them.

To further benefit from this, AMDGPU/HSA queue management is moved into the
AMDGPUStreamManager of an AMDGPUDevice. Streams may now use different HSA queues
during their lifetime and identify busy queues.

When a Stream is requested from the resource manager, it will search for and
try to assign an idle queue. During the search for an idle queue the manager
may initialize more queues, up to the set maximum (default: 4).
When no idle queue could be found: resort to round robin selection.

With contributions from Johannes Doerfert <[e-mail]>

Depends on D156245

[jdoerfert]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

791

rename Busy into sth more meaningful, e.g., NumUsers.
Spell out the inc/dec, maybe "addUser", "removeUser"

1438

Make the class final.

1446

Just resize.

1482

This ignores the queue size, doesn't it? We should not grow larger than that value.

1486

Start with Current = NextQueue.fetch_add(1, std::memory_order_relaxed) % Size when you iterate, that way you won't check the first queue all the time.

[jplehr]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1449

In case of error, I think we should report and return the error instead of reporting and keep going.

This is the initial queue, so if this fails, there is no queue available.

[mhalk]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1446

Initially, I wanted to do that, but AMDGPUQueueTy has a member (Mutex) with deleted copy ctor.

1449

Agreed; as discussed, we'll return Err and not report it at this level.

It will be reported by the higher layer, for example like this:
"PluginInterface" error: Failure to initialize device 0: Error in hsa_queue_create: HSA_STATUS_ERROR_INVALID_ARGUMENT: One of the actual arguments does not meet a precondition stated in the documentation of the corresponding formal argument.

1482

To the best of my knowledge we should not exceed the set (max) number of queues.
This will only resize the streams, as the size of Queues is only set once, during init(...).

1486

Agreed; as discussed, we're going with two loops to search for an idle queue.

[mhalk]

Thank you very much for the comments.
This should implement the current feedback.

[kevinsala]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

601

Do we need these three atomic operations with memory_order_seq_cst? Or a more relaxed memory order could be enough?

1477

I'm afraid most of the code in getResource and returnResource is duplicated from the generic resource manager. Would it be possible to use the original GenericDeviceResourceManagerTy::getResource() instead? For instance:

ResourceRef getResource() override {
  ResourceRef resource = GenericDeviceResourceManagerTy::getResource();
  // Perform any change on resource
  ...

  return resource;
}

(Note that with this modification, the changes on the resource are no longer protected by the stream manager mutex)

1510

I don't think NextQueue needs atomic operations if this function is called while holding the stream manager mutex.

1547

I feel this patch implements a Queue manager inside a Stream manager. Wouldn't it be better to define this logic inside a new AMDGPUQueueManagerTy and just have a reference of it in the AMDGPUStreamManagerTy?

[kevinsala]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1514

Also, NextQueue shouldn't be incremented while we advance in the search of non busy queues?

1527

I believe we can simplify and merge these two loops into a single one

[ye-luo]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

601

Is this function already under the stream manager mutex when being called? I feel NumUsers doesn't need to be atomic.

[kevinsala]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

601

Yes, as it is now (everything protected by the stream manager mutex), it seems it shouldn't be atomic.

[jdoerfert]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1450–1452

1484

As @kevinsala said, the above, and the auto &Resource = part below, can be replaced with ResourceRef Resource = GenericDeviceResourceManagerTy::getResource();

1502

The above, except removeUser, should just be GenericDeviceResourceManagerTy::returnResource(Resource);, no?

1516

early exit if (busy) continue

1527

Yeah, my bad, I suggested this, we can do sth like

for (I = 0; < MaxNumQueues; ++I) {
  Idx = StartIndex++;
  if (StartIndex == MaxNumQueues) StartIndex = 0;
  // use Idx not I
...
}

1547

We do not really manage the queues the same way. we can do more reuse, see above.

[mhalk]

Thanks for all the valuable feedback -- this should implement it.

Still trying to figure out which parts really need to be protected by Locks.
So, please take a close look at these parts of the code.

Since they are currently guarded, I removed the:
atomics as suggested.
Lock from initLazy

[mhalk]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1478–1481

Rather unhappy with this particular piece of code.
Now there's a chance for interleaving.
I don't know if that's good, "ok" or bad, but if I had to guess: the latter.

But without unlocking, this will cause a deadlock.
(Could also revert to a lock_guard and call it's dtor, if preferred.)

Thoughts and comments on this would be highly appreciated.

[mhalk]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1009–1015

Oversight: Will be removed.

[jdoerfert]

Looks pretty good to me, others?

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1498–1508

[kevinsala]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1478–1481

Yes, this seems correct, but I agree that it may produce unnecessary overhead.

To cover this use case, we could extend the generic getResource and returnResource to accept an (optional) template functor that processes the element before returning (i.e., with the lock acquired). Something like:

class GenericDeviceResourceManagerTy {
protected:
    template <typename Func>
    ResourceRef getResourceAndProcess(Func processor)
    {
        ResourceRef ref = ...;
        processor(ref);
        return ref;
    }

public:
    virtual ResourceRef getResource()
    {
        return getResource([](ResourceRef &) { /* do nothing */ });
    }
}

And your getResource function would look like something like:

ResourceRef getResource() override {
    return GenericDeviceResourceManagerTy::getResource(
        [this](ResourceRef &ref) {
            assignNextQueue(ref);
        });
}

I will implement this change in another patch. For the moment, this look fine to me. Thanks

[kevinsala]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1478–1481

Fix to my previous comment:

template <typename Func>
ResourceRef getResource(Func processor)
{
    std::lock_guard<std::mutex> Lock(Mutex);
    ResourceRef ref = ...;
    processor(ref);
    return ref;
}

[mhalk]

Thanks for checking and letting me know!

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1478–1481

Ok, great -- will leave this as it is, for now.

[mhalk]

Adapted to D156245 @kevinsala
Thanks for the heads-up!

Please take a look at the signature of the two lambdas, I decided to got with
AMDGPUStreamTy *, instead of duplicating the private ResourceHandleTy.

I'll try to do some further testing in the next days.

[kevinsala]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1469

I would rename the parameter to Resource, Handle, or similar. The parameter is no longer a ResourceReference. The same for the lambda's parmeter and the returnResource function.

1471

You can make this function to return the error and propagate it outside. Then, the assignNextQueue function can return the error when failing to initialize a queue, instead of handling it.

1479

1487

You can work with a specific parameter AMDGPUStreamTy *Stream directly.

1501

if (auto Err = Q->initLazy(Agent, QueueSize))
  return Err;

[mhalk]

Merged this diff with D152035 as discussed with @jdoerfert.
Updated commit message accordingly.

Implemented feedback from @kevinsala + rebased.
Thank you both!

[kevinsala]

Looks good to me now. Thanks!

[mhalk]

Fixed oversight and made AMDGPUQueueTy::init lazy by default.

[jdoerfert]

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

1445

Style: I would suggest HSA rather than Hsa.

[mhalk]

Sure!
Changed Hsa to HSA within names.

Thanks for all the feedback and patience :)

openmp/libomptarget/plugins-nextgen/amdgpu/src/rtl.cpp

 
 /// Classes for holding AMDGPU signals and managing signals.
 using AMDGPUSignalRef = AMDGPUResourceRef<AMDGPUSignalTy>;
 using AMDGPUSignalManagerTy = GenericDeviceResourceManagerTy<AMDGPUSignalRef>;
 
 /// Class holding an HSA queue to submit kernel and barrier packets.
 struct AMDGPUQueueTy {
   /// Create an empty queue.
-  AMDGPUQueueTy() : Queue(nullptr), Mutex() {}
+  AMDGPUQueueTy() : Queue(nullptr), Mutex(), NumUsers(0) {}
 
   /// Initialize a new queue belonging to a specific agent.
   Error init(hsa_agent_t Agent, int32_t QueueSize) {
     hsa_status_t Status =
         hsa_queue_create(Agent, QueueSize, HSA_QUEUE_TYPE_MULTI, callbackError,
                          nullptr, UINT32_MAX, UINT32_MAX, &Queue);
     return Plugin::check(Status, "Error in hsa_queue_create: %s");
   }
 
   /// If the queue is not initialized, do it now.
   Error initLazy(hsa_agent_t Agent, int32_t QueueSize) {
-    // Lock the queue during the lazy init
-    std::lock_guard<std::mutex> Lock(Mutex);
     if (Queue)
       return Plugin::success();
     return init(Agent, QueueSize);
   }
 
   /// Deinitialize the queue and destroy its resources.
   Error deinit() {
     std::lock_guard<std::mutex> Lock(Mutex);
     if (!Queue)
       return Plugin::success();
     hsa_status_t Status = hsa_queue_destroy(Queue);
     return Plugin::check(Status, "Error in hsa_queue_destroy: %s");
   }
 
+  /// Returns if this queue is considered busy
+  bool isBusy() const { return NumUsers > 0; }
+
+  /// Decrement user count of the queue object
+  void removeUser() { --NumUsers; }
+
+  /// Increase user count of the queue object
+  void addUser() { ++NumUsers; }
+
   /// Push a kernel launch to the queue. The kernel launch requires an output
   /// signal and can define an optional input signal (nullptr if none).
   Error pushKernelLaunch(const AMDGPUKernelTy &Kernel, void *KernelArgs,
                          uint32_t NumThreads, uint64_t NumBlocks,
                          uint32_t GroupSize, AMDGPUSignalTy *OutputSignal,
                          AMDGPUSignalTy *InputSignal) {
     assert(OutputSignal && "Invalid kernel output signal");
 
private:
   /// published in a multi-thread scenario. Without a queue lock, a thread T1
   /// could acquire packet P and thread T2 acquire packet P+1. Thread T2 could
   /// publish its packet P+1 (signaling the queue's doorbell) before packet P
   /// from T1 is ready to be processed. That scenario should be invalid. Thus,
   /// we use the following mutex to make packet acquiring and publishing atomic.
   /// TODO: There are other more advanced approaches to avoid this mutex using
   /// atomic operations. We can further investigate it if this is a bottleneck.
   std::mutex Mutex;
+
+  /// Indicates that the queue is busy when > 0
+  int NumUsers;
 };
 
 /// Struct that implements a stream of asynchronous operations for AMDGPU
 /// devices. This class relies on signals to implement streams and define the
 /// dependencies between asynchronous operations.
 struct AMDGPUStreamTy {
 private:
   /// Utility struct holding arguments for async H2H memory copies.
Error performAction() {
       return Plugin::success();
     }
   };
 
   /// The device agent where the stream was created.
   hsa_agent_t Agent;
 
   /// The queue that the stream uses to launch kernels.
-  AMDGPUQueueTy &Queue;
+  AMDGPUQueueTy *Queue;
 
   /// The manager of signals to reuse signals.
   AMDGPUSignalManagerTy &SignalManager;
 
   /// Array of stream slots. Use std::deque because it can dynamically grow
   /// without invalidating the already inserted elements. For instance, the
   /// std::vector may invalidate the elements by reallocating the internal
   /// array if there is not enough space on new insertions.
Error complete() {
     return Plugin::success();
   }
 
   /// Make the current stream wait on a specific operation of another stream.
   /// The idea is to make the current stream waiting on two signals: 1) the last
   /// signal of the current stream, and 2) the last signal of the other stream.
   /// Use a barrier packet with two input signals.
   Error waitOnStreamOperation(AMDGPUStreamTy &OtherStream, uint32_t Slot) {
+    if (Queue == nullptr)
+      return Plugin::error("Target queue was nullptr");
+
     /// The signal that we must wait from the other stream.
     AMDGPUSignalTy *OtherSignal = OtherStream.Slots[Slot].Signal;
 
     // Prevent the release of the other stream's signal.
     OtherSignal->increaseUseCount();
 
     // Retrieve an available signal for the operation's output.
     AMDGPUSignalTy *OutputSignal = SignalManager.getResource();
     OutputSignal->reset();
     OutputSignal->increaseUseCount();
 
     // Consume stream slot and compute dependencies.
     auto [Curr, InputSignal] = consume(OutputSignal);
 
     // Setup the post action to release the signal.
     if (auto Err = Slots[Curr].schedReleaseSignal(OtherSignal, &SignalManager))
       return Err;
 
     // Push a barrier into the queue with both input signals.
-    return Queue.pushBarrier(OutputSignal, InputSignal, OtherSignal);
+    return Queue->pushBarrier(OutputSignal, InputSignal, OtherSignal);
+  }
+
+  /// Assign another queue to this stream. This should only be called by a
+  /// ResourceManager when this stream is provided from a resource pool. By
+  /// using another queue, we may avoid putting work on an already busy queue.
+  void assignQueue(AMDGPUQueueTy *NewQueue) {
+    NewQueue->addUser();
+    Queue = NewQueue;
   }
 
   /// Callback for running a specific asynchronous operation. This callback is
   /// used for hsa_amd_signal_async_handler. The argument is the operation that
   /// should be executed. Notice we use the post action mechanism to codify the
   /// asynchronous operation.
   static bool asyncActionCallback(hsa_signal_value_t Value, void *Args) {
     StreamSlotTy *Slot = reinterpret_cast<StreamSlotTy *>(Args);
     assert(Slot && "Invalid slot");
public:
   /// Push a asynchronous kernel to the stream. The kernel arguments must be
   /// placed in a special allocation for kernel args and must keep alive until
   /// the kernel finalizes. Once the kernel is finished, the stream will release
   /// the kernel args buffer to the specified memory manager.
   Error pushKernelLaunch(const AMDGPUKernelTy &Kernel, void *KernelArgs,
                          uint32_t NumThreads, uint64_t NumBlocks,
                          uint32_t GroupSize,
                          AMDGPUMemoryManagerTy &MemoryManager) {
+    if (Queue == nullptr)
+      return Plugin::error("Target queue was nullptr");
+
     // Retrieve an available signal for the operation's output.
     AMDGPUSignalTy *OutputSignal = SignalManager.getResource();
     OutputSignal->reset();
     OutputSignal->increaseUseCount();
 
     std::lock_guard<std::mutex> StreamLock(Mutex);
 
     // Consume stream slot and compute dependencies.
     auto [Curr, InputSignal] = consume(OutputSignal);
 
     // Setup the post action to release the kernel args buffer.
     if (auto Err = Slots[Curr].schedReleaseBuffer(KernelArgs, MemoryManager))
       return Err;
 
     // Push the kernel with the output signal and an input signal (optional)
-    return Queue.pushKernelLaunch(Kernel, KernelArgs, NumThreads, NumBlocks,
+    return Queue->pushKernelLaunch(Kernel, KernelArgs, NumThreads, NumBlocks,
                                    GroupSize, OutputSignal, InputSignal);
   }
 
   /// Push an asynchronous memory copy between pinned memory buffers.
   Error pushPinnedMemoryCopyAsync(void *Dst, const void *Src,
                                   uint64_t CopySize) {
     // Retrieve an available signal for the operation's output.
     AMDGPUSignalTy *OutputSignal = SignalManager.getResource();
     OutputSignal->reset();
Expected<bool> query() {
     return true;
   }
 
   /// Record the state of the stream on an event.
   Error recordEvent(AMDGPUEventTy &Event) const;
 
   /// Make the stream wait on an event.
   Error waitEvent(const AMDGPUEventTy &Event);
+
+  friend struct AMDGPUStreamManagerTy;
 };
 
 /// Class representing an event on AMDGPU. The event basically stores some
 /// information regarding the state of the recorded stream.
 struct AMDGPUEventTy {
   /// Create an empty event.
   AMDGPUEventTy(AMDGPUDeviceTy &Device)
       : RecordedStream(nullptr), RecordedSlot(-1), RecordedSyncCycle(-1) {}
Error AMDGPUStreamTy::waitEvent(const AMDGPUEventTy &Event) {
   // operation's output signal is satisfied.
   if (!RecordedStream.Slots[Event.RecordedSlot].Signal->load())
     return Plugin::success();
 
   // Otherwise, make the current stream wait on the other stream's operation.
   return waitOnStreamOperation(RecordedStream, Event.RecordedSlot);
 }
 
+struct AMDGPUStreamManagerTy final
+    : GenericDeviceResourceManagerTy<AMDGPUResourceRef<AMDGPUStreamTy>> {
+  using ResourceRef = AMDGPUResourceRef<AMDGPUStreamTy>;
+  using ResourcePoolTy = GenericDeviceResourceManagerTy<ResourceRef>;
+
+  AMDGPUStreamManagerTy(GenericDeviceTy &Device, hsa_agent_t HsaAgent)
+      : GenericDeviceResourceManagerTy(Device), NextQueue(0), Agent(HsaAgent) {}
+
+  Error init(uint32_t InitialSize, int NumHsaQueues, int HsaQueueSize) {
+    Queues = std::vector<AMDGPUQueueTy>(NumHsaQueues);
+    QueueSize = HsaQueueSize;
+    MaxNumQueues = NumHsaQueues;
+    // Initialize one queue eagerly
+    if (auto Err = Queues.front().init(Agent, QueueSize))
+      return Err;
+
+    return GenericDeviceResourceManagerTy::init(InitialSize);
+  }
+
+  /// Deinitialize the resource pool and delete all resources. This function
+  /// must be called before the destructor.
+  Error deinit() override {
+    // De-init all queues
+    for (AMDGPUQueueTy &Queue : Queues) {
+      if (auto Err = Queue.deinit())
+        return Err;
+    }
+
+    return GenericDeviceResourceManagerTy::deinit();
+  }
+
+  /// Get resource from the pool or create new resources.
+  ResourceRef getResource() override {
+    ResourceRef Resource = GenericDeviceResourceManagerTy::getResource();
+    // Assign an ideally idle queue to the stream
+    assignNextQueue(Resource);
+    return Resource;
+  }
+
+  /// Return resource to the pool.
+  void returnResource(ResourceRef Resource) override {
+    std::unique_lock<std::mutex> Lock(Mutex);
+    (*Resource).Queue->removeUser();
+    Lock.unlock();
+    GenericDeviceResourceManagerTy::returnResource(Resource);
+  }
+
+private:
+  /// Search for and assign an prefereably idle queue to the given Stream. If
+  /// there is no queue without current users, resort to round robin selection.
+  inline void assignNextQueue(ResourceRef Resource) {
+    const std::lock_guard<std::mutex> Lock(Mutex);
+    uint32_t StartIndex = NextQueue % MaxNumQueues;
+    AMDGPUQueueTy *Q = nullptr;
+
+    for (int i = 0; i < MaxNumQueues; ++i) {
+      Q = &Queues[StartIndex++];
+      if (StartIndex == MaxNumQueues)
+        StartIndex = 0;
+      if (Q->isBusy())
+        continue;
+      else {
+        if (auto Err = Q->initLazy(Agent, QueueSize)) {
+          REPORT("Failure during queue init: %s\n",
+                 toString(std::move(Err)).data());
+          Q = &Queues[0];
+        }
+
+        Q->addUser();
+        (*Resource).Queue = Q;
+        return;
+      }
+    }
+
+    // All queues busy: Round robin (StartIndex has the original value again)
+    Queues[StartIndex].addUser();
+    (*Resource).Queue = &Queues[StartIndex];
+    ++NextQueue;
+  }
+
+  /// The next queue index to use for round robin selection.
+  uint32_t NextQueue;
+
+  /// The queues which are assigned to requested streams.
+  std::vector<AMDGPUQueueTy> Queues;
+
+  /// The corresponding device as HSA agent.
+  hsa_agent_t Agent;
+
+  /// The maximum number of queues.
+  int MaxNumQueues;
+
+  /// The size of created queues.
+  int QueueSize;
+};
+
 /// Abstract class that holds the common members of the actual kernel devices
 /// and the host device. Both types should inherit from this class.
 struct AMDGenericDeviceTy {
   AMDGenericDeviceTy() {}
 
   virtual ~AMDGenericDeviceTy() {}
 
   /// Create all memory pools which the device has access to and classify them.
   Error initMemoryPools() {
     // Retrieve all memory pools from the device agent(s).
     Error Err = retrieveAllMemoryPools();
     if (Err)
       return Err;
 
     for (AMDGPUMemoryPoolTy *MemoryPool : AllMemoryPools) {
       // Initialize the memory pool and retrieve some basic info.
       Error Err = MemoryPool->init();
       if (Err)
         return Err;
 
       if (!MemoryPool->isGlobal())
         continue;
 
AMDGPUDeviceTy(int32_t DeviceId, int32_t NumDevices,
         OMPX_NumQueues("LIBOMPTARGET_AMDGPU_NUM_HSA_QUEUES", 4),
         OMPX_QueueSize("LIBOMPTARGET_AMDGPU_HSA_QUEUE_SIZE", 512),
         OMPX_DefaultTeamsPerCU("LIBOMPTARGET_AMDGPU_TEAMS_PER_CU", 4),
         OMPX_MaxAsyncCopyBytes("LIBOMPTARGET_AMDGPU_MAX_ASYNC_COPY_BYTES",
                                1 * 1024 * 1024), // 1MB
         OMPX_InitialNumSignals("LIBOMPTARGET_AMDGPU_NUM_INITIAL_HSA_SIGNALS",
                                64),
         OMPX_StreamBusyWait("LIBOMPTARGET_AMDGPU_STREAM_BUSYWAIT", 2000000),
-        AMDGPUStreamManager(*this), AMDGPUEventManager(*this),
-        AMDGPUSignalManager(*this), Agent(Agent), HostDevice(HostDevice),
-        Queues() {}
+        AMDGPUStreamManager(*this, Agent), AMDGPUEventManager(*this),
+        AMDGPUSignalManager(*this), Agent(Agent), HostDevice(HostDevice) {}
 
   ~AMDGPUDeviceTy() {}
 
   /// Initialize the device, its resources and get its properties.
   Error initImpl(GenericPluginTy &Plugin) override {
     // First setup all the memory pools.
     if (auto Err = initMemoryPools())
       return Err;
Error initImpl(GenericPluginTy &Plugin) override {
     uint32_t MaxQueues;
     if (auto Err = getDeviceAttr(HSA_AGENT_INFO_QUEUES_MAX, MaxQueues))
       return Err;
 
     // Compute the number of queues and their size.
     OMPX_NumQueues = std::max(1U, std::min(OMPX_NumQueues.get(), MaxQueues));
     OMPX_QueueSize = std::min(OMPX_QueueSize.get(), MaxQueueSize);
 
-    // Construct and initialize each device queue.
-    Queues = std::vector<AMDGPUQueueTy>(OMPX_NumQueues);
-    // Initialize one queue eagerly.
-    if (auto Err = Queues.front().init(Agent, OMPX_QueueSize))
-      return Err;
-
     // Initialize stream pool.
-    if (auto Err = AMDGPUStreamManager.init(OMPX_InitialNumStreams))
+    if (auto Err = AMDGPUStreamManager.init(OMPX_InitialNumStreams,
+                                            OMPX_NumQueues, OMPX_QueueSize))
       return Err;
 
     // Initialize event pool.
     if (auto Err = AMDGPUEventManager.init(OMPX_InitialNumEvents))
       return Err;
 
     // Initialize signal pool.
     if (auto Err = AMDGPUSignalManager.init(OMPX_InitialNumSignals))
if (!LoadedImages.empty()) {
             static_cast<AMDGPUDeviceImageTy &>(*Image);
 
         // Unload the executable of the image.
         if (auto Err = AMDImage.unloadExecutable())
           return Err;
       }
     }
 
-    for (AMDGPUQueueTy &Queue : Queues) {
-      if (auto Err = Queue.deinit())
-        return Err;
-    }
-
     // Invalidate agent reference.
     Agent = {0};
 
     return Plugin::success();
   }
 
   const uint64_t getStreamBusyWaitMicroseconds() const {
     return OMPX_StreamBusyWait;
return utils::iterateAgentMemoryPools(
           AMDGPUMemoryPoolTy *MemoryPool =
               Plugin::get().allocate<AMDGPUMemoryPoolTy>();
           new (MemoryPool) AMDGPUMemoryPoolTy(HSAMemoryPool);
           AllMemoryPools.push_back(MemoryPool);
           return HSA_STATUS_SUCCESS;
         });
   }
 
-  /// Get the next queue in a round-robin fashion, includes lazy initialization.
-  AMDGPUQueueTy &getNextQueue() {
-    uint32_t Current = NextQueue.fetch_add(1, std::memory_order_relaxed);
-    uint32_t Idx = Current % Queues.size();
-    auto &Queue = Queues[Idx];
-    // Only queue 0 has been initialized eagerly. Others might need lazy/late
-    // initialization.
-    if (Idx == 0)
-      return Queue;
-
-    if (auto Err = Queue.initLazy(Agent, OMPX_QueueSize)) {
-      // Gracefully handle late initialization errors, but report them anyway.
-      REPORT("%s\n", toString(std::move(Err)).data());
-      return Queues[0];
-    }
-    return Queue;
-  }
-
 private:
-  using AMDGPUStreamRef = AMDGPUResourceRef<AMDGPUStreamTy>;
   using AMDGPUEventRef = AMDGPUResourceRef<AMDGPUEventTy>;
-
-  using AMDGPUStreamManagerTy = GenericDeviceResourceManagerTy<AMDGPUStreamRef>;
   using AMDGPUEventManagerTy = GenericDeviceResourceManagerTy<AMDGPUEventRef>;
 
   /// Envar for controlling the number of HSA queues per device. High number of
   /// queues may degrade performance.
   UInt32Envar OMPX_NumQueues;
 
   /// Envar for controlling the size of each HSA queue. The size is the number
   /// of HSA packets a queue is expected to hold. It is also the number of HSA
private:
   /// The GPU architecture.
   std::string ComputeUnitKind;
 
   /// The frequency of the steady clock inside the device.
   uint64_t ClockFrequency;
 
   /// Reference to the host device.
   AMDHostDeviceTy &HostDevice;
-
-  /// List of device packet queues.
-  std::vector<AMDGPUQueueTy> Queues;
-
-  // The next queue to be used for a new stream.
-  std::atomic<uint32_t> NextQueue = {0};
 };
 
 Error AMDGPUDeviceImageTy::loadExecutable(const AMDGPUDeviceTy &Device) {
   hsa_status_t Status;
   Status = hsa_code_object_deserialize(getStart(), getSize(), "", &CodeObject);
   if (auto Err =
           Plugin::check(Status, "Error in hsa_code_object_deserialize: %s"))
     return Err;
Error AMDGPUResourceRef<ResourceTy>::create(GenericDeviceTy &Device) {
   AMDGPUDeviceTy &AMDGPUDevice = static_cast<AMDGPUDeviceTy &>(Device);
 
   Resource = new ResourceTy(AMDGPUDevice);
 
   return Resource->init();
 }
 
 AMDGPUStreamTy::AMDGPUStreamTy(AMDGPUDeviceTy &Device)
-    : Agent(Device.getAgent()), Queue(Device.getNextQueue()),
+    : Agent(Device.getAgent()), Queue(nullptr),
       SignalManager(Device.getSignalManager()),
       // Initialize the std::deque with some empty positions.
       Slots(32), NextSlot(0), SyncCycle(0),
       StreamBusyWaitMicroseconds(Device.getStreamBusyWaitMicroseconds()) {}
 
 /// Class implementing the AMDGPU-specific functionalities of the global
 /// handler.
 struct AMDGPUGlobalHandlerTy final : public GenericGlobalHandlerTy {

openmp/libomptarget/plugins-nextgen/common/PluginInterface/PluginInterface.h

public:
   /// Initialize the resource pool.
   Error init(uint32_t InitialSize) {
     assert(ResourcePool.empty() && "Resource pool already initialized");
     return ResourcePoolTy::resizeResourcePool(InitialSize);
   }
 
   /// Deinitialize the resource pool and delete all resources. This function
   /// must be called before the destructor.
-  Error deinit() {
+  virtual Error deinit() {
     if (NextAvailable)
       DP("Missing %d resources to be returned\n", NextAvailable);
 
     // TODO: This prevents a bug on libomptarget to make the plugins fail. There
     // may be some resources not returned. Do not destroy these ones.
     if (auto Err = ResourcePoolTy::resizeResourcePool(NextAvailable))
       return Err;
 
     ResourcePool.clear();
 
     return Plugin::success();
   }
 
   /// Get resource from the pool or create new resources.
-  ResourceRef getResource() {
+  virtual ResourceRef getResource() {
     const std::lock_guard<std::mutex> Lock(Mutex);
 
     assert(NextAvailable <= ResourcePool.size() &&
            "Resource pool is corrupted");
 
     if (NextAvailable == ResourcePool.size()) {
       // By default we double the resource pool every time.
       if (auto Err = ResourcePoolTy::resizeResourcePool(NextAvailable * 2)) {
         REPORT("Failure to resize the resource pool: %s",
                toString(std::move(Err)).data());
         // Return an empty reference.
         return ResourceRef();
       }
     }
     return ResourcePool[NextAvailable++];
   }
 
   /// Return resource to the pool.
-  void returnResource(ResourceRef Resource) {
+  virtual void returnResource(ResourceRef Resource) {
     const std::lock_guard<std::mutex> Lock(Mutex);
 
     assert(NextAvailable > 0 && "Resource pool is corrupted");
     ResourcePool[--NextAvailable] = Resource;
   }
 
-private:
+protected:
   /// The resources between \p OldSize and \p NewSize need to be created or
   /// destroyed. The mutex is locked when this function is called.
   Error resizeResourcePoolImpl(uint32_t OldSize, uint32_t NewSize) {
     assert(OldSize != NewSize && "Resizing to the same size");
 
     if (auto Err = Device.setContext())
       return Err;