Reduce overhead of CUDA actors

CMakeLists.txt

@@ -59,6 +59,7 @@ option(CAF_ENABLE_PROTOBUF_EXAMPLES "Build examples with Google Protobuf" OFF)
 option(CAF_ENABLE_QT6_EXAMPLES "Build examples with the Qt6 framework" OFF)
 option(CAF_ENABLE_ROBOT_TESTS "Add the Robot tests to CTest " OFF)
 option(CAF_ENABLE_RUNTIME_CHECKS "Build CAF with extra runtime assertions" OFF)
+option(CAF_ENABLE_SC26_BENCHMARKS "Build experimental SC26 CUDA benchmarks" OFF)
 option(CAF_USE_STD_FORMAT "Enable std::format support" OFF)
 option(CAF_ENABLE_CUDA "Build caf with cuda support" ON)
 
@@ -110,6 +111,10 @@ if(MSVC AND CAF_SANITIZERS)
   message(FATAL_ERROR "Sanitizer builds are currently not supported on MSVC")
 endif()
 
+if(CAF_ENABLE_SC26_BENCHMARKS AND NOT CAF_ENABLE_CUDA)
+  message(FATAL_ERROR "SC26 benchmarks require CAF_ENABLE_CUDA=ON")
+endif()
+
 # -- doxygen setup -------------------------------------------------------------
 
 configure_file("${CMAKE_CURRENT_SOURCE_DIR}/cmake/Doxyfile.in"
@@ -388,6 +393,10 @@ if(CAF_ENABLE_CUDA)
 	add_subdirectory(libcaf_cuda)
 endif()
 
+if(CAF_ENABLE_SC26_BENCHMARKS)
+  add_subdirectory(libcaf_cuda/sc26)
+endif()
+
 
 
 # -- optionally add the Robot tests to CTest -----------------------------------

libcaf_cuda/CMakeLists.txt

@@ -60,13 +60,15 @@ caf_add_component(
   cuda
   DEPENDENCIES
     PUBLIC
-    CAF::core
+      CAF::core
       $<$<CXX_COMPILER_ID:MSVC>:ws2_32>
+      CUDA::cuda_driver
+      CUDA::cublas
+      CUDA::cusparse
     PRIVATE
       CAF::internal
       #CAF::CORE
       CUDA::nvrtc
-      CUDA::cuda_driver
     HEADERS
     ${CAF_CUDA_HEADERS}
   SOURCES

libcaf_cuda/caf/cuda/command_runner.hpp

@@ -11,6 +11,11 @@
 #include "caf/cuda/control-layer/launch_response_token.hpp"
 #include "caf/cuda/control-layer/memory_response_token.hpp"
 #include "caf/cuda/event.hpp"
+#include "caf/cuda/detail/context_guard.hpp"
+
+#include <cstdio>
+#include <exception>
+#include <memory>
 
 namespace caf::cuda {
 
@@ -20,6 +25,9 @@ namespace caf::cuda {
 // gpu actors if they wish 
 // Manages synchronous and asynchronous command execution with overloads
 // for actor_id, shared_memory, and device_number.
+// An actor_id is a submission serialization domain: submissions and stream
+// release for the same ID must not overlap. Different IDs may submit in
+// parallel.
 // ===========================================================================
 template <class... Ts>
 class command_runner {
@@ -37,11 +45,9 @@ class command_runner {
            int actor_id,
            Us&&... xs) 
   {
-      auto cmd = caf::make_counted<command_t>(std::move(program),
-                                              std::move(dims),
-                                              actor_id,
-                                              std::forward<Us>(xs)...);
-      return cmd->enqueue();
+    auto result = submit_impl(program, dims, actor_id, 0, -1,
+                              std::forward<Us>(xs)...);
+    return result.device->collect_output_buffers(result.memory);
   }
 
   // -------------------------------
@@ -54,12 +60,9 @@ class command_runner {
            int shared_memory,
            Us&&... xs) 
   {
-      auto cmd = caf::make_counted<command_t>(std::move(program),
-                                              std::move(dims),
-                                              actor_id,
-                                              shared_memory,
-                                              std::forward<Us>(xs)...);
-      return cmd->enqueue();
+    auto result = submit_impl(program, dims, actor_id, shared_memory, -1,
+                              std::forward<Us>(xs)...);
+    return result.device->collect_output_buffers(result.memory);
   }
 
   // -------------------------------
@@ -73,13 +76,9 @@ class command_runner {
            int device_number,
            Us&&... xs) 
   {
-      auto cmd = caf::make_counted<command_t>(std::move(program),
-                                              std::move(dims),
-                                              actor_id,
-                                              shared_memory,
-                                              device_number,
-                                              std::forward<Us>(xs)...);
-      return cmd->enqueue();
+    auto result = submit_impl(program, dims, actor_id, shared_memory,
+                              device_number, std::forward<Us>(xs)...);
+    return result.device->collect_output_buffers(result.memory);
   }
 
 
@@ -118,11 +117,9 @@ class command_runner {
                  int actor_id,
                  Us&&... xs) 
   {
-      auto cmd = caf::make_counted<base_command_t>(std::move(program),
-                                                   std::move(dims),
-                                                   actor_id,
-                                                   std::forward<Us>(xs)...);
-      return cmd->base_enqueue();
+    auto result = submit_impl(program, dims, actor_id, 0, -1,
+                              std::forward<Us>(xs)...);
+    return std::move(result.memory);
   }
 
   // -------------------------------
@@ -135,12 +132,9 @@ class command_runner {
                  int shared_memory,
                  Us&&... xs) 
   {
-      auto cmd = caf::make_counted<base_command_t>(std::move(program),
-                                                   std::move(dims),
-                                                   actor_id,
-                                                   shared_memory,
-                                                   std::forward<Us>(xs)...);
-      return cmd->base_enqueue();
+    auto result = submit_impl(program, dims, actor_id, shared_memory, -1,
+                              std::forward<Us>(xs)...);
+    return std::move(result.memory);
   }
 
   // -------------------------------
@@ -154,13 +148,9 @@ class command_runner {
                  int device_number,
                  Us&&... xs) 
   {
-      auto cmd = caf::make_counted<base_command_t>(std::move(program),
-                                                   std::move(dims),
-                                                   actor_id,
-                                                   shared_memory,
-                                                   device_number,
-                                                   std::forward<Us>(xs)...);
-      return cmd->base_enqueue();
+    auto result = submit_impl(program, dims, actor_id, shared_memory,
+                              device_number, std::forward<Us>(xs)...);
+    return std::move(result.memory);
   }
 
 
@@ -212,9 +202,9 @@ class command_runner {
                          int stream_id,
                          Us&&... args)
     {
-        auto cmd = caf::make_counted<bulk_memory_command<std::decay_t<Us>...>>(
-          device_number, stream_id, std::forward<Us>(args)...);
-        return cmd->enqueue();
+        bulk_memory_command<std::decay_t<Us>...> cmd{
+          device_number, stream_id, std::forward<Us>(args)...};
+        return cmd.enqueue();
     }
 
    // -------------------------------------------------------------------------
@@ -257,25 +247,28 @@ class command_runner {
         cmd.enqueue(dst, count);
     }
 
-    // Asynchronous copy back (default)
+    // Asynchronous copy back (default). CUDA host callbacks must not call CUDA
+    // APIs or release the final reference to an object whose destructor does.
     template <typename T, typename F>
     void copy_to_host_async(mem_ptr<T> ptr, F callback) {
-        auto cmd = caf::make_counted<copy_back_command<T>>(std::move(ptr));
-        cmd->run_async(std::move(callback));
+        copy_back_command<T> cmd{std::move(ptr)};
+        cmd.run_async(std::move(callback));
     }
 
-    // Asynchronous copy back with explicit stream/actor ID
+    // Asynchronous copy back with explicit stream/actor ID. CUDA host callbacks
+    // must not call CUDA APIs or release the final CUDA-owning reference.
     template <typename T, typename F>
     void copy_to_host_async(mem_ptr<T> ptr, int stream_id, F callback) {
-        auto cmd = caf::make_counted<copy_back_command<T>>(std::move(ptr), stream_id);
-        cmd->run_async(std::move(callback));
+        copy_back_command<T> cmd{std::move(ptr), stream_id};
+        cmd.run_async(std::move(callback));
     }
 
-    // Asynchronous copy back to user-provided buffer
+    // Asynchronous copy back to user-provided buffer. CUDA host callbacks must
+    // not call CUDA APIs or release the final CUDA-owning reference.
     template <typename T, typename F>
     void copy_to_host_async(mem_ptr<T> ptr, T* dst, size_t count, F callback) {
-        auto cmd = caf::make_counted<copy_back_command<T>>(std::move(ptr));
-        cmd->run_async(dst, count, std::move(callback));
+        copy_back_command<T> cmd{std::move(ptr)};
+        cmd.run_async(dst, count, std::move(callback));
     }
 
     // Enqueue an asynchronous free operation on the given stream
@@ -289,31 +282,47 @@ class command_runner {
   // -------------------------------
   // Destroy streams for a given actor ID
   // -------------------------------
+  // The caller must first ensure that no submission for this ID is in flight.
   void release_stream_for_actor(int actor_id) {
       auto plat = platform::create();
       plat->release_streams_for_actor(actor_id);
   }
 
   // -------------------------------
-  // Register a callback on the actor's stream
+  // Register a callback on the actor's stream. The callback must not call CUDA
+  // APIs or release the final reference to an object whose destructor does.
   // -------------------------------
   template <typename F>
   void add_callback(int stream_id, int device_number, F callback) {
       auto plat = platform::create();
       auto dev = plat->schedule(stream_id, device_number);
       auto stream = dev->get_stream_for_actor(stream_id);
-      auto* f_ptr = new F(std::move(callback));
-      auto res = cuLaunchHostFunc(stream, [](void* data) {
-          auto* f = static_cast<F*>(data);
-          (*f)();
-          delete f;
-      }, f_ptr);
-      if (res != CUDA_SUCCESS) { delete f_ptr; check(res, "cuLaunchHostFunc"); }
+      auto state = std::unique_ptr<F>{new F(std::move(callback))};
+      detail::context_guard context_scope{dev->getContext()};
+      auto res = cuLaunchHostFunc(stream, [](void* data) noexcept {
+          auto fn = std::unique_ptr<F>{static_cast<F*>(data)};
+          try {
+            (*fn)();
+          } catch (const std::exception& error) {
+            std::fprintf(stderr, "CUDA host callback failed: %s\n",
+                         error.what());
+          } catch (...) {
+            std::fprintf(stderr,
+                         "CUDA host callback failed: unknown error\n");
+          }
+      }, state.get());
+      if (res != CUDA_SUCCESS)
+        throw detail::cuda_error(res, "cuLaunchHostFunc");
+      state.release();
   }
 
   // -------------------------------
   // Resets the CUDA context for a given device number.
-  // This will force the device to flush its stream pool and create a new context.
+  // This will force the device to flush its context-bound resources and create
+  // a new context. The caller must first quiesce submissions, callbacks,
+  // program/memory/event lifecycle, and raw handle use on this device.
+  // Existing non-scalar memory, events, and raw handles become stale; retained
+  // programs reload lazily on their next launch.
   // -------------------------------
   void reset_context(int device_number) {
       auto plat = platform::create();
@@ -377,6 +386,34 @@ class command_runner {
       return dev->get_stream_for_actor(stream_number);
   }
 
+private:
+  using result_type = std::tuple<mem_ptr<raw_t<Ts>>...>;
+
+  struct submission_result {
+    device_ptr device;
+    result_type memory;
+  };
+
+  template <class... Us>
+  submission_result submit_impl(const program_ptr& program,
+                                const nd_range& dims,
+                                int actor_id,
+                                int shared_memory,
+                                int device_number,
+                                Us&&... xs) {
+    static_assert(sizeof...(Us) == sizeof...(Ts), "Argument count mismatch");
+
+    // Match base_command's conversion behavior while keeping only one tuple.
+    std::tuple<Ts...> arguments{
+      std::make_tuple(std::forward<Us>(xs)...)};
+
+    auto dev = program->device_for_launch(actor_id, device_number);
+    auto kernel = program->get_kernel(dev->getId());
+    auto memory = dev->launch_kernel_mem_ref_impl(
+      kernel, dims, arguments, actor_id, shared_memory);
+    return {std::move(dev), std::move(memory)};
+  }
+
 };
 
 } // namespace caf::cuda

libcaf_cuda/caf/cuda/control-layer/request_token.hpp

@@ -20,13 +20,6 @@ class CAF_CUDA_EXPORT request_token : public token {
 public:
   request_token(int dependency = INDEPENDENT)
     : token(dependency) {}
-
-  // Required for CAF message passing
-  request_token() = default;
-
-
-private:
-
 };
 
 using request_token_ptr = caf::intrusive_ptr<request_token>;

libcaf_cuda/caf/cuda/detail/context_guard.hpp

@@ -0,0 +1,57 @@
+#pragma once
+
+#include <cuda.h>
+
+#include <cstdio>
+#include <stdexcept>
+#include <string>
+
+namespace caf::cuda::detail {
+
+inline std::runtime_error cuda_error(CUresult result, const char* operation) {
+  const char* description = nullptr;
+  cuGetErrorString(result, &description);
+  return std::runtime_error(std::string{operation} + " failed: "
+                            + (description ? description : "unknown CUDA error"));
+}
+
+/// Temporarily makes a CUDA context current and restores the caller's context.
+class context_guard {
+public:
+  explicit context_guard(CUcontext context) {
+    if (!context)
+      throw std::invalid_argument("context_guard requires a valid context");
+    CUcontext current = nullptr;
+    auto result = cuCtxGetCurrent(&current);
+    if (result != CUDA_SUCCESS)
+      throw cuda_error(result, "cuCtxGetCurrent");
+    if (current != context) {
+      result = cuCtxPushCurrent(context);
+      if (result != CUDA_SUCCESS)
+        throw cuda_error(result, "cuCtxPushCurrent");
+      pushed_ = true;
+    }
+  }
+
+  ~context_guard() noexcept {
+    if (pushed_) {
+      CUcontext ignored = nullptr;
+      auto result = cuCtxPopCurrent(&ignored);
+      if (result != CUDA_SUCCESS) {
+        // Destructors cannot propagate a restoration failure.
+        const char* description = nullptr;
+        cuGetErrorString(result, &description);
+        std::fprintf(stderr, "cuCtxPopCurrent failed: %s\n",
+                     description ? description : "unknown CUDA error");
+      }
+    }
+  }
+
+  context_guard(const context_guard&) = delete;
+  context_guard& operator=(const context_guard&) = delete;
+
+private:
+  bool pushed_ = false;
+};
+
+} // namespace caf::cuda::detail