[SYCL][CUDA] Event callback worker implementation

Introduces a worker thread living in the platform for handling callbacks
for native events. This is needed as host functions enqueued by
cuLaunchHostFunc is strictly prohibited from doing any CUDA operations,
but as a SYCL object, such as the event itself, may end its life in the
callback we must be able to destroy the CUDA object.

Signed-off-by: Steffen Larsen <[email protected]>

Author: Steffen Larsen

sycl/plugins/cuda/CMakeLists.txt

@@ -4,6 +4,7 @@ message(STATUS "Including the PI API CUDA backend.")
  # we only require the CUDA driver API to be used
  # CUDA_CUDA_LIBRARY variable defines the path to libcuda.so, the CUDA Driver API library.
 
+find_package(Threads REQUIRED)
 find_package(CUDA 10.0 REQUIRED)
 
 add_library(cudadrv SHARED IMPORTED)
@@ -33,7 +34,7 @@ target_include_directories(pi_cuda PRIVATE "${sycl_inc_dir}")
 
 target_include_directories(pi_cuda INTERFACE ${CUDA_INCLUDE_DIRS})
 
-target_link_libraries(pi_cuda PUBLIC OpenCL-Headers cudadrv)
+target_link_libraries(pi_cuda PUBLIC OpenCL-Headers cudadrv Threads::Threads)
 
 target_link_libraries(sycl INTERFACE pi_cuda)
 

sycl/plugins/cuda/pi_cuda.cpp

@@ -97,6 +97,31 @@ pi_result cuda_piEventRetain(pi_event event);
 
 } // extern "C"
 
+void worker::execute() {
+  bool Terminate = false;
+  while (!Terminate) {
+    std::unique_lock<std::mutex> lock(workQueueGateMutex_);
+    workQueueGate_.wait(lock);
+    while (!workQueue_.empty()) {
+      work item = workQueue_.front();
+      workQueue_.pop();
+      switch (item.kind_) {
+      case work::kind::complete_event:
+        complete_event(static_cast<pi_event>(item.content_));
+        break;
+      case work::kind::terminate:
+        Terminate = true;
+        break;
+      }
+    }
+  }
+}
+
+void worker::complete_event(pi_event event) {
+  event->set_event_complete();
+  cuda_piEventRelease(event);
+}
+
 _pi_event::_pi_event(pi_command_type type, pi_context context, pi_queue queue)
     : commandType_{type}, refCount_{1}, isCompleted_{false},
       isRecorded_{false},
@@ -174,8 +199,9 @@ pi_result _pi_event::record() {
             cuStream,
             [](void *userData) {
               pi_event event = reinterpret_cast<pi_event>(userData);
-              event->set_event_complete();
-              cuda_piEventRelease(event);
+              pi_platform platform =
+                  event->get_context()->get_device()->platform_;
+              platform->worker_.enqueue_complete_event(event);
             },
             this));
       } catch (...) {

sycl/plugins/cuda/pi_cuda.hpp

@@ -17,15 +17,18 @@
 #include <array>
 #include <atomic>
 #include <cassert>
+#include <condition_variable>
 #include <cstring>
 #include <cuda.h>
+#include <functional>
 #include <limits>
+#include <mutex>
 #include <numeric>
+#include <queue>
 #include <stdint.h>
 #include <string>
+#include <thread>
 #include <vector>
-#include <functional>
-#include <mutex>
 
 extern "C" {
 
@@ -45,8 +48,48 @@ pi_result cuda_piKernelRelease(pi_kernel);
 
 }
 
+class worker {
+public:
+  worker()
+      : workQueue_{}, workQueueGateMutex_{}, workQueueGate_{},
+        workerThread_{&worker::execute, this} {}
+
+  ~worker() {
+    enqueue_work(work{work::kind::terminate, nullptr});
+    workerThread_.join();
+  }
+
+  void enqueue_complete_event(pi_event event) {
+    enqueue_work(work{work::kind::complete_event, event});
+  }
+
+private:
+  struct work {
+    enum kind { complete_event, terminate } kind_;
+    void *content_;
+
+    work(kind k, void *c) : kind_{k}, content_{c} {}
+  };
+
+  void enqueue_work(work w) {
+    std::unique_lock<std::mutex> lock(workQueueGateMutex_);
+    workQueue_.push(w);
+    workQueueGate_.notify_one();
+  }
+
+  void complete_event(pi_event event);
+
+  void execute();
+
+  std::queue<work> workQueue_;
+  std::mutex workQueueGateMutex_;
+  std::condition_variable workQueueGate_;
+  std::thread workerThread_;
+};
+
 struct _pi_platform {
   std::vector<std::unique_ptr<_pi_device>> devices_;
+  worker worker_;
 };
 
 struct _pi_device {

sycl/source/CMakeLists.txt

@@ -27,12 +27,16 @@ function(add_sycl_rt_library LIB_NAME)
       set_target_properties(${LIB_NAME} PROPERTIES LINK_DEPENDS ${linker_script})
   endif()
 
+  find_package(Threads REQUIRED)
+  set_property(TARGET ${LIB_NAME} PROPERTY LINK_WHAT_YOU_USE TRUE)
+
   target_include_directories(
       ${LIB_NAME} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR} "${sycl_inc_dir}")
   target_link_libraries(${LIB_NAME}
       PRIVATE OpenCL::Headers
       PRIVATE ${OpenCL_LIBRARIES}
       PRIVATE ${CMAKE_DL_LIBS}
+      PRIVATE Threads::Threads
   )
 
   add_common_options(${LIB_NAME})