SWDEV-24806 - Added support for event wait and event record graph APIs

hip graph implementation for APIs AddEventRecordNode, AddEventWaitNode, EventRecordNodeGetEvent, EventWaitNodeGetEvent, EventRecordNodeSetEvent,EventWaitNodeSetEvent, ExecEventRecordNodeSetEvent, ExecEventWaitNodeSetEvent

Change-Id: Ifde51f4591067795d6cbf5595372cbdbd8a7bc26


[ROCm/clr commit: f6456c5e27]

projects/clr/hipamd/src/amdhip.def

@@ -332,3 +332,11 @@ hipGraphExecMemcpyNodeSetParamsToSymbol
 hipGraphExecMemcpyNodeSetParams
 hipGraphMemcpyNodeSetParams1D
 hipGraphExecMemcpyNodeSetParams1D
+hipGraphAddEventRecordNode
+hipGraphEventRecordNodeGetEvent
+hipGraphEventRecordNodeSetEvent
+hipGraphExecEventRecordNodeSetEvent
+hipGraphAddEventWaitNode
+hipGraphEventWaitNodeGetEvent
+hipGraphEventWaitNodeSetEvent
+hipGraphExecEventWaitNodeSetEvent

projects/clr/hipamd/src/hip_event.cpp

@@ -32,6 +32,37 @@ void ipcEventCallback(hipStream_t stream, hipError_t status, void* user_data)
   return;
 }
 
