Cleanups for OBJ S3

src/plugins/obj/obj_backend.cpp

@@ -31,16 +31,12 @@ namespace {
 
 template<typename... Args>
 std::unique_ptr<nixlObjEngineImpl>
-createAccelEngine(nixl_b_params_t *custom_params, Args &&...args) {
-    // A missing `type` selects the standard, protocol-compliant engine, which
-    // registers under "s3". Normalize to "s3" rather than looking up "" so
-    // resolution stays deterministic regardless of static-init order.
-    std::string type = getAccelType(custom_params);
-    if (type.empty()) {
-        type = "s3";
-    }
+createAccelEngine(const nixl_b_params_t *custom_params, Args &&...args) {
+    // The registry maps an empty type to the standard engine and reports that
+    // case with its own message, so pass the type through unchanged.
     try {
-        return objAccelEngineRegistry::instance().create(type, std::forward<Args>(args)...);
+        return objAccelEngineRegistry::instance().create(getAccelType(custom_params),
+                                                         std::forward<Args>(args)...);
     }
     catch (const std::exception &e) {
         NIXL_ERROR << "Failed to create accelerated engine: " << e.what();

src/plugins/obj/obj_engine_registry.cpp

@@ -5,9 +5,19 @@
 
 #include "obj_engine_registry.h"
 #include "obj_backend.h"
+#include "engine_utils.h"
 #include "common/nixl_log.h"
 #include <stdexcept>
 
+namespace {
+// An empty `type` means the standard, protocol-compliant engine, registered
+// under default_accel_engine_type.
+std::string
+resolveType(const std::string &type) {
+    return type.empty() ? std::string(default_accel_engine_type) : type;
+}
+} // namespace
+
 objAccelEngineRegistry &
 objAccelEngineRegistry::instance() {
     static objAccelEngineRegistry registry;
@@ -27,13 +37,13 @@ objAccelEngineRegistry::add(const std::string &type, objAccelEngineEntry entry)
 bool
 objAccelEngineRegistry::has(const std::string &type) const {
     const std::lock_guard<std::mutex> lock(mutex_);
-    return entries_.count(type) > 0;
+    return entries_.count(resolveType(type)) > 0;
 }
 
 objAccelEngineEntry
 objAccelEngineRegistry::lookupOrThrow(const std::string &type) const {
     const std::lock_guard<std::mutex> lock(mutex_);
-    auto it = entries_.find(type);
+    auto it = entries_.find(resolveType(type));
     if (it != entries_.end()) {
         return it->second;
     }

src/plugins/obj/s3_accel/client.cpp

@@ -10,6 +10,7 @@
 #include <utility>
 
 #include "object/engine_utils.h"
+#include "object/rdma/rdma.h"
 #include "common/nixl_log.h"
 
 awsS3AccelClient::awsS3AccelClient(nixl_b_params_t *custom_params,

src/plugins/obj/s3_accel/client.h

@@ -10,9 +10,15 @@
 #include <string_view>
 
 #include "s3/client.h"
-#include "object/rdma/rdma.h"
 #include "nixl_types.h"
 
+// Declarations only: the members below are a pointer and a shared_ptr, so
+// including rdma.h here would pull cuObject into every including TU for nothing.
+namespace nixl_obj_rdma {
+class SharedCuObjClient;
+class S3RdmaControlPlane;
+} // namespace nixl_obj_rdma
+
 /**
  * S3 Accelerated Object Client - the generic, protocol-compliant S3-over-RDMA
  * client, and the base for vendor-specific accelerated clients.

src/plugins/obj/s3_accel/engine_impl.cpp

@@ -9,6 +9,8 @@
 #include <memory>
 #include <string>
 
+#include "client.h"
+#include "object/engine_utils.h"
 #include "object/rdma/rdma.h"
 #include "obj_engine_registry.h"
 #include "common/nixl_log.h"
@@ -40,10 +42,11 @@ class accelObjMetadata : public nixlBackendMD {
     bool rdmaRegistered = false;
 };
 
-// Register the standard-S3 engine under "s3"; obj_backend normalizes a missing
-// `type` to "s3", so `accelerated=true` with no type resolves here too.
+// Register the standard-S3 engine as the default accelerated engine; the
+// registry resolves a missing `type` to it, so `accelerated=true` with no type
+// lands here too.
 objAccelEngineRegistrar reg_s3_accel(
-    "s3",
+    default_accel_engine_type,
     [](const nixlBackendInitParams *p) { return std::make_unique<S3AccelObjEngineImpl>(p); },
     [](const nixlBackendInitParams *p, std::shared_ptr<iS3Client> s3, std::shared_ptr<iS3Client>) {
         return std::make_unique<S3AccelObjEngineImpl>(p, std::move(s3));

src/plugins/obj/s3_accel/engine_impl.h

@@ -7,7 +7,6 @@
 #define OBJ_PLUGIN_S3_ACCEL_ENGINE_IMPL_H
 
 #include "s3/engine_impl.h"
-#include "s3_accel/client.h"
 
 /**
  * Generic, protocol-compliant S3-over-RDMA engine (selected by `accelerated=true`
@@ -40,8 +39,8 @@ class S3AccelObjEngineImpl : public DefaultObjEngineImpl {
     // True iff the current client is an accelerated client whose generic
     // S3-over-RDMA fast path is fully ready. Gates VRAM advertisement and buffer
     // pinning. False for an injected non-accel client (e.g. a test mock, which is
-    // not an awsS3AccelClient). Named to avoid clashing with the client's own
-    // rdmaReady().
+    // not an awsS3AccelClient). Named apart from the client's rdmaReady()
+    // because it delegates to the stronger supportsRdma().
     [[nodiscard]] bool
     rdmaEngReady() const;
 };

src/utils/object/engine_utils.h

@@ -14,23 +14,27 @@
 #include <thread>
 
 [[nodiscard]] inline std::size_t
-getNumThreads(nixl_b_params_t *custom_params) {
+getNumThreads(const nixl_b_params_t *custom_params) {
     const std::size_t fallback = std::max(1u, std::thread::hardware_concurrency() / 2);
     return nixl::getBackendParamDefaulted(custom_params, "num_threads", fallback);
 }
 
 [[nodiscard]] inline size_t
-getCrtMinLimit(nixl_b_params_t *custom_params) {
+getCrtMinLimit(const nixl_b_params_t *custom_params) {
     return nixl::getBackendParamDefaulted(custom_params, "crtMinLimit", size_t(0));
 }
 
 [[nodiscard]] inline bool
-isAcceleratedRequested(nixl_b_params_t *custom_params) {
+isAcceleratedRequested(const nixl_b_params_t *custom_params) {
     return nixl::getBackendParamDefaulted(custom_params, "accelerated", false);
 }
 
+// Accelerated engine type used when `type` is unset: the standard,
+// protocol-compliant S3-over-RDMA engine.
+inline constexpr const char *default_accel_engine_type = "s3";
+
 [[nodiscard]] inline std::string
-getAccelType(nixl_b_params_t *custom_params) {
+getAccelType(const nixl_b_params_t *custom_params) {
     return nixl::getBackendParamDefaulted(custom_params, "type", std::string());
 }
 
@@ -44,9 +48,10 @@ getAccelType(nixl_b_params_t *custom_params) {
 // caller must opt in; on a decline/failure the transfer errors rather than
 // silently falling back to HTTP.
 [[nodiscard]] inline bool
-isGenericAccelRequested(nixl_b_params_t *custom_params) {
+isGenericAccelRequested(const nixl_b_params_t *custom_params) {
     const std::string type = getAccelType(custom_params);
-    return isAcceleratedRequested(custom_params) && (type.empty() || type == "s3");
+    return isAcceleratedRequested(custom_params) &&
+        (type.empty() || type == default_accel_engine_type);
 }
 
 #endif

src/utils/object/rdma/s3_control_plane_http.cpp

@@ -54,9 +54,9 @@ namespace {
     // because that signer hashes the empty body over plain HTTP — the S3 RDMA
     // server only skips content-sha256 validation when the header is exactly
     // UNSIGNED-PAYLOAD, and the data here travels out-of-band over RDMA. This
-    // mirrors the standard S3 SigV4 signing. All non-signed headers (host,
-    // x-amz-rdma-token, content-*) must already be set on the request
-    // before calling.
+    // mirrors the standard S3 SigV4 signing. Op headers (x-amz-rdma-token,
+    // content-*) must be set before calling; this adds host, x-amz-date and
+    // x-amz-content-sha256 and signs every header on the request.
     void
     signV4(Aws::Http::HttpRequest &req,
            const Aws::String &access_key,

test/gtest/plugins/meson.build

@@ -53,7 +53,7 @@ plugins_test_exe = executable('plugins_gtest',
     sources : plugins_gtest_sources,
     include_directories: [nixl_inc_dirs, utils_inc_dirs, plugins_inc_dirs, '.'],
     cpp_args : plugins_gtest_cpp_flags,
-    dependencies : [nixl_dep, gtest_dep, absl_strings_dep, absl_time_dep, plugin_deps,
+    dependencies : [nixl_dep, gtest_dep, gmock_dep, absl_strings_dep, absl_time_dep, plugin_deps,
                     obj_backend_interface] + plugins_gtest_cuda_deps,
     link_with: [nixl_build_lib],
     install : true

test/gtest/plugins/obj_cuobj_plugin.cpp

@@ -17,6 +17,7 @@
 
 #if defined HAVE_CUOBJ_CLIENT
 
+#include <gmock/gmock.h>
 #include <gtest/gtest.h>
 #include <cstdlib>
 
@@ -35,6 +36,7 @@ nixl_b_params_t obj_dell_params = {{"accelerated", "true"},
                                    {"type", "dell"},
                                    {"req_checksum", "required"},
                                    {"scheme", "http"}};
+constexpr const char *endpoint_env_var = "NIXL_OBJ_ENDPOINT_OVERRIDE";
 const std::string accel_agent_name = "Agent3-Accel";
 const std::string dell_agent_name = "Agent4-Dell";
 
@@ -59,32 +61,37 @@ const nixlBackendInitParams obj_dell_test_params = {.localAgent = dell_agent_nam
 class setupObjAccelTestFixture : public setupBackendTestFixture {
 protected:
     nixl_b_params_t localParams_;
-    std::string skipReason_;
+    std::string constructionError_;
 
     setupObjAccelTestFixture() {
         localParams_ = *GetParam().customParams;
-        const char *endpoint = std::getenv("NIXL_OBJ_ENDPOINT_OVERRIDE");
+        const char *endpoint = std::getenv(endpoint_env_var);
         if (endpoint && endpoint[0] != '\0') {
             localParams_["endpoint_override"] = endpoint;
             localParams_["req_checksum"] = "required";
-        }
-        nixlBackendInitParams initParams = GetParam();
-        initParams.customParams = &localParams_;
-        // accelerated=true has no HTTP fallback, so the engine throws when the RDMA
-        // fast path is unavailable (e.g. a cuObject build with no RDMA NIC, as the
-        // CI container produces). Skip these tests there rather than fail.
-        try {
-            localBackendEngine_ = std::make_shared<nixlObjEngine>(&initParams);
-        }
-        catch (const std::exception &e) {
-            skipReason_ = e.what();
+            nixlBackendInitParams initParams = GetParam();
+            initParams.customParams = &localParams_;
+            // accelerated=true has no HTTP fallback, so the engine throws when the
+            // fast path is missing. SetUp() decides whether that is a skip.
+            try {
+                localBackendEngine_ = std::make_shared<nixlObjEngine>(&initParams);
+            }
+            catch (const std::exception &e) {
+                constructionError_ = e.what();
+            }
         }
     }
 
     void
     SetUp() override {
+        const char *endpoint = std::getenv(endpoint_env_var);
+        if (!endpoint || endpoint[0] == '\0') {
+            GTEST_SKIP() << endpoint_env_var << " not set, skipping accelerated tests";
+        }
         if (!localBackendEngine_) {
-            GTEST_SKIP() << "S3 accelerated engine unavailable: " << skipReason_;
+            // Only a missing fast path is a skip; anything else must fail.
+            ASSERT_THAT(constructionError_, testing::HasSubstr("fast path is unavailable"));
+            GTEST_SKIP() << "S3 accelerated engine unavailable: " << constructionError_;
         }
         setupBackendTestFixture::SetUp();
     }
@@ -167,7 +174,7 @@ class setupObjDellTestFixture : public setupBackendTestFixture {
 
     setupObjDellTestFixture() {
         localParams_ = *GetParam().customParams;
-        const char *endpoint = std::getenv("NIXL_OBJ_ENDPOINT_OVERRIDE");
+        const char *endpoint = std::getenv(endpoint_env_var);
         if (endpoint && endpoint[0] != '\0') {
             localParams_["endpoint_override"] = endpoint;
             nixlBackendInitParams initParams = GetParam();
@@ -178,9 +185,9 @@ class setupObjDellTestFixture : public setupBackendTestFixture {
 
     void
     SetUp() override {
-        const char *endpoint = std::getenv("NIXL_OBJ_ENDPOINT_OVERRIDE");
+        const char *endpoint = std::getenv(endpoint_env_var);
         if (!endpoint || endpoint[0] == '\0') {
-            GTEST_SKIP() << "NIXL_OBJ_ENDPOINT_OVERRIDE not set, skipping Dell tests";
+            GTEST_SKIP() << endpoint_env_var << " not set, skipping Dell tests";
         }
         setupBackendTestFixture::SetUp();
     }