[OpenMP] Only initialize a single queue/stream/event eagerly

[jdoerfert]

Summary

Initialization for many queues/streams/events might come at a cost even
if we do not use them. This patch lazily initializes them, otherwise
nothing (major) is supposed to change. A minor difference is the
handling of an error in the initialization of an AMD queue (other than
the first). We now report an error but continue with the first queue;
unsure if this will ever come up.

[tianshilei1992]

openmp/docs/design/Runtimes.rst

1200

FWIW, events are used more frequently than streams even for single thread offloading.

[jdoerfert]

FWIW, currently talking to @jplehr and @mhalk about the queue <-> stream mapping. Might be a follow up.

openmp/docs/design/Runtimes.rst

1200

True, not sure what the right value is. I can keep them at 32 if ppl prefer.

[mhalk]

openmp/docs/design/Runtimes.rst

1200

Just wanted to point out (since it was not directly obvious to me):
for value LIBOMPTARGET_NUM_INITIAL_STREAMS=n, getNextQueue will be called n times.

So, the patch would have to be adapted if LIBOMPTARGET_NUM_INITIAL_STREAMS will be increased.
Otherwise, one would have again n <= OMPX_NumQueues HSA queues from the get-go, even if they're not used.

[kevinsala]

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

585

Is there any reason to maintain both init and initLazy instead of making init lazy by default?

[jdoerfert]

I am unsure we need this after all. @mhalk, can u check, if we don't need this we can scrap it.

[mhalk]

Just looked through this patch and D154523.

Don't have a clear "yes"/"no".
IMO merging the two makes sense as they achieve the same: "single HSA queue eager init".

The busy tracking will revert like half of this patch's changes in amdgpu/src/rtl.cpp.
Should you tend to scrap it, I will have to incorporate ~30ish LoC from this one.
That is, the docs & setting of default/initial values (esp. outside of amdgpu/src/rtl.cpp).

[jdoerfert]

Subsumed by D154523.

openmp/docs/design/Runtimes.rst

 """"""""""""""""""""""""""""""""
 
 This environment variable sets the number of pre-created streams in the plugin
 (if supported) at initialization. More streams will be created dynamically
 throughout the execution if needed. A stream is a queue of asynchronous
 operations (e.g., kernel launches and memory copies) that are executed
 sequentially. Parallelism is achieved by featuring multiple streams. The
 ``libomptarget`` leverages streams to exploit parallelism between plugin
-operations. The default value is ``32``.
+operations. The default value is ``1``, more streams are created as needed.
 
 LIBOMPTARGET_NUM_INITIAL_EVENTS
 """""""""""""""""""""""""""""""
 
 This environment variable sets the number of pre-created events in the
 plugin (if supported) at initialization. More events will be created
 dynamically throughout the execution if needed. An event is used to synchronize
-a stream with another efficiently. The default value is ``32``.
+a stream with another efficiently. The default value is ``1``, more events are
+created as needed.
 
 LIBOMPTARGET_LOCK_MAPPED_HOST_BUFFERS
 """""""""""""""""""""""""""""""""""""
 
 This environment variable indicates whether the host buffers mapped by the user
 should be automatically locked/pinned by the plugin. Pinned host buffers allow
 true asynchronous copies between the host and devices. Enabling this feature can
 increase the performance of applications that are intensive in host-device

openmp/libomptarget/include/Utilities.h

if (const char *EnvStr = getenv(Name.data())) {
 
       if (!IsPresent) {
         DP("Ignoring invalid value %s for envar %s\n", EnvStr, Name.data());
         Data = Default;
       }
     }
   }
 
+  Envar<Ty> &operator=(const Ty &V) {
+    Data = V;
+    Initialized = true;
+    return *this;
+  }
+
   /// Get the definitive value.
   const Ty &get() const {
     // Throw a runtime error in case this envar is not initialized.
     if (!Initialized)
       FATAL_MESSAGE0(1, "Consulting envar before initialization");
 
     return Data;
   }

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

struct AMDGPUQueueTy {
   /// 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");
   }
 
   /// 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");
 
     // Lock the queue during the packet publishing process. Notice this blocks
     // the addition of other packets to the queue. The following piece of code
     // should be lightweight; do not block the thread, allocate memory, etc.
     std::lock_guard<std::mutex> Lock(Mutex);