+// ================================================================================================
+bool createIpcEventShmemIfNeeded(hip::Event::ihipIpcEvent_t& ipc_evt) {
+#if !defined(_MSC_VER)
+  if (ipc_evt.ipc_shmem_) {
+    // ipc_shmem_ already created, no need to create it again
+    return true;
+  }
+  char name_template[] = "/tmp/eventXXXXXX";
+  int temp_fd = mkstemp(name_template);
+
+  ipc_evt.ipc_name_ = name_template;
+  ipc_evt.ipc_name_.replace(0, 5, "/hip_");
+  if (!amd::Os::MemoryMapFileTruncated(ipc_evt.ipc_name_.c_str(),
+      const_cast<const void**> (reinterpret_cast<void**>(&(ipc_evt.ipc_shmem_))),
+      sizeof(hip::ihipIpcEventShmem_t))) {
+    return false;
+  }
+  ipc_evt.ipc_shmem_->owners = 1;
+  ipc_evt.ipc_shmem_->read_index = -1;
+  ipc_evt.ipc_shmem_->write_index = 0;
+  for (uint32_t sig_idx = 0; sig_idx < IPC_SIGNALS_PER_EVENT; ++sig_idx) {
+    ipc_evt.ipc_shmem_->signal[sig_idx] = 0;
+  }
+
+  close(temp_fd);
+  return true;
+#else
+  return false;
+#endif
+}
+
 namespace hip {
 
 bool Event::ready() {
@@ -135,53 +166,135 @@ int64_t Event::time() const {
   }
 }
 
-hipError_t Event::streamWait(amd::HostQueue* hostQueue, uint flags) {
-  if ((event_ == nullptr) || (event_->command().queue() == hostQueue)) {
-    return hipSuccess;
+hipError_t Event::streamWaitCommand(amd::Command*& command, amd::HostQueue* queue) {
+  if (flags & hipEventInterprocess) {
+    command = new amd::Marker(*queue, false);
+  } else {
+    amd::Command::EventWaitList eventWaitList;
+    eventWaitList.push_back(event_);
+    command = new amd::Marker(*queue, kMarkerDisableFlush, eventWaitList);
   }
-
-  amd::ScopedLock lock(lock_);
-  bool retain = false;
-
-  if (!event_->notifyCmdQueue()) {
-    return hipErrorLaunchOutOfResources;
-  }
-  amd::Command::EventWaitList eventWaitList;
-  eventWaitList.push_back(event_);
-
-  amd::Command* command = new amd::Marker(*hostQueue, kMarkerDisableFlush, eventWaitList);
   if (command == NULL) {
     return hipErrorOutOfMemory;
   }
-  command->enqueue();
-  command->release();
-
   return hipSuccess;
 }
 
-void Event::addMarker(amd::HostQueue* queue, amd::Command* command, bool record) {
-  // Keep the lock always at the beginning of this to avoid a race. SWDEV-277847
-  amd::ScopedLock lock(lock_);
+hipError_t Event::enqueueStreamWaitCommand(hipStream_t stream, amd::Command* command) {
+  if (flags & hipEventInterprocess) {
+    auto t{new CallbackData{ipc_evt_.ipc_shmem_->read_index, ipc_evt_.ipc_shmem_}};
+    StreamCallback* cbo = new StreamCallback(
+        stream, reinterpret_cast<hipStreamCallback_t>(WaitThenDecrementSignal), t, command);
+    if (!command->setCallback(CL_COMPLETE, ihipStreamCallback, cbo)) {
+      command->release();
+      return hipErrorInvalidHandle;
+    }
+    command->enqueue();
+    command->awaitCompletion();
+    return hipSuccess;
+  } else {
+    command->enqueue();
+  }
+}
 
-  if (command == nullptr) {
+hipError_t Event::streamWait(hipStream_t stream, uint flags) {
+  amd::HostQueue* queue = hip::getQueue(stream);
+  if ((event_ == nullptr) || (event_->command().queue() == queue)) {
+    return hipSuccess;
+  }
+  amd::ScopedLock lock(lock_);
+  if (!this->flags & hipEventInterprocess) {
+    if (!event_->notifyCmdQueue()) {
+      return hipErrorLaunchOutOfResources;
+    }
+  }
+  amd::Command* command;
+  streamWaitCommand(command, queue);
+  enqueueStreamWaitCommand(stream, command);
+  if (!this->flags & hipEventInterprocess) {
+    command->release();
+  }
+  return hipSuccess;
+}
+
+hipError_t Event::recordCommand(amd::Command*& command, amd::HostQueue* queue) {
+  bool recorded = isRecorded();
+  if ((flags & hipEventInterprocess) && !recorded) {
+    command = new amd::Marker(*queue, kMarkerDisableFlush);
+  } else if (command == nullptr) {
     static constexpr bool kRecordExplicitGpuTs = true;
     // Always submit a EventMarker.
     command = new hip::EventMarker(*queue, !kMarkerDisableFlush, kRecordExplicitGpuTs);
+  }
+  return hipSuccess;
+}
+
+hipError_t Event::enqueueRecordCommand(hipStream_t stream, amd::Command* command, bool record) {
+  amd::HostQueue* queue = hip::getQueue(stream);
+  bool recorded = isRecorded();
+  if ((flags & hipEventInterprocess) && !recorded) {
+    amd::Event& tEvent = command->event();
+    createIpcEventShmemIfNeeded(ipc_evt_);
+    int write_index = ipc_evt_.ipc_shmem_->write_index++;
+    int offset = write_index % IPC_SIGNALS_PER_EVENT;
+    while (ipc_evt_.ipc_shmem_->signal[offset] != 0) {
+      amd::Os::sleep(1);
+    }
+    // Lock signal.
+    ipc_evt_.ipc_shmem_->signal[offset] = 1;
+    ipc_evt_.ipc_shmem_->owners_device_id = deviceId();
+
+    std::atomic<int>* signal = &ipc_evt_.ipc_shmem_->signal[offset];
+    StreamCallback* cbo = new StreamCallback(
+        stream, reinterpret_cast<hipStreamCallback_t>(ipcEventCallback), signal, command);
+    if (!tEvent.setCallback(CL_COMPLETE, ihipStreamCallback, cbo)) {
+      command->release();
+      return hipErrorInvalidHandle;
+    }
     command->enqueue();
+    tEvent.notifyCmdQueue();
+
+    // Add the new barrier to stall the stream, until the callback is done
+    amd::Command::EventWaitList eventWaitList;
+    eventWaitList.push_back(command);
+    amd::Command* block_command = new amd::Marker(*queue, !kMarkerDisableFlush, eventWaitList);
+    if (block_command == nullptr) {
+      return hipErrorInvalidValue;
+    }
+    block_command->enqueue();
+    block_command->release();
+
+    // Update read index to indicate new signal.
+    int expected = write_index - 1;
+    while (!ipc_evt_.ipc_shmem_->read_index.compare_exchange_weak(expected, write_index)) {
+      amd::Os::sleep(1);
+    }
+  } else {
+    command->enqueue();
+    if (event_ == &command->event()) return hipSuccess;
+    if (event_ != nullptr) {
+      event_->release();
+    }
+    event_ = &command->event();
+    recorded_ = record;
   }
-
-  if (event_ == &command->event()) return;
-
-  if (event_ != nullptr) {
-    event_->release();
-  }
-
-  event_ = &command->event();
-  recorded_ = record;
+  return hipSuccess;
 }
 
+hipError_t Event::addMarker(hipStream_t stream, amd::Command* command, bool record) {
+  amd::HostQueue* queue = hip::getQueue(stream);
+  // Keep the lock always at the beginning of this to avoid a race. SWDEV-277847
+  amd::ScopedLock lock(lock_);
+  hipError_t status = recordCommand(command, queue);
+  if (status != hipSuccess) {
+    return hipSuccess;
+  }
+  status = enqueueRecordCommand(stream, command, record);
+  return status;
 }
 
+}  // namespace hip
+
 hipError_t ihipEventCreateWithFlags(hipEvent_t* event, unsigned flags) {
   if (event == nullptr) {
     return hipErrorInvalidValue;
@@ -280,37 +393,6 @@ hipError_t hipEventElapsedTime(float *ms, hipEvent_t start, hipEvent_t stop) {
   HIP_RETURN(eStart->elapsedTime(*eStop, *ms), "Elapsed Time = ", *ms);
 }
 
-// ================================================================================================
-bool createIpcEventShmemIfNeeded(hip::Event::ihipIpcEvent_t& ipc_evt) {
-#if !defined(_MSC_VER)
-  if (ipc_evt.ipc_shmem_) {
-    // ipc_shmem_ already created, no need to create it again
-    return true;
-  }
-  char name_template[] = "/tmp/eventXXXXXX";
-  int temp_fd = mkstemp(name_template);
-
-  ipc_evt.ipc_name_ = name_template;
-  ipc_evt.ipc_name_.replace(0, 5, "/hip_");
-  if (!amd::Os::MemoryMapFileTruncated(ipc_evt.ipc_name_.c_str(),
-      const_cast<const void**> (reinterpret_cast<void**>(&(ipc_evt.ipc_shmem_))),
-      sizeof(hip::ihipIpcEventShmem_t))) {
-    return false;
-  }
-  ipc_evt.ipc_shmem_->owners = 1;
-  ipc_evt.ipc_shmem_->read_index = -1;
-  ipc_evt.ipc_shmem_->write_index = 0;
-  for (uint32_t sig_idx = 0; sig_idx < IPC_SIGNALS_PER_EVENT; ++sig_idx) {
-    ipc_evt.ipc_shmem_->signal[sig_idx] = 0;
-  }
-
-  close(temp_fd);
-  return true;
-#else
-  return false;
-#endif
-}
-
 // ================================================================================================
 hipError_t hipEventRecord(hipEvent_t event, hipStream_t stream) {
   HIP_INIT_API(hipEventRecord, event, stream);
@@ -329,49 +411,7 @@ hipError_t hipEventRecord(hipEvent_t event, hipStream_t stream) {
     HIP_RETURN(hipErrorInvalidHandle);
   }
 
-  bool isRecorded = e->isRecorded();
-  if ((e->flags & hipEventInterprocess) && !isRecorded) {
-    amd::Command* command = new amd::Marker(*queue, kMarkerDisableFlush);
-    amd::Event& tEvent = command->event();
-    createIpcEventShmemIfNeeded(e->ipc_evt_);
-    int write_index = e->ipc_evt_.ipc_shmem_->write_index++;
-    int offset = write_index % IPC_SIGNALS_PER_EVENT;
-    while (e->ipc_evt_.ipc_shmem_->signal[offset] != 0) {
-      amd::Os::sleep(1);
-    }
-    // Lock signal.
-    e->ipc_evt_.ipc_shmem_->signal[offset] = 1;
-    e->ipc_evt_.ipc_shmem_->owners_device_id = e->deviceId();
-
-    std::atomic<int> *signal = &e->ipc_evt_.ipc_shmem_->signal[offset];
-    StreamCallback* cbo = new StreamCallback(stream,
-                    reinterpret_cast<hipStreamCallback_t> (ipcEventCallback), signal, command);
-    if (!tEvent.setCallback(CL_COMPLETE, ihipStreamCallback,cbo)) {
-      command->release();
-      return hipErrorInvalidHandle;
-    }
-    command->enqueue();
-    tEvent.notifyCmdQueue();
-
-    // Add the new barrier to stall the stream, until the callback is done
-    amd::Command::EventWaitList eventWaitList;;
-    eventWaitList.push_back(command);
-    amd::Command* block_command = new amd::Marker(*queue, !kMarkerDisableFlush, eventWaitList);
-    if (block_command == nullptr) {
-      return hipErrorInvalidValue;
-    }
-    block_command->enqueue();
-    block_command->release();
-
-    // Update read index to indicate new signal.
-    int expected = write_index - 1;
-    while (!e->ipc_evt_.ipc_shmem_->read_index.compare_exchange_weak(expected, write_index)) {
-      amd::Os::sleep(1);
-    }
-  } else {
-    e->addMarker(queue, nullptr, true);
-  }
-  HIP_RETURN(hipSuccess);
+  HIP_RETURN(e->addMarker(stream, nullptr, true));
 }
 
 // ================================================================================================

projects/clr/hipamd/src/hip_event.hpp

@@ -90,9 +90,15 @@ class Event {
   hipError_t query();
   hipError_t synchronize();
   hipError_t elapsedTime(Event& stop, float& ms);
-  hipError_t streamWait(amd::HostQueue* queue, uint flags);
 
-  void addMarker(amd::HostQueue* queue, amd::Command* command, bool record);
+  hipError_t streamWaitCommand(amd::Command*& command, amd::HostQueue* queue);
+  hipError_t enqueueStreamWaitCommand(hipStream_t stream, amd::Command* command);
+  hipError_t streamWait(hipStream_t stream, uint flags);
+
+  hipError_t recordCommand(amd::Command*& command, amd::HostQueue* queue);
+  hipError_t enqueueRecordCommand(hipStream_t stream, amd::Command* command, bool record);
+  hipError_t addMarker(hipStream_t stream, amd::Command* command, bool record);
+
   bool isRecorded() { return recorded_; }
   amd::Monitor& lock() { return lock_; }
   const int deviceId() { return device_id_; }
@@ -149,4 +155,9 @@ private:
 
 };
 
+struct CallbackData {
+  int previous_read_index;
+  hip::ihipIpcEventShmem_t* shmem;
+};
+
 #endif // HIP_EVEMT_H

projects/clr/hipamd/src/hip_graph.cpp

@@ -28,7 +28,7 @@ thread_local std::vector<hipStream_t> g_captureStreams;
 std::unordered_map<amd::Command*, hipGraphExec_t> hipGraphExec::activeGraphExec_;
 
 inline void ihipGraphAddNode(hipGraphNode_t graphNode, hipGraph_t graph,
-                            const hipGraphNode_t* pDependencies, size_t numDependencies) {
+                             const hipGraphNode_t* pDependencies, size_t numDependencies) {
   graph->AddNode(graphNode);
   for (size_t i = 0; i < numDependencies; i++) {
     pDependencies[i]->AddEdge(graphNode);
@@ -925,3 +925,82 @@ hipError_t hipGraphExecMemcpyNodeSetParamsToSymbol(hipGraphExec_t hGraphExec, hi
   HIP_RETURN(reinterpret_cast<hipGraphMemcpyNodeToSymbol*>(node)->SetCommandParams(
       symbol, src, count, offset, kind));
 }
+
+hipError_t hipGraphAddEventRecordNode(hipGraphNode_t* pGraphNode, hipGraph_t graph,
+                                      const hipGraphNode_t* pDependencies, size_t numDependencies,
+                                      hipEvent_t event) {
+  HIP_INIT_API(hipGraphAddEventRecordNode, pGraphNode, graph, pDependencies, numDependencies,
+               event);
+  if (graph == nullptr || (numDependencies > 0 && pDependencies == nullptr) || event == nullptr) {
+    HIP_RETURN(hipErrorInvalidValue);
+  }
+  *pGraphNode = new hipGraphEventRecordNode(event);
+  ihipGraphAddNode(*pGraphNode, graph, pDependencies, numDependencies);
+  HIP_RETURN(hipSuccess);
+}
+
+hipError_t hipGraphEventRecordNodeGetEvent(hipGraphNode_t node, hipEvent_t* event_out) {
+  HIP_INIT_API(hipGraphEventRecordNodeGetEvent, node, event_out);
+  if (node == nullptr || event_out == nullptr) {
+    HIP_RETURN(hipErrorInvalidValue);
+  }
+  reinterpret_cast<hipGraphEventRecordNode*>(node)->GetParams(event_out);
+  HIP_RETURN(hipSuccess);
+}
+
+hipError_t hipGraphEventRecordNodeSetEvent(hipGraphNode_t node, hipEvent_t event) {
+  HIP_INIT_API(hipGraphEventRecordNodeSetEvent, node, event);
+  if (node == nullptr || event == nullptr) {
+    HIP_RETURN(hipErrorInvalidValue);
+  }
+  reinterpret_cast<hipGraphEventRecordNode*>(node)->SetParams(event);
+  HIP_RETURN(hipSuccess);
+}
+
+hipError_t hipGraphExecEventRecordNodeSetEvent(hipGraphExec_t hGraphExec, hipGraphNode_t hNode,
+                                               hipEvent_t event) {
+  HIP_INIT_API(hipGraphExecEventRecordNodeSetEvent, hGraphExec, hNode, event);
+  if (hGraphExec == nullptr || hNode == nullptr || event == nullptr) {
+    HIP_RETURN(hipErrorInvalidValue);
+  }
+  HIP_RETURN(reinterpret_cast<hipGraphEventRecordNode*>(hNode)->SetExecParams(event));
+}
+
+hipError_t hipGraphAddEventWaitNode(hipGraphNode_t* pGraphNode, hipGraph_t graph,
+                                    const hipGraphNode_t* pDependencies, size_t numDependencies,
+                                    hipEvent_t event) {
+  HIP_INIT_API(hipGraphAddEventWaitNode, pGraphNode, graph, pDependencies, numDependencies, event);
+  if (graph == nullptr || (numDependencies > 0 && pDependencies == nullptr) || event == nullptr) {
+    HIP_RETURN(hipErrorInvalidValue);
+  }
+  *pGraphNode = new hipGraphEventWaitNode(event);
+  ihipGraphAddNode(*pGraphNode, graph, pDependencies, numDependencies);
+  HIP_RETURN(hipSuccess);
+}
+
+hipError_t hipGraphEventWaitNodeGetEvent(hipGraphNode_t node, hipEvent_t* event_out) {
+  HIP_INIT_API(hipGraphEventWaitNodeGetEvent, node, event_out);
+  if (node == nullptr || *event_out == nullptr) {
+    HIP_RETURN(hipErrorInvalidValue);
+  }
+  reinterpret_cast<hipGraphEventWaitNode*>(node)->GetParams(event_out);
+  HIP_RETURN(hipSuccess);
+}
+
+hipError_t hipGraphEventWaitNodeSetEvent(hipGraphNode_t node, hipEvent_t event) {
+  HIP_INIT_API(hipGraphEventWaitNodeSetEvent, node, event);
+  if (node == nullptr || event == nullptr) {
+    HIP_RETURN(hipErrorInvalidValue);
+  }
+  reinterpret_cast<hipGraphEventWaitNode*>(node)->SetParams(event);
+  HIP_RETURN(hipSuccess);
+}
+
+hipError_t hipGraphExecEventWaitNodeSetEvent(hipGraphExec_t hGraphExec, hipGraphNode_t hNode,
+                                             hipEvent_t event) {
+  HIP_INIT_API(hipGraphExecEventWaitNodeSetEvent, hGraphExec, hNode, event);
+  if (hGraphExec == nullptr || hNode == nullptr || event == nullptr) {
+    HIP_RETURN(hipErrorInvalidValue);
+  }
+  HIP_RETURN(reinterpret_cast<hipGraphEventRecordNode*>(hNode)->SetExecParams(event));
+}

projects/clr/hipamd/src/hip_graph_internal.cpp

@@ -578,7 +578,7 @@ hipError_t hipGraphExec::Run(hipStream_t stream) {
   }
   rootCommand_->enqueue();
   for (auto& node : levelOrder_) {
-    node->EnqueueCommands();
+    node->EnqueueCommands(stream);
   }
 
   amd::Command* command = new amd::Marker(*queue, false, graphLastCmdWaitList_);

projects/clr/hipamd/src/hip_graph_internal.hpp

@@ -30,6 +30,7 @@
 #include "hip/hip_runtime.h"
 #include "hip_internal.hpp"
 #include "hip_graph_helper.hpp"
+#include "hip_event.hpp"
 
 typedef hipGraphNode* Node;
 hipError_t ihipValidateKernelParams(const hipKernelNodeParams* pNodeParams);
@@ -50,15 +51,14 @@ struct hipGraphNode {
   struct ihipGraph* parentGraph_;
 
  public:
-  hipGraphNode(hipGraphNodeType type) {
-    type_ = type;
-    level_ = 0;
-    visited_ = false;
-    inDegree_ = 0;
-    outDegree_ = 0;
-    id_ = nextID++;
-    parentGraph_ = nullptr;
-  }
+  hipGraphNode(hipGraphNodeType type)
+      : type_(type),
+        level_(0),
+        visited_(false),
+        inDegree_(0),
+        outDegree_(0),
+        id_(nextID++),
+        parentGraph_(nullptr) {}
   /// Copy Constructor
   hipGraphNode(const hipGraphNode& node) {
     level_ = node.level_;
@@ -160,7 +160,7 @@ struct hipGraphNode {
   /// Update waitlist of the nodes embedded as part of the graphnode(e.g. ChildGraph)
   virtual void UpdateEventWaitLists() {}
   /// Enqueue commands part of the node
-  virtual void EnqueueCommands() {
+  virtual void EnqueueCommands(hipStream_t stream) {
     for (auto& command : commands_) {
       command->enqueue();
     }
@@ -285,13 +285,13 @@ struct hipChildGraphNode : public hipGraphNode {
 
   void LevelOrder(std::vector<Node>& levelOrder) { childGraph_->LevelOrder(levelOrder); }
 
-  void EnqueueCommands() {
+  void EnqueueCommands(hipStream_t stream) {
     if (commands_.size() == 2) {
       // enqueue child graph start command
       commands_[0]->enqueue();
       // enqueue nodes in child graph in level order
       for (auto& node : childGraphlevelOrder_) {
-        node->EnqueueCommands();
+        node->EnqueueCommands(stream);
       }
       // enqueue child graph end command
       commands_[1]->enqueue();
@@ -595,10 +595,40 @@ class hipGraphEventRecordNode : public hipGraphNode {
     return new hipGraphEventRecordNode(static_cast<hipGraphEventRecordNode const&>(*this));
   }
 
-  hipError_t CreateCommand(amd::HostQueue* queue);
+  hipError_t CreateCommand(amd::HostQueue* queue) {
+    hip::Event* e = reinterpret_cast<hip::Event*>(event_);
+    commands_.reserve(1);
+    amd::Command* command;
+    hipError_t status = e->recordCommand(command, queue);
+    commands_.emplace_back(command);
+    return status;
+  }
+
+  void EnqueueCommands(hipStream_t stream) {
+    if (!commands_.empty()) {
+      hip::Event* e = reinterpret_cast<hip::Event*>(event_);
+      hipError_t status = e->enqueueRecordCommand(stream, commands_[0], true);
+      if (status != hipSuccess) {
+        ClPrint(amd::LOG_ERROR, amd::LOG_CODE,
+                "[hipGraph] enqueue event record command failed for node %p - status %d\n", this,
+                status);
+      }
+    }
+  }
 
   void GetParams(hipEvent_t* event) { *event = event_; }
+
   void SetParams(hipEvent_t event) { event_ = event; }
+
+  hipError_t SetExecParams(hipEvent_t event) {
+    amd::HostQueue* queue;
+    if (!commands_.empty()) {
+      queue = commands_[0]->queue();
+      commands_[0]->release();
+    }
+    commands_.clear();
+    return CreateCommand(queue);
+  }
 };
 
 class hipGraphEventWaitNode : public hipGraphNode {
@@ -613,10 +643,40 @@ class hipGraphEventWaitNode : public hipGraphNode {
     return new hipGraphEventWaitNode(static_cast<hipGraphEventWaitNode const&>(*this));
   }
 
-  hipError_t CreateCommand(amd::HostQueue* queue);
+  hipError_t CreateCommand(amd::HostQueue* queue) {
+    hip::Event* e = reinterpret_cast<hip::Event*>(event_);
+    commands_.reserve(1);
+    amd::Command* command;
+    hipError_t status = e->streamWaitCommand(command, queue);
+    commands_.emplace_back(command);
+    return status;
+  }
+
+  void EnqueueCommands(hipStream_t stream) {
+    if (!commands_.empty()) {
+      hip::Event* e = reinterpret_cast<hip::Event*>(event_);
+      hipError_t status = e->enqueueStreamWaitCommand(stream, commands_[0]);
+      if (status != hipSuccess) {
+        ClPrint(amd::LOG_ERROR, amd::LOG_CODE,
+                "[hipGraph] enqueue stream wait command failed for node %p - status %d\n", this,
+                status);
+      }
+    }
+  }
 
   void GetParams(hipEvent_t* event) { *event = event_; }
+
   void SetParams(hipEvent_t event) { event_ = event; }
+
+  hipError_t SetExecParams(hipEvent_t event) {
+    amd::HostQueue* queue;
+    if (!commands_.empty()) {
+      queue = commands_[0]->queue();
+      commands_[0]->release();
+    }
+    commands_.clear();
+    return CreateCommand(queue);
+  }
 };
 
 class hipGraphHostNode : public hipGraphNode {

projects/clr/hipamd/src/hip_hcc.def.in

@@ -330,3 +330,11 @@ hipGraphExecMemcpyNodeSetParamsToSymbol
 hipGraphExecMemcpyNodeSetParams
 hipGraphMemcpyNodeSetParams1D
 hipGraphExecMemcpyNodeSetParams1D
+hipGraphAddEventRecordNode
+hipGraphEventRecordNodeGetEvent
+hipGraphEventRecordNodeSetEvent
+hipGraphExecEventRecordNodeSetEvent
+hipGraphAddEventWaitNode
+hipGraphEventWaitNodeGetEvent
+hipGraphEventWaitNodeSetEvent
+hipGraphExecEventWaitNodeSetEvent

projects/clr/hipamd/src/hip_hcc.map.in

@@ -356,6 +356,14 @@ global:
     hipGraphExecMemcpyNodeSetParams;
     hipGraphMemcpyNodeSetParams1D;
     hipGraphExecMemcpyNodeSetParams1D;
+    hipGraphAddEventRecordNode;
+    hipGraphEventRecordNodeGetEvent;
+    hipGraphEventRecordNodeSetEvent;
+    hipGraphExecEventRecordNodeSetEvent;
+    hipGraphAddEventWaitNode;
+    hipGraphEventWaitNodeGetEvent;
+    hipGraphEventWaitNodeSetEvent;
+    hipGraphExecEventWaitNodeSetEvent;
 local:
     *;
 } hip_4.4;

projects/clr/hipamd/src/hip_internal.hpp

@@ -26,7 +26,6 @@
 #include "trace_helper.h"
 #include "utils/debug.hpp"
 #include "hip_formatting.hpp"
-
 #include "hip_graph_capture.hpp"
 
 #include <unordered_set>
@@ -145,10 +144,8 @@ namespace hc {
 class accelerator;
 class accelerator_view;
 };
-
 namespace hip {
   class Device;
-
   class Stream {
   public:
     enum Priority : int { High = -1, Normal = 0, Low = 1 };
@@ -353,6 +350,7 @@ namespace hip {
   extern bool isValid(hipStream_t& stream);
 };
 
+extern void WaitThenDecrementSignal(hipStream_t stream, hipError_t status, void* user_data);
 struct ihipExec_t {
   dim3 gridDim_;
   dim3 blockDim_;

projects/clr/hipamd/src/hip_module.cpp

@@ -391,15 +391,20 @@ hipError_t ihipModuleLaunchKernel(hipFunction_t f, uint32_t globalWorkSizeX,
   hip::Event* eStart = reinterpret_cast<hip::Event*>(startEvent);
   hip::Event* eStop = reinterpret_cast<hip::Event*>(stopEvent);
   command->enqueue();
-
   if (startEvent != nullptr) {
-    eStart->addMarker(queue, command, false);
+    status = eStart->addMarker(hStream, command, false);
     command->retain();
   }
+  if (status != hipSuccess) {
+    return status;
+  }
   if (stopEvent != nullptr) {
-    eStop->addMarker(queue, command, true);
+    status = eStop->addMarker(hStream, command, true);
     command->retain();
   }
+  if (status != hipSuccess) {
+    return status;
+  }
   command->release();
 
   return hipSuccess;

projects/clr/hipamd/src/hip_stream.cpp

@@ -415,18 +415,13 @@ hipError_t hipStreamDestroy(hipStream_t stream) {
   HIP_RETURN(hipSuccess);
 }
 
-struct CallbackData {
-    int previous_read_index;
-    hip::ihipIpcEventShmem_t *shmem;
-};
-
-void WaitThenDecrementSignal(hipStream_t stream, hipError_t status, void* user_data){
-    CallbackData *data = (CallbackData *)user_data;
-    int offset = data->previous_read_index % IPC_SIGNALS_PER_EVENT;
-    while (data->shmem->read_index < data->previous_read_index+IPC_SIGNALS_PER_EVENT
-                    && data->shmem->signal[offset] != 0) {
-    }
-    delete data;
+void WaitThenDecrementSignal(hipStream_t stream, hipError_t status, void* user_data) {
+  CallbackData* data = (CallbackData*)user_data;
+  int offset = data->previous_read_index % IPC_SIGNALS_PER_EVENT;
+  while (data->shmem->read_index < data->previous_read_index + IPC_SIGNALS_PER_EVENT &&
+         data->shmem->signal[offset] != 0) {
+  }
+  delete data;
 }
 
 // ================================================================================================
@@ -443,24 +438,8 @@ hipError_t hipStreamWaitEvent(hipStream_t stream, hipEvent_t event, unsigned int
     HIP_RETURN(hipErrorContextIsDestroyed);
   }
 
-  amd::HostQueue* queue = hip::getQueue(stream);
-
   hip::Event* e = reinterpret_cast<hip::Event*>(event);
-  if (e->flags & hipEventInterprocess) {
-    amd::Command* command = new amd::Marker(*queue, false);
-    auto t{new CallbackData{e->ipc_evt_.ipc_shmem_->read_index, e->ipc_evt_.ipc_shmem_}};
-    StreamCallback* cbo = new StreamCallback(stream,
-                    reinterpret_cast<hipStreamCallback_t> (WaitThenDecrementSignal), t, command);
-    if (!command->setCallback(CL_COMPLETE, ihipStreamCallback,cbo)) {
-      command->release();
-      return hipErrorInvalidHandle;
-    }
-    command->enqueue();
-    command->awaitCompletion();
-    HIP_RETURN(hipSuccess);
-  } else {
-    HIP_RETURN(e->streamWait(queue, flags));
-  }
+  HIP_RETURN(e->streamWait(stream, flags));
 }
 
 // ================================================================================================