Making merge into GPU actors

.gitignore

@@ -1,6 +1,6 @@
 .make-release-steps.bash
 .push-doxygen-steps.bash
-Makefile
+# Makefile
 bin/*
 build/*
 doxygen-log.txt
@@ -11,3 +11,25 @@ manual/libcaf_core
 manual/libcaf_io
 manual/libcaf_net
 manual/libcaf_openssl
+libcaf_cuda/sc26/Runtime-Overhead/compile.sh
+libcaf_cuda/sc26/Runtime-Overhead/cuda_native
+libcaf_cuda/sc26/Runtime-Overhead/mmul.cubin
+libcaf_cuda/sc26/Sequence-Independent-Tasks/compile.sh
+libcaf_cuda/sc26/Runtime-Overhead/command_runner
+libcaf_cuda/sc26/Fault-Tolerance/compile.sh
+libcaf_cuda/sc26/Fault-Tolerance/main
+libcaf_cuda/sc26/Runtime-Overhead/actor_facade
+libcaf_cuda/sc26/Fault-Tolerance/monte_carlo.cubin
+libcaf_cuda/sc26/Sequence-Independent-Tasks/main_actor_facade
+libcaf_cuda/sc26/Sequence-Independent-Tasks/main_command_runner
+libcaf_cuda/sc26/Sequence-Independent-Tasks/main_cuda_native
+libcaf_cuda/sc26/Sequence-Independent-Tasks/mmul.cubin
+libcaf_cuda/sc26/Fault-Tolerance/run.log
+libcaf_cuda/sc26/Fault-Tolerance/baseline.log
+libcaf_cuda/sc26/benchmarks/__pycache__/
+libcaf_cuda/sc26/scripts/Fault-Tolerance/.venv/
+libcaf_cuda/sc26/scripts/Fault-Tolerance/plots
+libcaf_cuda/sc26/Runtime-Overhead/results/benchmark_results.txt
+libcaf_cuda/sc26/Runtime-Overhead/results/benchmark_results_1.txt
+libcaf_cuda/sc26/Sequence-Independent-Tasks/results/cleaned.txt
+libcaf_cuda/sc26/Sequence-Independent-Tasks/results/benchmark_results.txt

CMakeLists.txt

@@ -60,7 +60,7 @@ 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_USE_STD_FORMAT "Enable std::format support" OFF)
-option(CAF_ENABLE_CUDA "Build caf with cuda support" OFF)
+option(CAF_ENABLE_CUDA "Build caf with cuda support" ON)
 
 # -- CAF options that are on by default ----------------------------------------
 

libcaf_cuda/CMakeLists.txt

@@ -1,10 +1,46 @@
 #project(caf_opencl C CXX)
 include(CMakePrintHelpers)
 
-
-#get the cuda related dependencies 
+# Get the cuda related dependencies first so we have the nvcc path
 find_package(CUDAToolkit REQUIRED)
 
+# Set the GPU architecture. 'native' detects the local GPU capability.
+# You can override this with -DCAF_CUDA_ARCH=sm_80 for example.
+set(CAF_CUDA_ARCH "native" CACHE STRING "Target CUDA architecture (e.g., sm_70, native)")
+
+# Find all .cu files recursively in the caf/ directory relative to this CMakeLists.txt
+file(GLOB_RECURSE CAF_CUDA_KERNELS "${CMAKE_CURRENT_SOURCE_DIR}/caf/*.cu")
+
+foreach(CU_FILE ${CAF_CUDA_KERNELS})
+    # Get the base name of the .cu file (e.g., "some_kernel")
+    get_filename_component(BASENAME ${CU_FILE} NAME_WE)
+    # Get the directory of the .cu file (e.g., "/path/to/libcaf_cuda/caf/cuda")
+    get_filename_component(DIRNAME ${CU_FILE} DIRECTORY)
+
+    # Construct the output .cubin file path (e.g., "/path/to/libcaf_cuda/caf/cuda/some_kernel.cubin")
+    set(CUBIN_FILE "${DIRNAME}/${BASENAME}.cubin")
+
+    # Create a unique target name based on the relative path to avoid collisions
+    file(RELATIVE_PATH REL_PATH "${CMAKE_CURRENT_SOURCE_DIR}/caf" ${CU_FILE})
+    string(MAKE_C_IDENTIFIER "cubin_${REL_PATH}" TARGET_NAME)
+
+    # Add a custom command to compile the .cu file into a .cubin
+    add_custom_command(
+        OUTPUT ${CUBIN_FILE}
+        COMMAND ${CUDAToolkit_NVCC_EXECUTABLE}
+            -cubin
+            -arch=${CAF_CUDA_ARCH}
+            -o ${CUBIN_FILE}
+            ${CU_FILE}
+        DEPENDS ${CU_FILE}
+        COMMENT "Compiling CUDA kernel ${CU_FILE} to ${CUBIN_FILE}"
+        VERBATIM
+    )
+
+    # Add a custom target to ensure the cubin is built.
+    add_custom_target(${TARGET_NAME} ALL DEPENDS ${CUBIN_FILE})
+endforeach()
+
 # get header files; only needed by CMake generators,
 # e.g., for creating proper Xcode projects
 file(GLOB_RECURSE CAF_CUDA_HEADERS "caf/*.hpp")
@@ -14,14 +50,8 @@ file(GLOB_RECURSE CAF_CUDA_HEADERS "caf/*.hpp")
 
 
 # list cpp files excluding platform-dependent files
-set(LIBCAF_CUDA_SRCS
-  src/manager.cpp
-  src/platform.cpp
-  src/program.cpp
-  src/helpers.cpp
-  src/test.cpp
-  src/scheduler.cpp
-  src/streampool.cpp
+file(GLOB_RECURSE LIBCAF_CUDA_SRCS
+    src/*.cpp
 )
 
 # -- add targets ---------------------------------------------------------------

libcaf_cuda/caf/actorBLAS/actorBLAS.hpp

@@ -0,0 +1,4 @@
+#pragma once
+#include "caf/actorBLAS/gemv-actor/gemv-actor.hpp"
+#include "caf/actorBLAS/syrk-actor/syrk-actor.hpp"
+#include "caf/actorBLAS/axpy-actor/axpy-actor.hpp"

libcaf_cuda/caf/actorBLAS/axpy-actor/axpy-actor.hpp

@@ -0,0 +1,130 @@
+#pragma once
+
+#include <caf/actor.hpp>
+#include <caf/actor_cast.hpp>
+#include <caf/actor_config.hpp>
+#include <caf/actor_system.hpp>
+#include <caf/anon_mail.hpp>
+#include <caf/event_based_actor.hpp>
+
+#include "caf/cuda/device.hpp"
+#include "caf/cuda/mem_ref.hpp"
+#include "caf/cuda/command_runner.hpp"
+#include "caf/cuda/platform.hpp"
+#include "caf/cuda/types.hpp"
+
+namespace caf::cuda {
+
+/// AXPY Actor for single-precision vector-vector addition.
+/// Message Signature: (in<float> x, in_out<float> y, int n, float alpha)
+class axpy_actor : public event_based_actor {
+public:
+  static caf::actor spawn(caf::actor_system& sys, int reply_id = 0) {
+    return sys.spawn<axpy_actor>(reply_id);
+  }
+
+  axpy_actor(caf::actor_config& cfg, int reply_id = 0) 
+    : event_based_actor(cfg), reply_id_(reply_id) {
+    actor_id_ = static_cast<int>(this->id());
+  }
+
+  ~axpy_actor() override {
+    command_runner<> runner;
+    runner.release_stream_for_actor(actor_id_);
+  }
+
+  caf::behavior make_behavior() override {
+    return {
+      // Standard host buffer based calls
+      [this](in<float> x, in_out<float> y, int n, float alpha) {
+        enqueue_axpy(-1, actor_id_, x, y, n, alpha, false);
+      },
+      // Routing control overloads
+      [this](int device_num, int stream_id, in<float> x, in_out<float> y, int n, float alpha) {
+        enqueue_axpy(device_num, stream_id, x, y, n, alpha, false);
+      },
+      // mem_ptr based calls (useful for pipelines)
+      [this](mem_ptr<float> x, mem_ptr<float> y, int n, float alpha) {
+        enqueue_axpy(-1, actor_id_, x, y, n, alpha, false);
+      },
+      [this](int device_num, int stream_id, mem_ptr<float> x, mem_ptr<float> y, int n, float alpha) {
+        enqueue_axpy(device_num, stream_id, x, y, n, alpha, false);
+      },
+      // Mem ptr return overloads
+      [this](return_mem_ptr_atom, in<float> x, in_out<float> y, int n, float alpha) {
+        enqueue_axpy(-1, actor_id_, x, y, n, alpha, true);
+      },
+      [this](return_mem_ptr_atom, int device_num, int stream_id, in<float> x, in_out<float> y, int n, float alpha) {
+        enqueue_axpy(device_num, stream_id, x, y, n, alpha, true);
+      },
+      [this](return_mem_ptr_atom, mem_ptr<float> x, mem_ptr<float> y, int n, float alpha) {
+        enqueue_axpy(-1, actor_id_, x, y, n, alpha, true);
+      }, [this](return_mem_ptr_atom,int device_num, int stream_id,mem_ptr<float> x, mem_ptr<float> y, int n, float alpha) {
+        enqueue_axpy(device_num, stream_id, x, y, n, alpha, true);
+      },
+    };
+  }
+
+private:
+  // Overload for Host-wrapped buffers
+  void enqueue_axpy(int device_num, int stream_id, 
+                    in<float> x_arg, in_out<float> y_arg, 
+                    int n, float alpha, bool return_ptrs) {
+    command_runner<in<float>, in_out<float>> runner;
+
+    // Allocate/Transfer memory.
+    auto results = runner.transfer_memory(device_num, stream_id, x_arg, y_arg);
+    execute_and_reply(device_num, stream_id, std::get<0>(results), 
+                      std::get<1>(results), n, alpha, return_ptrs);
+  }
+
+  // Overload for already-existing Device buffers
+  void enqueue_axpy(int device_num, int stream_id, 
+                    mem_ptr<float> x_ptr, mem_ptr<float> y_ptr, 
+                    int n, float alpha, bool return_ptrs) {
+    // Pass through command_runner to ensure proper ref-counting/scheduling
+    command_runner<mem_ptr<float>, mem_ptr<float>> runner;
+    auto results = runner.transfer_memory(device_num, stream_id, x_ptr, y_ptr);
+    execute_and_reply(device_num, stream_id, std::get<0>(results), 
+                      std::get<1>(results), n, alpha, return_ptrs);
+  }
+
+  void execute_and_reply(int device_num, int stream_id, 
+                         mem_ptr<float> x, mem_ptr<float> y, 
+                         int n, float alpha, bool return_ptrs) {
+    auto plat = platform::create();
+    device_ptr dev;
+    if (device_num == -1)
+      dev = plat->schedule(stream_id);
+    else
+      dev = plat->schedule(stream_id, device_num);
+
+    // Perform cuBLAS operation
+    dev->saxpy(stream_id, n, alpha, x, y);
+
+    handle_reply(device_num, stream_id, x, y, return_ptrs);
+  }
+
+  void handle_reply(int, int, mem_ptr<float> x_ptr, mem_ptr<float> y_ptr, bool return_ptrs) {
+    command_runner<mem_ptr<float>> runner;
+    auto sender = actor_cast<actor>(this->current_sender());
+    if (!sender) return;
+
+    auto r_id = reply_id_;
+
+    if (return_ptrs) {
+      caf::anon_mail(r_id, x_ptr, y_ptr).send(sender);
+    } else {
+      runner.copy_to_host_async(y_ptr, [sender, r_id](std::vector<float>&& data) {
+        if (sender) {
+          caf::anon_mail(r_id, 1, std::move(data)).send(sender);
+        }
+      });
+    }
+  }
+
+  int actor_id_;
+  int reply_id_;
+};
+
+} // namespace caf::cuda

libcaf_cuda/caf/actorBLAS/batched-gemm-actor/batched-gemm-actor.hpp

@@ -0,0 +1,150 @@
+#pragma once
+
+#include <caf/actor.hpp>
+#include <caf/actor_cast.hpp>
+#include <caf/actor_config.hpp>
+#include <caf/actor_system.hpp>
+#include <caf/anon_mail.hpp>
+#include <caf/event_based_actor.hpp>
+
+#include "caf/cuda/device.hpp"
+#include "caf/cuda/mem_ref.hpp"
+#include "caf/cuda/command_runner.hpp"
+#include "caf/cuda/platform.hpp"
+#include "caf/cuda/types.hpp"
+
+namespace caf::cuda {
+
+/// GEMM Batched Actor for performing matrix-matrix multiplication on batches of matrices.
+/// Uses Strided Batched cuBLAS calls.
+template <typename T>
+class batched_gemm_actor : public event_based_actor {
+public:
+  static caf::actor spawn(caf::actor_system& sys, int reply_id = 0) {
+    return sys.spawn<batched_gemm_actor<T>>(reply_id);
+  }
+
+  batched_gemm_actor(caf::actor_config& cfg, int reply_id = 0)
+    : event_based_actor(cfg), reply_id_(reply_id) {
+    actor_id_ = static_cast<int>(this->id());
+  }
+
+  ~batched_gemm_actor() override {
+    command_runner<> runner;
+    runner.release_stream_for_actor(actor_id_);
+  }
+
+  caf::behavior make_behavior() override {
+    return {
+      // Standard host buffer based calls (alpha=1.0, beta=0.0)
+      [this](in<T> A, in<T> B, out<T> C, int m, int n, int k, int batchCount) {
+        enqueue_gemm_batched(-1, actor_id_, A, B, C, m, n, k, batchCount, static_cast<T>(1.0), static_cast<T>(0.0), false);
+      },
+      // mem_ptr based calls (Zero-copy Pipelines)
+      [this](mem_ptr<T> A, mem_ptr<T> B, mem_ptr<T> C, int m, int n, int k, int batchCount) {
+        enqueue_gemm_batched(-1, actor_id_, A, B, C, m, n, k, batchCount, static_cast<T>(1.0), static_cast<T>(0.0), false);
+      },
+      // Return mem_ptr atom overloads
+      [this](return_mem_ptr_atom, in<T> A, in<T> B, out<T> C, int m, int n, int k, int batchCount) {
+        enqueue_gemm_batched(-1, actor_id_, A, B, C, m, n, k, batchCount, static_cast<T>(1.0), static_cast<T>(0.0), true);
+      },
+      [this](return_mem_ptr_atom, mem_ptr<T> A, mem_ptr<T> B, mem_ptr<T> C, int m, int n, int k, int batchCount) {
+        enqueue_gemm_batched(-1, actor_id_, A, B, C, m, n, k, batchCount, static_cast<T>(1.0), static_cast<T>(0.0), true);
+      },
+      // Routing control (Explicit device/stream)
+      [this](int device_num, int stream_id, in<T> A, in<T> B, out<T> C, int m, int n, int k, int batchCount) {
+        enqueue_gemm_batched(device_num, stream_id, A, B, C, m, n, k, batchCount, static_cast<T>(1.0), static_cast<T>(0.0), false);
+      },
+      [this](int device_num, int stream_id, mem_ptr<T> A, mem_ptr<T> B, mem_ptr<T> C, int m, int n, int k, int batchCount) {
+        enqueue_gemm_batched(device_num, stream_id, A, B, C, m, n, k, batchCount, static_cast<T>(1.0), static_cast<T>(0.0), false);
+      },
+      [this](return_mem_ptr_atom, int device_num, int stream_id, in<T> A, in<T> B, out<T> C, int m, int n, int k, int batchCount) {
+        enqueue_gemm_batched(device_num, stream_id, A, B, C, m, n, k, batchCount, static_cast<T>(1.0), static_cast<T>(0.0), true);
+      },
+      [this](return_mem_ptr_atom, int device_num, int stream_id, mem_ptr<T> A, mem_ptr<T> B, mem_ptr<T> C, int m, int n, int k, int batchCount) {
+        enqueue_gemm_batched(device_num, stream_id, A, B, C, m, n, k, batchCount, static_cast<T>(1.0), static_cast<T>(0.0), true);
+      },
+      // Full Parameter GEMM (custom alpha/beta)
+      [this](in<T> A, in<T> B, out<T> C, int m, int n, int k, int batchCount, T alpha, T beta) {
+        enqueue_gemm_batched(-1, actor_id_, A, B, C, m, n, k, batchCount, alpha, beta, false);
+      },
+      [this](return_mem_ptr_atom, in<T> A, in<T> B, out<T> C, int m, int n, int k, int batchCount, T alpha, T beta) {
+        enqueue_gemm_batched(-1, actor_id_, A, B, C, m, n, k, batchCount, alpha, beta, true);
+      }
+    };
+  }
+
+private:
+  // Logic for host-wrapped buffers
+  template <typename OutType>
+  void enqueue_gemm_batched(int device_num, int stream_id,
+                            in<T> A_arg, in<T> B_arg, OutType C_arg,
+                            int m, int n, int k, int batchCount, T alpha, T beta, bool return_ptrs) {
+    command_runner<in<T>, in<T>, OutType> runner;
+    auto results = runner.transfer_memory(device_num, stream_id, A_arg, B_arg, C_arg);
+    execute_and_reply(device_num, stream_id, std::get<0>(results),
+                      std::get<1>(results), std::get<2>(results),
+                      m, n, k, batchCount, alpha, beta, return_ptrs);
+  }
+
+  // Logic for existing Device buffers
+  void enqueue_gemm_batched(int device_num, int stream_id,
+                            mem_ptr<T> A_ptr, mem_ptr<T> B_ptr, mem_ptr<T> C_ptr,
+                            int m, int n, int k, int batchCount, T alpha, T beta, bool return_ptrs) {
+    command_runner<mem_ptr<T>, mem_ptr<T>, mem_ptr<T>> runner;
+    auto results = runner.transfer_memory(device_num, stream_id, A_ptr, B_ptr, C_ptr);
+    execute_and_reply(device_num, stream_id, std::get<0>(results),
+                      std::get<1>(results), std::get<2>(results),
+                      m, n, k, batchCount, alpha, beta, return_ptrs);
+  }
+
+  void execute_and_reply(int device_num, int stream_id,
+                         mem_ptr<T> A, mem_ptr<T> B, mem_ptr<T> C,
+                         int m, int n, int k, int batchCount, T alpha, T beta, bool return_ptrs) {
+    auto plat = platform::create();
+    device_ptr dev = (device_num == -1) ? plat->schedule(stream_id) : plat->schedule(stream_id, device_num);
+
+    // Default packed strides
+    long long int strideA = static_cast<long long int>(m) * k;
+    long long int strideB = static_cast<long long int>(k) * n;
+    long long int strideC = static_cast<long long int>(m) * n;
+
+    if constexpr (std::is_same_v<T, float>) {
+      dev->sgemm_strided_batched(stream_id, m, n, k, alpha, A, strideA, B, strideB, beta, C, strideC, batchCount);
+    } else if constexpr (std::is_same_v<T, double>) {
+      dev->dgemm_strided_batched(stream_id, m, n, k, alpha, A, strideA, B, strideB, beta, C, strideC, batchCount);
+    } else {
+      static_assert(std::is_same_v<T, float> || std::is_same_v<T, double>, 
+                    "Unsupported type for GEMM batched actor. Only float and double are supported.");
+    }
+
+    handle_reply(A, B, C, return_ptrs);
+  }
+
+  void handle_reply(mem_ptr<T> A_ptr, mem_ptr<T> B_ptr, mem_ptr<T> C_ptr, bool return_ptrs) {
+    command_runner<mem_ptr<T>> runner;
+    auto sender = actor_cast<actor>(this->current_sender());
+    if (!sender) return;
+
+    auto r_id = reply_id_;
+
+    if (return_ptrs) {
+      // Requirement: Always 1 message. 
+      // We bundle all pointers into a single message.
+      caf::anon_mail(r_id, A_ptr, B_ptr, C_ptr).send(sender);
+    } else {
+      // Requirement: Always 1 message. 
+      // We only copy back the result matrix (index 2) and send it.
+      runner.copy_to_host_async(C_ptr, [sender, r_id](std::vector<T>&& data) {
+        if (sender) {
+          caf::anon_mail(r_id, 2, std::move(data)).send(sender);
+        }
+      });
+    }
+  }
+
+  int actor_id_;
+  int reply_id_;
+};
+
+} // namespace caf::cuda