[SYCL] Fix error handling in non-blocking pipe operations (#13166)

When a non-blocking pipe operation fails,
CL_EXEC_STATUS_ERROR_FOR_EVENTS_IN_WAIT_LIST is expected. The runtime
needs to handle that case instead of throwing the exception.

Author: sergey-semenov

sycl/include/sycl/ext/intel/experimental/pipes.hpp

@@ -39,14 +39,15 @@ namespace ext {
 namespace intel {
 namespace experimental {
 
-// A helper templateless base class to get the host_pipe name.
+// A helper templateless base class.
 class pipe_base {
 
 protected:
   pipe_base();
   ~pipe_base();
 
   __SYCL_EXPORT static std::string get_pipe_name(const void *HostPipePtr);
+  __SYCL_EXPORT static bool wait_non_blocking(const event &E);
 };
 
 template <class _name, class _dataT, int32_t _min_capacity = 0,
@@ -95,15 +96,13 @@ class pipe : public pipe_base {
       CGH.ext_intel_read_host_pipe(PipeName, DataPtr,
                                    sizeof(_dataT) /* non-blocking */);
     });
-    E.wait();
-    if (E.get_info<sycl::info::event::command_execution_status>() ==
-        sycl::info::event_command_status::complete) {
-      Success = true;
-      return *(_dataT *)DataPtr;
-    } else {
-      Success = false;
-      return _dataT();
-    }
+    // In OpenCL 1.0 waiting for a failed event does not return an error, so we
+    // need to check the execution status here as well.
+    Success = wait_non_blocking(E) &&
+              E.get_info<sycl::info::event::command_execution_status>() ==
+                  sycl::info::event_command_status::complete;
+    ;
+    return Success ? *(_dataT *)DataPtr : _dataT();
   }
 
   static void write(queue &Q, const _dataT &Data, bool &Success,
@@ -126,9 +125,11 @@ class pipe : public pipe_base {
       CGH.ext_intel_write_host_pipe(PipeName, DataPtr,
                                     sizeof(_dataT) /* non-blocking */);
     });
-    E.wait();
-    Success = E.get_info<sycl::info::event::command_execution_status>() ==
-              sycl::info::event_command_status::complete;
+    // In OpenCL 1.0 waiting for a failed event does not return an error, so we
+    // need to check the execution status here as well.
+    Success = wait_non_blocking(E) &&
+              E.get_info<sycl::info::event::command_execution_status>() ==
+                  sycl::info::event_command_status::complete;
   }
 
   // Reading from pipe is lowered to SPIR-V instruction OpReadPipe via SPIR-V

sycl/source/detail/event_impl.cpp

@@ -58,10 +58,23 @@ event_impl::~event_impl() {
     getPlugin()->call<PiApiKind::piEventRelease>(MEvent);
 }
 
-void event_impl::waitInternal() {
+void event_impl::waitInternal(bool *Success) {
   if (!MHostEvent && MEvent) {
     // Wait for the native event
-    getPlugin()->call<PiApiKind::piEventsWait>(1, &MEvent);
+    sycl::detail::pi::PiResult Err =
+        getPlugin()->call_nocheck<PiApiKind::piEventsWait>(1, &MEvent);
+    // TODO drop the PI_ERROR_UKNOWN from here once the UR counterpart to
+    // PI_ERROR_EXEC_STATUS_ERROR_FOR_EVENTS_IN_WAIT_LIST is added:
+    // https://github.com/oneapi-src/unified-runtime/issues/1459
+    if (Success != nullptr &&
+        (Err == PI_ERROR_UNKNOWN ||
+         Err == PI_ERROR_EXEC_STATUS_ERROR_FOR_EVENTS_IN_WAIT_LIST))
+      *Success = false;
+    else {
+      getPlugin()->checkPiResult(Err);
+      if (Success != nullptr)
+        *Success = true;
+    }
   } else if (MState == HES_Discarded) {
     // Waiting for the discarded event is invalid
     throw sycl::exception(
@@ -229,7 +242,8 @@ void event_impl::instrumentationEpilog(void *TelemetryEvent,
 #endif
 }
 
-void event_impl::wait(std::shared_ptr<sycl::detail::event_impl> Self) {
+void event_impl::wait(std::shared_ptr<sycl::detail::event_impl> Self,
+                      bool *Success) {
   if (MState == HES_Discarded)
     throw sycl::exception(make_error_code(errc::invalid),
                           "wait method cannot be used for a discarded event.");
@@ -251,9 +265,9 @@ void event_impl::wait(std::shared_ptr<sycl::detail::event_impl> Self) {
   if (MEvent)
     // presence of MEvent means the command has been enqueued, so no need to
     // go via the slow path event waiting in the scheduler
-    waitInternal();
+    waitInternal(Success);
   else if (MCommand)
-    detail::Scheduler::getInstance().waitForEvent(Self);
+    detail::Scheduler::getInstance().waitForEvent(Self, Success);
 
 #ifdef XPTI_ENABLE_INSTRUMENTATION
   instrumentationEpilog(TelemetryEvent, Name, StreamID, IId);

sycl/source/detail/event_impl.hpp

@@ -75,7 +75,12 @@ class event_impl {
   /// Self is needed in order to pass shared_ptr to Scheduler.
   ///
   /// \param Self is a pointer to this event.
-  void wait(std::shared_ptr<sycl::detail::event_impl> Self);
+  /// \param Success is an optional parameter that, when set to a non-null
+  ///        pointer, indicates that failure is a valid outcome for this wait
+  ///        (e.g., in case of a non-blocking read from a pipe), and the value
+  ///        it's pointing to is then set according to the outcome.
+  void wait(std::shared_ptr<sycl::detail::event_impl> Self,
+            bool *Success = nullptr);
 
   /// Waits for the event.
   ///
@@ -108,7 +113,11 @@ class event_impl {
   ~event_impl();
 
   /// Waits for the event with respect to device type.
-  void waitInternal();
+  /// \param Success is an optional parameter that, when set to a non-null
+  ///        pointer, indicates that failure is a valid outcome for this wait
+  ///        (e.g., in case of a non-blocking read from a pipe), and the value
+  ///        it's pointing to is then set according to the outcome.
+  void waitInternal(bool *Success = nullptr);
 
   /// Marks this event as completed.
   void setComplete();

sycl/source/detail/pipes.cpp

@@ -6,6 +6,7 @@
 //
 //===----------------------------------------------------------------------===//
 
+#include <detail/event_impl.hpp>
 #include <detail/host_pipe_map_entry.hpp>
 #include <detail/program_manager/program_manager.hpp>
 #include <sycl/ext/intel/experimental/pipes.hpp>
@@ -22,6 +23,14 @@ __SYCL_EXPORT std::string pipe_base::get_pipe_name(const void *HostPipePtr) {
       ->MUniqueId;
 }
 
+__SYCL_EXPORT bool pipe_base::wait_non_blocking(const event &E) {
+  bool Success = false;
+  std::shared_ptr<sycl::detail::event_impl> EImpl =
+      sycl::detail::getSyclObjImpl(E);
+  EImpl->wait(EImpl, &Success);
+  return Success;
+}
+
 } // namespace experimental
 } // namespace intel
 } // namespace ext

sycl/source/detail/scheduler/graph_processor.cpp

@@ -24,7 +24,7 @@ static Command *getCommand(const EventImplPtr &Event) {
 void Scheduler::GraphProcessor::waitForEvent(const EventImplPtr &Event,
                                              ReadLockT &GraphReadLock,
                                              std::vector<Command *> &ToCleanUp,
-                                             bool LockTheLock) {
+                                             bool LockTheLock, bool *Success) {
   Command *Cmd = getCommand(Event);
   // Command can be nullptr if user creates sycl::event explicitly or the
   // event has been waited on by another thread
@@ -41,7 +41,7 @@ void Scheduler::GraphProcessor::waitForEvent(const EventImplPtr &Event,
   assert(Cmd->getEvent() == Event);
 
   GraphReadLock.unlock();
-  Event->waitInternal();
+  Event->waitInternal(Success);
 
   if (LockTheLock)
     GraphReadLock.lock();

sycl/source/detail/scheduler/scheduler.cpp

@@ -266,13 +266,13 @@ bool Scheduler::isInstanceAlive() {
   return GlobalHandler::instance().isSchedulerAlive();
 }
 
-void Scheduler::waitForEvent(const EventImplPtr &Event) {
+void Scheduler::waitForEvent(const EventImplPtr &Event, bool *Success) {
   ReadLockT Lock = acquireReadLock();
   // It's fine to leave the lock unlocked upon return from waitForEvent as
   // there's no more actions to do here with graph
   std::vector<Command *> ToCleanUp;
   GraphProcessor::waitForEvent(std::move(Event), Lock, ToCleanUp,
-                               /*LockTheLock=*/false);
+                               /*LockTheLock=*/false, Success);
   cleanupCommands(ToCleanUp);
 }
 

sycl/source/detail/scheduler/scheduler.hpp

@@ -396,7 +396,12 @@ class Scheduler {
   /// corresponding function of device API.
   ///
   /// \param Event is a pointer to event to wait on.
-  void waitForEvent(const EventImplPtr &Event);
+  /// \param Success is an optional parameter that, when set to a non-null
+  ///        pointer, indicates that failure is a valid outcome for this wait
+  ///        (e.g., in case of a non-blocking read from a pipe), and the value
+  ///        it's pointing to is then set according to the outcome.
+
+  void waitForEvent(const EventImplPtr &Event, bool *Success = nullptr);
 
   /// Removes buffer from the graph.
   ///
@@ -886,13 +891,17 @@ class Scheduler {
     /// \param GraphReadLock read-lock which is already acquired for reading
     /// \param ToCleanUp container for commands that can be cleaned up.
     /// \param LockTheLock selects if graph lock should be locked upon return
+    /// \param Success is an optional parameter that, when set to a non-null
+    ///        pointer, indicates that failure is a valid outcome for this wait
+    ///        (e.g., in case of a non-blocking read from a pipe), and the value
+    ///        it's pointing to is then set according to the outcome.
     ///
     /// The function may unlock and lock GraphReadLock as needed. Upon return
     /// the lock is left in locked state if and only if LockTheLock is true.
     static void waitForEvent(const EventImplPtr &Event,
                              ReadLockT &GraphReadLock,
                              std::vector<Command *> &ToCleanUp,
-                             bool LockTheLock = true);
+                             bool LockTheLock = true, bool *Success = nullptr);
 
     /// Enqueues the command and all its dependencies.
     ///

sycl/test/abi/sycl_symbols_linux.dump

@@ -3656,6 +3656,7 @@ _ZN4sycl3_V122accelerator_selector_vERKNS0_6deviceE
 _ZN4sycl3_V13ext5intel12experimental15online_compilerILNS3_15source_languageE0EE7compileIJSt6vectorINSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEESaISE_EEEEES8_IhSaIhEERKSE_DpRKT_
 _ZN4sycl3_V13ext5intel12experimental15online_compilerILNS3_15source_languageE1EE7compileIJSt6vectorINSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEESaISE_EEEEES8_IhSaIhEERKSE_DpRKT_
 _ZN4sycl3_V13ext5intel12experimental9pipe_base13get_pipe_nameB5cxx11EPKv
+_ZN4sycl3_V13ext5intel12experimental9pipe_base17wait_non_blockingERKNS0_5eventE
 _ZN4sycl3_V13ext6oneapi10level_zero10make_eventERKNS0_7contextEmb
 _ZN4sycl3_V13ext6oneapi10level_zero10make_queueERKNS0_7contextERKNS0_6deviceEmbbRKNS0_13property_listE
 _ZN4sycl3_V13ext6oneapi10level_zero11make_deviceERKNS0_8platformEm

sycl/test/abi/sycl_symbols_windows.dump

@@ -5162,6 +5162,7 @@
 ?wait_and_throw@event@_V1@sycl@@SAXAEBV?$vector@Vevent@_V1@sycl@@V?$allocator@Vevent@_V1@sycl@@@std@@@std@@@Z
 ?wait_and_throw@queue@_V1@sycl@@QEAAXAEBUcode_location@detail@23@@Z
 ?wait_and_throw_proxy@queue@_V1@sycl@@QEAAXAEBUcode_location@detail@23@@Z
+?wait_non_blocking@pipe_base@experimental@intel@ext@_V1@sycl@@KA_NAEBVevent@56@@Z
 ?wait_proxy@queue@_V1@sycl@@QEAAXAEBUcode_location@detail@23@@Z
 ?what@exception@_V1@sycl@@UEBAPEBDXZ
 DllMain

sycl/unittests/pipes/host_pipe_registration.cpp

@@ -81,12 +81,14 @@ static int PipeWriteVal = 0;
 pi_result redefinedEnqueueReadHostPipe(pi_queue, pi_program, const char *,
                                        pi_bool, void *ptr, size_t, pi_uint32,
                                        const pi_event *, pi_event *event) {
+  *event = createDummyHandle<pi_event>();
   *(((int *)ptr)) = PipeReadVal;
   return PI_SUCCESS;
 }
 pi_result redefinedEnqueueWriteHostPipe(pi_queue, pi_program, const char *,
                                         pi_bool, void *ptr, size_t, pi_uint32,
                                         const pi_event *, pi_event *event) {
+  *event = createDummyHandle<pi_event>();
   PipeWriteVal = 9;
   return PI_SUCCESS;
 }
@@ -142,14 +144,15 @@ class PipeTest : public ::testing::Test {
   queue q;
 };
 
+static sycl::unittest::PiImage Img = generateDefaultImage();
+static sycl::unittest::PiImageArray<1> ImgArray{&Img};
+
 TEST_F(PipeTest, Basic) {
   // Fake extension
   Mock.redefineAfter<sycl::detail::PiApiKind::piDeviceGetInfo>(
       after_piDeviceGetInfo);
 
   // Device registration
-  static sycl::unittest::PiImage Img = generateDefaultImage();
-  static sycl::unittest::PiImageArray<1> ImgArray{&Img};
 
   // Testing read
   int HostPipeReadData;
@@ -161,3 +164,45 @@ TEST_F(PipeTest, Basic) {
   Pipe::write(q, HostPipeWriteData);
   EXPECT_EQ(PipeWriteVal, 9);
 }
+
+bool EventsWaitFails = true;
+pi_result redefinedEventsWait(pi_uint32 num_events,
+                              const pi_event *event_list) {
+  return EventsWaitFails ? PI_ERROR_UNKNOWN : PI_SUCCESS;
+}
+
+pi_result after_piEventGetInfo(pi_event event, pi_event_info param_name,
+                               size_t param_value_size, void *param_value,
+                               size_t *param_value_size_ret) {
+  if (param_name == PI_EVENT_INFO_COMMAND_EXECUTION_STATUS) {
+    if (param_value)
+      *static_cast<pi_event_status *>(param_value) = pi_event_status(-1);
+    if (param_value_size_ret)
+      *param_value_size_ret = sizeof(pi_event_status);
+  }
+  return PI_SUCCESS;
+}
+
+TEST_F(PipeTest, NonBlockingOperationFail) {
+  Mock.redefineAfter<sycl::detail::PiApiKind::piDeviceGetInfo>(
+      after_piDeviceGetInfo);
+  Mock.redefine<sycl::detail::PiApiKind::piEventsWait>(redefinedEventsWait);
+
+  bool Success = false;
+  Pipe::read(q, Success);
+  ASSERT_FALSE(Success);
+
+  Pipe::write(q, 0, Success);
+  ASSERT_FALSE(Success);
+
+  // Test the OpenCL 1.0 case: no error code after waiting.
+  EventsWaitFails = false;
+  Mock.redefineAfter<sycl::detail::PiApiKind::piEventGetInfo>(
+      after_piEventGetInfo);
+
+  Pipe::read(q, Success);
+  ASSERT_FALSE(Success);
+
+  Pipe::write(q, 0, Success);
+  ASSERT_FALSE(Success);
+}