+    assert(Queue && "Interacted with a non-initialized queue!");
 
     // Avoid defining the input dependency if already satisfied.
     if (InputSignal && !InputSignal->load())
       InputSignal = nullptr;
 
     // Add a barrier packet before the kernel packet in case there is a pending
     // preceding operation. The barrier packet will delay the processing of
     // subsequent queue's packets until the barrier input signal are satisfied.
struct AMDGPUQueueTy {
 
   /// Push a barrier packet that will wait up to two input signals. All signals
   /// are optional (nullptr if none).
   Error pushBarrier(AMDGPUSignalTy *OutputSignal,
                     const AMDGPUSignalTy *InputSignal1,
                     const AMDGPUSignalTy *InputSignal2) {
     // Lock the queue during the packet publishing process.
     std::lock_guard<std::mutex> Lock(Mutex);
+    assert(Queue && "Interacted with a non-initialized queue!");
 
     // Push the barrier with the lock acquired.
     return pushBarrierImpl(OutputSignal, InputSignal1, InputSignal2);
   }
 
 private:
   /// Push a barrier packet that will wait up to two input signals. Assumes the
   /// the queue lock is acquired.
Error initImpl(GenericPluginTy &Plugin) override {
     if (auto Err = getDeviceAttr(HSA_AGENT_INFO_QUEUE_MAX_SIZE, MaxQueueSize))
       return Err;
 
     uint32_t MaxQueues;
     if (auto Err = getDeviceAttr(HSA_AGENT_INFO_QUEUES_MAX, MaxQueues))
       return Err;
 
     // Compute the number of queues and their size.
-    const uint32_t NumQueues = std::min(OMPX_NumQueues.get(), MaxQueues);
-    const uint32_t QueueSize = std::min(OMPX_QueueSize.get(), MaxQueueSize);
+    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>(NumQueues);
-    for (AMDGPUQueueTy &Queue : Queues)
-      if (auto Err = Queue.init(Agent, QueueSize))
-        return Err;
+    Queues = std::vector<AMDGPUQueueTy>(OMPX_NumQueues);
+    // Initialize one queue eagerly.
+    if (auto Err = Queues.front().init(Agent, OMPX_QueueSize))
 
     // Initialize stream pool.
     if (auto Err = AMDGPUStreamManager.init(OMPX_InitialNumStreams))
       return Err;
 
     // Initialize event pool.
     if (auto Err = AMDGPUEventManager.init(OMPX_InitialNumEvents))
       return Err;
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.
+  /// Get the next queue in a round-robin fashion, includes lazy initialization.
   AMDGPUQueueTy &getNextQueue() {
-    static std::atomic<uint32_t> NextQueue(0);
-
     uint32_t Current = NextQueue.fetch_add(1, std::memory_order_relaxed);
-    return Queues[Current % Queues.size()];
+    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>;
private:
   /// The GPU architecture.
   std::string ComputeUnitKind;
 
   /// 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;

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

: MemoryManager(nullptr), OMP_TeamLimit("OMP_TEAM_LIMIT"),
       OMP_NumTeams("OMP_NUM_TEAMS"),
       OMP_TeamsThreadLimit("OMP_TEAMS_THREAD_LIMIT"),
       OMPX_DebugKind("LIBOMPTARGET_DEVICE_RTL_DEBUG"),
       OMPX_SharedMemorySize("LIBOMPTARGET_SHARED_MEMORY_SIZE"),
       // Do not initialize the following two envars since they depend on the
       // device initialization. These cannot be consulted until the device is
       // initialized correctly. We intialize them in GenericDeviceTy::init().
       OMPX_TargetStackSize(), OMPX_TargetHeapSize(),
-      // By default, the initial number of streams and events are 32.
-      OMPX_InitialNumStreams("LIBOMPTARGET_NUM_INITIAL_STREAMS", 32),
-      OMPX_InitialNumEvents("LIBOMPTARGET_NUM_INITIAL_EVENTS", 32),
+      // By default, the initial number of streams and events is 1.
+      OMPX_InitialNumStreams("LIBOMPTARGET_NUM_INITIAL_STREAMS", 1),
+      OMPX_InitialNumEvents("LIBOMPTARGET_NUM_INITIAL_EVENTS", 1),
       DeviceId(DeviceId), GridValues(OMPGridValues),
       PeerAccesses(NumDevices, PeerAccessState::PENDING), PeerAccessesLock(),
       PinnedAllocs(*this) {}
 
 Error GenericDeviceTy::init(GenericPluginTy &Plugin) {
   if (auto Err = initImpl(Plugin))
     return Err;