SWDEV-232918

hipEventRecord is much slower in hipclang/vdi

- Make sure default streams don't sync each other.
- Add null stream into the list of default streams.
- Code clean-up to simplify queue look-up.

Change-Id: I36e1fc8d86a600e3dce806694d95d146ed8afd03


[ROCm/hip commit: f7f7337bae]

projects/hip/vdi/hip_context.cpp

@@ -82,42 +82,17 @@ amd::HostQueue* getQueue(hipStream_t stream) {
  if (stream == nullptr) {
     return getNullStream();
   } else {
-    hip::Stream* s = reinterpret_cast<hip::Stream*>(stream);
-    // Wait for null stream
-    if ((s->flags & hipStreamNonBlocking) == 0) {
-      amd::HostQueue* nullStream = getNullStream();
-      amd::Command::EventWaitList eventWaitList;
-
-      amd::Command* command = nullStream->getLastQueuedCommand(true);
-      if ((command != nullptr) &&
-        // Check the current active status
-          (command->status() != CL_COMPLETE)) {
-        eventWaitList.push_back(command);
-      }
-
-      // Check if we have to wait anything
-      if (eventWaitList.size() > 0) {
-        amd::Command* command = new amd::Marker(*s->asHostQueue(), false, eventWaitList);
-        if (command != nullptr) {
-          command->enqueue();
-          command->release();
-        }
-      }
-
-      // Release all active commands. It's safe after the marker was enqueued
-      for (const auto& it : eventWaitList) {
-        it->release();
-      }
-    }
-
-    return s->asHostQueue();
+    constexpr bool WaitNullStreamOnly = true;
+    amd::HostQueue* queue = reinterpret_cast<hip::Stream*>(stream)->asHostQueue();
+    iHipWaitActiveStreams(queue, WaitNullStreamOnly);
+    return queue;
   }
 }
 
 amd::HostQueue* getNullStream(amd::Context& ctx) {
  for (auto& it : g_devices) {
    if (it->asContext() == &ctx) {
-     return it->defaultStream();
+     return it->NullStream();
    }
  }
  return nullptr;
@@ -125,7 +100,7 @@ amd::HostQueue* getNullStream(amd::Context& ctx) {
 
 amd::HostQueue* getNullStream() {
   Device* device = getCurrentDevice();
-  return device ? device->defaultStream() : nullptr;
+  return device ? device->NullStream() : nullptr;
 }
 
 };

projects/hip/vdi/hip_device.cpp

@@ -24,20 +24,14 @@
 
 namespace hip {
 
-amd::HostQueue* Device::defaultStream() {
-  if (defaultStream_ == nullptr) {
-    const cl_command_queue_properties properties = CL_QUEUE_PROFILING_ENABLE;
-    defaultStream_ = new amd::HostQueue(*asContext(), *devices()[0], properties,
-                                        amd::CommandQueue::RealTimeDisabled,
-                                        amd::CommandQueue::Priority::Normal);
-    if ((defaultStream_ == nullptr) ||
-        !defaultStream_->create()) {
-      return nullptr;
-    }
+amd::HostQueue* Device::NullStream() {
+  amd::HostQueue* null_queue = null_stream_.asHostQueue();
+  if (null_queue == nullptr) {
+    return nullptr;
   }
   // Wait for all active streams before executing commands on the default
-  iHipWaitActiveStreams(defaultStream_);
-  return defaultStream_;
+  iHipWaitActiveStreams(null_queue);
+  return null_queue;
 }
 
 };

projects/hip/vdi/hip_internal.hpp

@@ -78,6 +78,31 @@ class accelerator_view;
 };
 
 namespace hip {
+  class Device;
+
+  class Stream {
+    amd::HostQueue* queue_;
+    mutable amd::Monitor lock_;
+    Device* device_;
+    amd::CommandQueue::Priority priority_;
+    unsigned int flags_;
+    bool null_;
+
+  public:
+    Stream(Device* dev, amd::CommandQueue::Priority p, unsigned int f = 0, bool null_stream = false);
+    bool create();
+    amd::HostQueue* asHostQueue();
+    void destroy();
+    void finish() const;
+    /// Get device ID associated with the current stream;
+    int DeviceId() const;
+    /// Returns if stream is null stream
+    bool Null() const { return null_; }
+    /// Returns the lock object for the current stream
+    amd::Monitor& Lock() const { return lock_; }
+    /// Returns the creation flags for the current stream
+    unsigned int Flags() const { return flags_; }
+  };
 
   /// HIP Device class
   class Device {
@@ -85,14 +110,17 @@ namespace hip {
     /// VDI context
     amd::Context* context_;
     /// VDI host queue for default streams
-    amd::HostQueue* defaultStream_ = nullptr;
+    Stream null_stream_;
     /// Device's ID
     /// Store it here so we don't have to loop through the device list every time
     int deviceId_;
     //Maintain list of user enabled peers
     std::list<int> userEnabledPeers;
+
   public:
-    Device(amd::Context* ctx, int devId): context_(ctx), deviceId_(devId) { assert(ctx != nullptr); }
+    Device(amd::Context* ctx, int devId):
+      context_(ctx), deviceId_(devId), null_stream_(this, amd::CommandQueue::Priority::Normal, 0, true)
+        { assert(ctx != nullptr); }
     ~Device() {}
 
     amd::Context* asContext() const { return context_; }
@@ -119,7 +147,7 @@ namespace hip {
         return hipErrorPeerAccessNotEnabled;
       }
     }
-    amd::HostQueue* defaultStream();
+    amd::HostQueue* NullStream();
   };
 
   extern std::once_flag g_ihipInitialized;
@@ -154,20 +182,6 @@ namespace hip {
     static Function* asFunction(hipFunction_t f) { return reinterpret_cast<Function*>(f); }
   };
 
-  struct Stream {
-    amd::HostQueue* queue;
-    amd::Monitor lock;
-    Device* device;
-    amd::CommandQueue::Priority priority;
-    unsigned int flags;
-
-    Stream(Device* dev, amd::CommandQueue::Priority p, unsigned int f);
-    void create();
-    amd::HostQueue* asHostQueue();
-    void destroy();
-    void finish();
-  };
-
 };
 
 struct ihipExec_t {
@@ -300,7 +314,7 @@ public:
 
 /// Wait all active streams on the blocking queue. The method enqueues a wait command and
 /// doesn't stall the current thread
-extern void iHipWaitActiveStreams(amd::HostQueue* blocking_queue);
+extern void iHipWaitActiveStreams(amd::HostQueue* blocking_queue, bool wait_null_stream = false);
 
 extern std::vector<hip::Device*> g_devices;
 extern hipError_t ihipDeviceGetCount(int* count);

projects/hip/vdi/hip_memory.cpp

@@ -48,10 +48,7 @@ hipError_t ihipFree(void *ptr)
   }
   if (amd::SvmBuffer::malloced(ptr)) {
     for (auto& dev : g_devices) {
-      amd::HostQueue* queue = hip::getNullStream(*dev->asContext());
-      if (queue != nullptr) {
-        queue->finish();
-      }
+      dev->NullStream()->finish();
     }
     amd::SvmBuffer::free(*hip::getCurrentDevice()->asContext(), ptr);
     return hipSuccess;
@@ -283,10 +280,7 @@ hipError_t ihipArrayDestroy(hipArray* array) {
     return hipErrorInvalidValue;
   }
   for (auto& dev : g_devices) {
-    amd::HostQueue* queue = hip::getNullStream(*dev->asContext());
-    if (queue != nullptr) {
-      queue->finish();
-    }
+    dev->NullStream()->finish();
   }
   as_amd(memObj)->release();
 
@@ -684,10 +678,7 @@ hipError_t hipHostUnregister(void* hostPtr) {
   HIP_INIT_API(hipHostUnregister, hostPtr);
 
   for (auto& dev : g_devices) {
-    amd::HostQueue* queue = hip::getNullStream(*dev->asContext());
-    if (queue != nullptr) {
-      queue->finish();
-    }
+    dev->NullStream()->finish();
   }
 
   if (amd::SvmBuffer::malloced(hostPtr)) {

projects/hip/vdi/hip_platform.cpp

@@ -733,7 +733,7 @@ extern "C" hipError_t hipLaunchByPtr(const void *hostFunction)
   PlatformState::instance().popExec(exec);
 
   hip::Stream* stream = reinterpret_cast<hip::Stream*>(exec.hStream_);
-  int deviceId = (stream != nullptr)? stream->device->deviceId() : ihipGetDevice();
+  int deviceId = (stream != nullptr)? stream->DeviceId() : ihipGetDevice();
   if (deviceId == -1) {
     DevLogPrintfError("Wrong DeviceId: %d \n", deviceId);
     HIP_RETURN(hipErrorNoDevice);
@@ -1212,7 +1212,7 @@ extern "C" hipError_t hipLaunchKernel(const void *hostFunction,
                stream);
 
   hip::Stream* s = reinterpret_cast<hip::Stream*>(stream);
-  int deviceId = (s != nullptr)? s->device->deviceId() : ihipGetDevice();
+  int deviceId = (s != nullptr)? s->DeviceId() : ihipGetDevice();
   if (deviceId == -1) {
     DevLogPrintfError("Wrong Device Id: %d \n", deviceId);
     HIP_RETURN(hipErrorNoDevice);

projects/hip/vdi/hip_stream.cpp

@@ -42,65 +42,68 @@ class StreamCallback {
 
 namespace hip {
 
-void syncStreams() {
-  amd::ScopedLock lock(streamSetLock);
+Stream::Stream(hip::Device* dev, amd::CommandQueue::Priority p,
+    unsigned int f, bool null_stream)
+  : queue_(nullptr), lock_("Stream Callback lock"), device_(dev),
+    priority_(p), flags_(f), null_(null_stream) {}
 
-  for (const auto& it : streamSet) {
-    if (it->device->deviceId() == getCurrentDevice()->deviceId()) {
-      it->finish();
-    }
-  }
-}
-
-Stream::Stream(hip::Device* dev, amd::CommandQueue::Priority p, unsigned int f) :
-  queue(nullptr), lock("Stream Callback lock"), device(dev), priority(p), flags(f) {}
-
-void Stream::create() {
+bool Stream::create() {
   cl_command_queue_properties properties = CL_QUEUE_PROFILING_ENABLE;
-  queue = new amd::HostQueue(*device->asContext(), *device->devices()[0], properties,
-                             amd::CommandQueue::RealTimeDisabled, priority);
-  assert(queue != nullptr);
-  queue->create();
+  queue_ = new amd::HostQueue(*device_->asContext(), *device_->devices()[0], properties,
+                             amd::CommandQueue::RealTimeDisabled, priority_);
+  assert(queue_ != nullptr);
+  return queue_->create();
 }
 
 amd::HostQueue* Stream::asHostQueue() {
-  if (queue == nullptr) {
-    create();
+  if (queue_ == nullptr) {
+    if (!create()) {
+      return nullptr;
+    } else if (Null()) {
+      // Make sure the null stream is inserted into the list of default/blocking streams
+      amd::ScopedLock lock(streamSetLock);
+      streamSet.insert(this);
+    }
   }
-  return queue;
+  return queue_;
 }
 
 void Stream::destroy() {
-  if (queue != nullptr) {
-    queue->release();
-    queue = nullptr;
+  if (queue_ != nullptr) {
+    queue_->release();
+    queue_ = nullptr;
   }
 }
 
-void Stream::finish() {
-  if (queue != nullptr) {
-    queue->finish();
+void Stream::finish() const {
+  if (queue_ != nullptr) {
+    queue_->finish();
   }
 }
 
+int Stream::DeviceId() const {
+  return device_->deviceId();
+}
+
 };
 
-void iHipWaitActiveStreams(amd::HostQueue* blocking_queue) {
+void iHipWaitActiveStreams(amd::HostQueue* blocking_queue, bool wait_null_stream) {
   amd::Command::EventWaitList eventWaitList;
   {
     amd::ScopedLock lock(streamSetLock);
 
-    for (const auto& it : streamSet) {
+    for (const auto& stream : streamSet) {
+      amd::HostQueue* active_queue = stream->asHostQueue();
       // If it's the current device
-      if ((it->queue != nullptr) && (&it->queue->device() == &blocking_queue->device()) &&
-          // and it's a blocking streamclan
-          ((it->flags & hipStreamNonBlocking) == 0) &&
+      if ((active_queue != nullptr) && (&active_queue->device() == &blocking_queue->device()) &&
           // and it's not the current stream
-          (it->asHostQueue() != blocking_queue)) {
+          (active_queue != blocking_queue) &&
+          // check for a wait on the null stream
+          (stream->Null() == wait_null_stream)) {
         // Get the last valid so command
-        amd::Command* command = it->asHostQueue()->getLastQueuedCommand(true);
+        amd::Command* command = active_queue->getLastQueuedCommand(true);
         if ((command != nullptr) &&
-          // Check the current active status
+            // Check the current active status
             (command->status() != CL_COMPLETE)) {
           eventWaitList.push_back(command);
         }
@@ -127,7 +130,7 @@ void CL_CALLBACK ihipStreamCallback(cl_event event, cl_int command_exec_status,
   hipError_t status = hipSuccess;
   StreamCallback* cbo = reinterpret_cast<StreamCallback*>(user_data);
   {
-    amd::ScopedLock lock(reinterpret_cast<hip::Stream*>(cbo->stream_)->lock);
+    amd::ScopedLock lock(reinterpret_cast<hip::Stream*>(cbo->stream_)->Lock());
     cbo->callBack_(cbo->stream_, status, cbo->userData_);
   }
   cbo->command_->release();
@@ -142,12 +145,8 @@ static hipError_t ihipStreamCreate(hipStream_t *stream, unsigned int flags, amd:
   }
 
   if (!(flags & hipStreamNonBlocking)) {
-    hip::syncStreams();
-
-    {
-      amd::ScopedLock lock(streamSetLock);
-      streamSet.insert(hStream);
-    }
+    amd::ScopedLock lock(streamSetLock);
+    streamSet.insert(hStream);
   }
 
   *stream = reinterpret_cast<hipStream_t>(hStream);
@@ -194,13 +193,13 @@ hipError_t hipDeviceGetStreamPriorityRange(int* leastPriority, int* greatestPrio
   return HIP_RETURN(hipSuccess);
 }
 
-hipError_t hipStreamGetFlags(hipStream_t stream, unsigned int *flags) {
+hipError_t hipStreamGetFlags(hipStream_t stream, unsigned int* flags) {
   HIP_INIT_API(hipStreamGetFlags, stream, flags);
 
   hip::Stream* hStream = reinterpret_cast<hip::Stream*>(stream);
 
-  if(flags != nullptr && hStream != nullptr) {
-    *flags = hStream->flags;
+  if (flags != nullptr && hStream != nullptr) {
+    *flags = hStream->Flags();
   } else {
     HIP_RETURN(hipErrorInvalidValue);
   }
@@ -239,13 +238,7 @@ hipError_t hipStreamDestroy(hipStream_t stream) {
 hipError_t hipStreamWaitEvent(hipStream_t stream, hipEvent_t event, unsigned int flags) {
   HIP_INIT_API(hipStreamWaitEvent, stream, event, flags);
 
-  amd::HostQueue* queue;
-
-  if (stream == nullptr) {
-    queue = hip::getNullStream();
-  } else {
-    queue = reinterpret_cast<hip::Stream*>(stream)->asHostQueue();
-  }
+  amd::HostQueue* queue = hip::getQueue(stream);
 
   if (event == nullptr) {
     HIP_RETURN(hipErrorInvalidHandle);
@@ -259,12 +252,7 @@ hipError_t hipStreamWaitEvent(hipStream_t stream, hipEvent_t event, unsigned int
 hipError_t hipStreamQuery(hipStream_t stream) {
   HIP_INIT_API(hipStreamQuery, stream);
 
-  amd::HostQueue* hostQueue;
-  if (stream == nullptr) {
-    hostQueue = hip::getNullStream();
-  } else {
-    hostQueue = reinterpret_cast<hip::Stream*>(stream)->asHostQueue();
-  }
+  amd::HostQueue* hostQueue = hip::getQueue(stream);
 
   amd::Command* command = hostQueue->getLastQueuedCommand(true);
   if (command == nullptr) {
@@ -284,8 +272,7 @@ hipError_t hipStreamAddCallback(hipStream_t stream, hipStreamCallback_t callback
                                 unsigned int flags) {
   HIP_INIT_API(hipStreamAddCallback, stream, callback, userData, flags);
 
-  amd::HostQueue* hostQueue = reinterpret_cast<hip::Stream*>
-                              (stream)->asHostQueue();
+  amd::HostQueue* hostQueue = reinterpret_cast<hip::Stream*>(stream)->asHostQueue();
   amd::Command* command = hostQueue->getLastQueuedCommand(true);
   if (command == nullptr) {
     amd::Command::EventWaitList eventWaitList;