refactor: introduce rdma_factory and make ENABLE_RDMA build optional

Add rdma_factory (make_rdma_sender / make_rdma_receiver) as a transport
backend abstraction layer to eliminate #ifdef LIMESTONE_ENABLE_RDMA from
business logic. ENABLE_RDMA=ON links rdma_comm_sender/receiver;
ENABLE_RDMA=OFF uses null backends that return failure on initialize()
to prevent silent no-ops.

This completes the OSS preparation steps:
- Introduce rdma_sender_base / rdma_receiver_base / rdma_send_stream_base
  as abstract transport interfaces
- Add null backend (rdma/null/) and rdma_comm backend (rdma/rdma_comm/)
- Remove #ifdef from datastore_impl.cpp and replica_server.cpp
- Reorganize rdma/ source tree: null/ and rdma_comm/ subdirectories
- Fix null_rdma_sender/receiver::initialize() to return failure when RDMA
  is disabled, making misconfiguration visible rather than silent
- Guard RDMA-specific tests and test parameters with #ifdef
  LIMESTONE_ENABLE_RDMA across datastore_impl_test, replica_server_test,
  message_rdma_init_test, scenario_test, and datastore_replication_test
- Add ENABLE_RDMA CMake option documentation to README.md
- Add CMake Configure(PUT_ONLY=ON, ENABLE_RDMA=ON) task to tasks.json

Author: umegane

CMakeLists.txt

@@ -80,8 +80,14 @@ if (ENABLE_ALTIMETER)
     list(APPEND boost_components container)
 endif()
 find_package(Boost COMPONENTS ${boost_components} REQUIRED)
-# rdma_comm is always required because limestone core engine depends on RDMA communication.
-find_package(rdma_comm REQUIRED)
+
+option(ENABLE_RDMA "Enable RDMA transport backend (requires rdma_comm library)" OFF)
+if(ENABLE_RDMA)
+    find_package(rdma_comm REQUIRED)
+    message(STATUS "RDMA transport backend: enabled")
+else()
+    message(STATUS "RDMA transport backend: disabled (ENABLE_RDMA=OFF)")
+endif()
 
 add_subdirectory(third_party) # should be before enable_testing()
 

README.md

@@ -47,6 +47,7 @@ available options:
 * `-DRECOVERY_SORTER_KVSLIB=<library>` - select the eKVS library using at recovery process. (`LEVELDB` or `ROCKSDB` (default), case-insensitive)
 * `-DRECOVERY_SORTER_PUT_ONLY=OFF` - don't use (faster) put-only method at recovery process
 * `-DBUILD_REPLICATION_TESTS=ON` - (temporary) enable experimental replication tests (excluded by default)
+* `-DENABLE_RDMA=ON` - enable RDMA-based replication backend (requires rdma_comm library; OFF by default)
 * `-DENABLE_ALTIMETER=ON` - enable Altimeter event logging for WAL operations
 * for debugging only
   * `-DENABLE_SANITIZER=OFF` - disable sanitizers (requires `-DCMAKE_BUILD_TYPE=Debug`)

src/CMakeLists.txt

@@ -2,8 +2,16 @@ file(GLOB SOURCES
         "limestone/*.cpp"
         "limestone/replication/*.cpp"
         "limestone/partitioned_cursor/*.cpp"
+        "limestone/rdma/rdma_socket_io.cpp"
 )
 
+if(ENABLE_RDMA)
+    file(GLOB RDMA_SOURCES "limestone/rdma/rdma_comm/*.cpp")
+else()
+    file(GLOB RDMA_SOURCES "limestone/rdma/null/*.cpp")
+endif()
+list(APPEND SOURCES ${RDMA_SOURCES})
+
 add_library(${package_name}
         ${SOURCES}
 )
@@ -29,7 +37,6 @@ endif()
 
 target_link_libraries(${package_name}
         PUBLIC limestone-api
-        PRIVATE rdma_comm
         PRIVATE Boost::boost
         PRIVATE Boost::filesystem
         PRIVATE glog::glog
@@ -39,6 +46,11 @@ target_link_libraries(${package_name}
         PRIVATE OpenSSL::Crypto
 )
 
+if(ENABLE_RDMA)
+    target_compile_definitions(${package_name} PRIVATE LIMESTONE_ENABLE_RDMA)
+    target_link_libraries(${package_name} PRIVATE rdma_comm)
+endif()
+
 if (ENABLE_ALTIMETER)
     target_link_libraries(${package_name}
         PRIVATE altimeter
@@ -65,9 +77,13 @@ target_link_libraries(limestone-impl
         INTERFACE nlohmann_json::nlohmann_json
         INTERFACE OpenSSL::SSL
         INTERFACE OpenSSL::Crypto
-        INTERFACE rdma_comm
 )
 
+if(ENABLE_RDMA)
+    target_compile_definitions(limestone-impl INTERFACE LIMESTONE_ENABLE_RDMA)
+    target_link_libraries(limestone-impl INTERFACE rdma_comm)
+endif()
+
 # utils
 file(GLOB DBLOGUTIL_SOURCES
         "limestone/dblogutil/*.cpp"

src/limestone/datastore.cpp

@@ -20,10 +20,8 @@
 #include <stdexcept>
 #include <future>
 #include <cerrno>
+#include <cstdint>
 #include <limits>
-#include <rdma_comm/channel_id_type.h>
-#include <rdma_comm/rdma_sender.h>
-#include <rdma_comm/unique_fd.h>
 
 #include <boost/filesystem/fstream.hpp>
 
@@ -1082,9 +1080,7 @@ void datastore::maybe_register_rdma_stream(log_channel& channel, std::size_t id)
         if (socket_fd < 0) {
             LOG_LP(FATAL) << "Failed to obtain socket fd for RDMA acknowledgements.";
         }
-        rdma::communication::unique_fd ack_fd{socket_fd};
-        auto stream_result = acquire_fn(
-            static_cast<rdma::communication::channel_id_type>(id), std::move(ack_fd));
+        auto stream_result = acquire_fn(static_cast<std::uint16_t>(id), socket_fd);
         if (! stream_result.status.success || stream_result.stream == nullptr) {
             LOG_LP(FATAL) << "Failed to acquire RDMA send stream: "
                           << stream_result.status.error_message;
@@ -1093,7 +1089,7 @@ void datastore::maybe_register_rdma_stream(log_channel& channel, std::size_t id)
     };
 
     if (impl_->has_rdma_stream_factory_for_test()) {
-        auto factory = impl_->get_rdma_stream_factory_for_test();
+        auto const* factory = impl_->get_rdma_stream_factory_for_test();
         if (factory == nullptr) {
             LOG_LP(FATAL) << "RDMA stream factory test hook missing.";
         }
@@ -1107,19 +1103,17 @@ void datastore::maybe_register_rdma_stream(log_channel& channel, std::size_t id)
     if (rdma_sender == nullptr || ! impl_->is_rdma_enabled()) {
         return;
     }
-    if (id > std::numeric_limits<rdma::communication::channel_id_type>::max()) {
+    if (id > std::numeric_limits<std::uint16_t>::max()) {
         LOG_LP(FATAL) << "RDMA channel_id overflow: id=" << id;
     }
     auto* replica_connector = channel.get_impl()->get_replica_connector();
     if (replica_connector == nullptr) {
         LOG_LP(FATAL) << "replica_connector missing during RDMA stream registration.";
     }
     auto socket_fd = impl_->rdma_ack_fd_for_test().value_or(replica_connector->get_socket_fd());
-    acquire_stream(
-        [rdma_sender](rdma::communication::channel_id_type cid, rdma::communication::unique_fd fd) {
-            return rdma_sender->get_send_stream(cid, std::move(fd));
-        },
-        socket_fd);
+    acquire_stream([rdma_sender](std::uint16_t cid, int fd) {
+        return rdma_sender->get_send_stream(cid, fd);
+    }, socket_fd);
 }
 
 } // namespace limestone::api

src/limestone/datastore_impl.cpp

@@ -44,7 +44,7 @@
 #include <replication/message_error.h>
 #include <replication/message_rdma_init.h>
 #include <replication/message_rdma_init_ack.h>
-#include <rdma_comm/rdma_config.h>
+#include <rdma/rdma_factory.h>
 #include <manifest.h>
 
 namespace limestone::api {
@@ -466,28 +466,17 @@ void datastore_impl::initialize_rdma_slots() {
 }
 
 bool datastore_impl::initialize_rdma_sender(uint32_t slot_count, uint64_t remote_dma_address) {
-    rdma::communication::rdma_config config{};
-    auto capacity = static_cast<std::size_t>(slot_count);
-    constexpr std::size_t chunk_size = 4096U;
-    config.send_buffer.region_size_bytes = capacity * chunk_size;
-    config.send_buffer.chunk_size_bytes = chunk_size;
-    config.send_buffer.ring_capacity = capacity;
-    config.remote_buffer = config.send_buffer;
-    config.completion_queue_depth = 1024U;
-    config.write_log_mode = rdma::communication::rdma_write_log_mode::full;
-
-    rdma_sender_ = std::make_unique<rdma::communication::rdma_sender>(config);
+    rdma_sender_ = make_rdma_sender(slot_count);
     auto result = rdma_sender_->initialize(remote_dma_address);
-    if (!result.success) {
+    if (! result.success) {
         rdma_sender_.reset();
-        LOG_LP(ERROR) << "rdma_sender::initialize() failed.";
+        LOG_LP(ERROR) << "rdma_sender::initialize() failed: " << result.error_message;
         return false;
     }
-
     return true;
 }
 
-rdma::communication::rdma_sender* datastore_impl::get_rdma_sender() const noexcept {
+rdma_sender_base* datastore_impl::get_rdma_sender() const noexcept {
     return rdma_sender_.get();
 }
 
@@ -506,7 +495,7 @@ bool datastore_impl::shutdown_rdma_sender() noexcept {
     return true;
 }
 
-void datastore_impl::set_rdma_sender_for_test(std::unique_ptr<rdma::communication::rdma_sender> sender) noexcept {
+void datastore_impl::set_rdma_sender_for_test(std::unique_ptr<rdma_sender_base> sender) noexcept {
     rdma_sender_ = std::move(sender);
 }
 
@@ -516,13 +505,11 @@ void datastore_impl::set_log_channel_connector_factory_for_test(
 }
 
 void datastore_impl::set_rdma_stream_factory_for_test(
-    std::function<rdma::communication::rdma_sender::stream_acquire_result(
-        rdma::communication::channel_id_type, rdma::communication::unique_fd)> factory) noexcept {
+        std::function<rdma_sender_base::stream_acquire_result(std::uint16_t, int)> factory) noexcept {
     rdma_stream_factory_for_test_ = std::move(factory);
 }
 
-std::function<rdma::communication::rdma_sender::stream_acquire_result(
-    rdma::communication::channel_id_type, rdma::communication::unique_fd)> const*
+std::function<rdma_sender_base::stream_acquire_result(std::uint16_t, int)> const*
 datastore_impl::get_rdma_stream_factory_for_test() const noexcept {
     if (rdma_stream_factory_for_test_) {
         return &rdma_stream_factory_for_test_;

src/limestone/datastore_impl.h

@@ -30,7 +30,7 @@
 #include "manifest.h"
 #include "replication/replica_connector.h"
 #include "replication/replication_endpoint.h"
-#include <rdma_comm/rdma_sender.h>
+#include <rdma/rdma_sender_base.h>
 
 namespace limestone::api {
 
@@ -121,7 +121,7 @@ class datastore_impl {
      * @brief Get RDMA sender instance if initialized.
      * @return pointer to RDMA sender or nullptr if not available.
      */
-    [[nodiscard]] rdma::communication::rdma_sender* get_rdma_sender() const noexcept;
+    [[nodiscard]] rdma_sender_base* get_rdma_sender() const noexcept;
 
     // Getter for migration_info_
     [[nodiscard]] const std::optional<manifest::migration_info>& get_migration_info() const noexcept;
@@ -224,7 +224,7 @@ class datastore_impl {
      * @param sender RDMA sender ownership to set for testing.
      * @note Test-only; do not use in production code.
      */
-    void set_rdma_sender_for_test(std::unique_ptr<rdma::communication::rdma_sender> sender) noexcept;
+    void set_rdma_sender_for_test(std::unique_ptr<rdma_sender_base> sender) noexcept;
 
     /**
      * @brief Test hook to override replica connector factory for log channels.
@@ -240,11 +240,9 @@ class datastore_impl {
      * @note Test-only; do not use in production code.
      */
     void set_rdma_stream_factory_for_test(
-        std::function<rdma::communication::rdma_sender::stream_acquire_result(
-            rdma::communication::channel_id_type, rdma::communication::unique_fd)> factory) noexcept;
+        std::function<rdma_sender_base::stream_acquire_result(std::uint16_t, int)> factory) noexcept;
 
-    [[nodiscard]] std::function<rdma::communication::rdma_sender::stream_acquire_result(
-        rdma::communication::channel_id_type, rdma::communication::unique_fd)> const*
+    [[nodiscard]] std::function<rdma_sender_base::stream_acquire_result(std::uint16_t, int)> const*
     get_rdma_stream_factory_for_test() const noexcept;
 
     /**
@@ -298,14 +296,13 @@ class datastore_impl {
     void initialize_rdma_slots();
 
     // RDMA sender owned by master for RDMA replication path.
-    std::unique_ptr<rdma::communication::rdma_sender> rdma_sender_{};
+    std::unique_ptr<rdma_sender_base> rdma_sender_{};
 
     // Test hook: factory to override log channel connector creation.
     std::function<std::unique_ptr<replication::replica_connector>()> log_channel_connector_factory_for_test_{};
 
     // Test hook: factory to override RDMA stream acquisition.
-    std::function<rdma::communication::rdma_sender::stream_acquire_result(
-        rdma::communication::channel_id_type, rdma::communication::unique_fd)> rdma_stream_factory_for_test_{};
+    std::function<rdma_sender_base::stream_acquire_result(std::uint16_t, int)> rdma_stream_factory_for_test_{};
 
     // Test hook: override ack fd used for RDMA registration.
     std::optional<int> rdma_ack_fd_for_test_{};

src/limestone/log_channel_impl.cpp

@@ -8,13 +8,15 @@
 #include "replication/message_log_entries.h"
 #include "replication/replication_message.h"
 #include "replication/socket_io.h"
-#include "replication/rdma_socket_io.h"
+#include "rdma/rdma_socket_io.h"
 #include "limestone/api/datastore.h"
 #include "limestone/logging.h"
 #include "logging_helper.h"
 
 namespace limestone::api {
 
+using limestone::replication::rdma_send_stream_base;
+
 namespace {
 
 constexpr auto rdma_flush_timeout = std::chrono::milliseconds{30000};
@@ -164,7 +166,7 @@ std::future<void> log_channel_impl::flush_rdma_stream_async() {
     return fut;
 }
 
-void log_channel_impl::set_rdma_send_stream(std::unique_ptr<rdma::communication::rdma_send_stream> stream) noexcept {
+void log_channel_impl::set_rdma_send_stream(std::unique_ptr<rdma_send_stream_base> stream) noexcept {
     std::lock_guard<std::mutex> lock(mtx_replica_connector_);
     rdma_send_stream_ = std::move(stream);
 }

src/limestone/log_channel_impl.h

@@ -21,7 +21,7 @@
 #include <vector>
 #include <boost/asio.hpp>
 #include <future>
-#include <rdma_comm/rdma_sender.h>
+#include <rdma/rdma_send_stream_base.h>
 #include <boost/filesystem.hpp>
 
 #include "limestone/api/blob_id_type.h"
@@ -30,7 +30,6 @@
 #include "limestone/status.h"
 #include "replication/replica_connector.h"
 #include "replication/socket_io.h"
-#include "replication/rdma_socket_io.h"
 #include "replication/message_log_entries.h"
 
 namespace limestone::api {
@@ -104,7 +103,7 @@ class log_channel_impl {
      * @brief Sets RDMA send stream for replication.
      * @param stream RDMA stream instance to take ownership of.
      */
-    void set_rdma_send_stream(std::unique_ptr<rdma::communication::rdma_send_stream> stream) noexcept;
+    void set_rdma_send_stream(std::unique_ptr<replication::rdma_send_stream_base> stream) noexcept;
 
     /**
      * @brief Sets the datastore reference used for BLOB operations on the RDMA path.
@@ -132,7 +131,7 @@ class log_channel_impl {
 
 private:
     std::unique_ptr<replication::replica_connector> replica_connector_;
-    std::unique_ptr<rdma::communication::rdma_send_stream> rdma_send_stream_;
+    std::unique_ptr<replication::rdma_send_stream_base> rdma_send_stream_;
     replication::socket_io rdma_serializer_io_;
     datastore* datastore_{nullptr};
     std::unique_ptr<boost::asio::thread_pool> ack_thread_pool_;

src/limestone/rdma/null/null_rdma_receiver.cpp

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2022-2025 Project Tsurugi.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <rdma/null_rdma_receiver.h>
+
+namespace limestone::replication {
+
+rdma_receiver_base::operation_result null_rdma_receiver::initialize(
+        rdma_receive_handler /*handler*/) noexcept {
+    return {false, "RDMA is not enabled in this build (ENABLE_RDMA=OFF)"};
+}
+
+rdma_receiver_base::operation_result null_rdma_receiver::shutdown() noexcept {
+    return {true, ""};
+}
+
+rdma_receiver_base::operation_result null_rdma_receiver::register_channel(
+        std::uint16_t /*channel_id*/,
+        int /*ack_socket*/) noexcept {
+    return {true, ""};
+}
+
+std::optional<std::uint64_t> null_rdma_receiver::get_dma_address() const noexcept {
+    return std::nullopt;
+}
+
+} // namespace limestone::replication

src/limestone/rdma/null/null_rdma_send_stream.cpp

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2022-2025 Project Tsurugi.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <rdma/null_rdma_send_stream.h>
+
+namespace limestone::replication {
+
+rdma_send_stream_base::send_result null_rdma_send_stream::send_bytes(
+        std::vector<std::uint8_t> const& /*payload*/,
+        std::size_t /*offset*/,
+        std::size_t length) noexcept {
+    return {true, "", length};
+}
+
+rdma_send_stream_base::send_result null_rdma_send_stream::send_all_bytes(
+        std::vector<std::uint8_t> const& /*payload*/,
+        std::size_t /*offset*/,
+        std::size_t length) noexcept {
+    return {true, "", length};
+}
+
+rdma_send_stream_base::flush_result null_rdma_send_stream::flush(
+        std::chrono::milliseconds /*timeout*/) noexcept {
+    return {true, ""};
+}
+
+} // namespace limestone::replication