feat(umbp): GPU Direct Storage (hipfile) SSD-to-GPU read path

src/umbp/CMakeLists.txt

@@ -431,6 +431,20 @@ target_link_libraries(
 
 target_compile_features(umbp_common PUBLIC cxx_std_17)
 
+# ---------- Optional hipfile GDS transfer engine -----------------------------
+# GdsEngine reads an SSD segment's byte range straight into GPU memory via
+# hipFileRead (no host bounce). Auto-enabled when hipfile is installed; a build
+# without it keeps only the memory engines and the host-staged SSD path.
+find_package(hipfile QUIET CONFIG GLOBAL)
+if(TARGET hip::hipfile)
+  message(STATUS "[UMBP] hipfile found — GDS transfer engine ENABLED")
+  target_sources(umbp_common PRIVATE distributed/transfer/gds_engine.cpp)
+  target_link_libraries(umbp_common PUBLIC hip::hipfile)
+  target_compile_definitions(umbp_common PUBLIC UMBP_ENABLE_GDS)
+else()
+  message(STATUS "[UMBP] hipfile not found — GDS transfer engine disabled")
+endif()
+
 # ---------- Optional Redis / RESP master metadata backend --------------------
 # Makes the master stateless by storing all metadata in an external RESP store
 # (Redis / Dragonfly / Valkey). OFF by default so existing builds are

src/umbp/distributed/peer/backend/instrumented_backend.cpp

@@ -158,12 +158,13 @@ std::vector<bool> InstrumentedBackend::BatchAbort(const std::vector<uint64_t>& s
 }
 
 std::vector<ResolvedEntry> InstrumentedBackend::BatchResolve(const std::vector<std::string>& keys,
-                                                             bool include_descs) {
+                                                             bool include_descs,
+                                                             bool allow_file_refs) {
   ops_[kResolve].batches.fetch_add(1, std::memory_order_relaxed);
   std::vector<ResolvedEntry> out;
   {
     ScopedNanos timer(ops_[kResolve].nanos);
-    out = inner_->BatchResolve(keys, include_descs);
+    out = inner_->BatchResolve(keys, include_descs, allow_file_refs);
   }
   uint64_t bytes = 0;
   for (const ResolvedEntry& r : out) {

src/umbp/distributed/peer/backend/page_backend.cpp

@@ -489,7 +489,8 @@ ResolvedEntry PageBackend::Resolve(const std::string& key) {
 }
 
 std::vector<ResolvedEntry> PageBackend::BatchResolve(const std::vector<std::string>& keys,
-                                                     bool include_descs) {
+                                                     bool include_descs, bool allow_file_refs) {
+  (void)allow_file_refs;  // pages are this medium's storage; there is no file to name
   std::vector<ResolvedEntry> out(keys.size());
   if (keys.empty()) return out;
   std::lock_guard<std::mutex> lock(mutex_);

src/umbp/distributed/peer/backend/ssd_backend.cpp

@@ -74,6 +74,14 @@ bool SsdBackend::Init(MemoryRegistrar* registrar) {
 
   registrar_ = registrar;
 
+  // Opt-in switch for the file->GPU (GDS) read path.  Off by default even when
+  // the build has hipfile; set UMBP_ENABLE_GDS=1 to route O_DIRECT SSD reads
+  // through the GdsEngine instead of the staging arena (no rebuild needed).
+  if (const char* env = std::getenv("UMBP_ENABLE_GDS")) {
+    const std::string v(env);
+    gds_enabled_ = (v == "1" || v == "true" || v == "TRUE" || v == "on" || v == "ON");
+  }
+
   // ONE buffer covering every staging page, so each page is contiguous and a
   // single registration serves the arena.  Host memory deliberately: this is
   // what the backend publishes, and publishing ordinary registered DRAM is the
@@ -138,6 +146,17 @@ void SsdBackend::Shutdown() {
   buffer_ref_ = TransferRef{};
   buffer_desc_.clear();
 
+  // Release the GDS file handles obtained lazily during resolves.
+  {
+    std::lock_guard<std::mutex> lock(gds_mutex_);
+    if (registrar_ != nullptr) {
+      for (auto& [fd, handle] : gds_handles_) {
+        if (handle != nullptr) registrar_->Deregister(TransferRef::File(fd, 0, 0, handle));
+      }
+    }
+    gds_handles_.clear();
+  }
+
   // Deregister before release, same ordering rule as PageBackend::Shutdown.
   if (staging_source_ != nullptr) staging_source_->Release();
   staging_source_.reset();
@@ -252,6 +271,18 @@ void SsdBackend::ClearLocal() {
     unshipped_events_ = 0;
   }
   for (const auto& key : migration_keys) ssd_->UnpinForMigration(key);
+  // The SSD wipe below can close and reopen segment fds, so drop the cached GDS
+  // handles (they are re-registered lazily on the next resolve) rather than risk
+  // a stale handle if an fd number is later reused.
+  {
+    std::lock_guard<std::mutex> lock(gds_mutex_);
+    if (registrar_ != nullptr) {
+      for (auto& [fd, handle] : gds_handles_) {
+        if (handle != nullptr) registrar_->Deregister(TransferRef::File(fd, 0, 0, handle));
+      }
+    }
+    gds_handles_.clear();
+  }
   clear_full_sync_pending_.store(true, std::memory_order_release);
   ssd_->ClearLocal();
 }
@@ -430,7 +461,7 @@ std::vector<bool> SsdBackend::BatchAbort(const std::vector<uint64_t>& slot_ids)
 }
 
 std::vector<ResolvedEntry> SsdBackend::BatchResolve(const std::vector<std::string>& keys,
-                                                    bool include_descs) {
+                                                    bool include_descs, bool allow_file_refs) {
   std::vector<ResolvedEntry> results(keys.size());
   if (keys.empty()) return results;
   const auto now = std::chrono::steady_clock::now();
@@ -484,6 +515,30 @@ std::vector<ResolvedEntry> SsdBackend::BatchResolve(const std::vector<std::strin
         continue;
       }
 
+      // Zero-copy GDS (UMBP_ENABLE_GDS): when the caller can read a file ref,
+      // the switch is on, a file engine is present, and the record is on an
+      // O_DIRECT fd, publish a FileRef and skip the staging arena — the reader
+      // (GdsEngine) DMAs the range into device memory.
+      //
+      // allow_file_refs is tested FIRST on purpose.  A caller that cannot use a
+      // file ref — a peer serving the wire, or an existence check reading only
+      // `found` — must fall through to the staging path and get real pages, and
+      // testing it first also spares it LocateRecord and the hipFile
+      // registration GdsHandleForFd performs as a side effect.
+      std::optional<RecordLocation> loc;
+      if (allow_file_refs && gds_enabled_ && (loc = ssd_->LocateRecord(keys[i])) &&
+          loc->direct_io && loc->fd >= 0) {
+        if (void* handle = GdsHandleForFd(loc->fd)) {
+          results[i].outcome = ResolveOutcome::kFound;
+          results[i].found = true;
+          results[i].size = loc->value_size;
+          results[i].page_size = cfg_.page_size;
+          results[i].file_ref =
+              TransferRef::File(loc->fd, loc->value_offset, loc->readable_size, handle);
+          continue;
+        }
+      }
+
       const uint64_t size = ssd_->SizeOf(keys[i]);
       const uint32_t page_count = SizeToPages(size, cfg_.page_size);
       if (size == 0) continue;
@@ -677,6 +732,22 @@ TransferRef SsdBackend::BufferRef(uint32_t buffer_index) const {
   return buffer_ref_;
 }
 
+void* SsdBackend::GdsHandleForFd(int fd) {
+  std::lock_guard<std::mutex> lock(gds_mutex_);
+  auto it = gds_handles_.find(fd);
+  if (it != gds_handles_.end()) return it->second;  // cached, possibly nullptr
+  // A zero-length RegisterFile just obtains (and ref-counts) the fd's handle;
+  // the per-key ranges are built by BatchResolve.  An invalid ref means no file
+  // engine is configured — cache nullptr so the fd is not probed again.
+  void* handle = nullptr;
+  if (registrar_ != nullptr) {
+    TransferRef ref = registrar_->RegisterFile(fd, 0, 0);
+    if (ref.IsFile()) handle = ref.gds_handle;
+  }
+  gds_handles_[fd] = handle;
+  return handle;
+}
+
 // ---------------------------------------------------------------------------
 //  Reaper
 // ---------------------------------------------------------------------------

src/umbp/distributed/peer/ssd/peer_ssd_manager.cpp

@@ -151,6 +151,15 @@ void PeerSsdManager::UnpinForMigration(const std::string& key) {
   if (inflight_reads_.empty()) reads_drained_cv_.notify_all();
 }
 
+std::optional<RecordLocation> PeerSsdManager::LocateRecord(const std::string& key) const {
+  std::lock_guard<std::mutex> lock(mutex_);
+  if (backend_ == nullptr) return std::nullopt;
+  // A key mid-eviction is a miss, exactly as PrepareRead treats it.
+  if (evicting_.count(key) != 0) return std::nullopt;
+  if (owned_.find(key) == owned_.end()) return std::nullopt;
+  return backend_->LocateRecord(key);
+}
+
 void PeerSsdManager::TouchLocked(const std::string& key) {
   auto it = owned_.find(key);
   if (it == owned_.end()) return;

src/umbp/distributed/pool/peer_pool.cpp

@@ -563,7 +563,7 @@ std::vector<bool> PeerPool::BatchAbort(const std::vector<PoolSlotRef>& slots) {
 }
 
 std::vector<PoolResolvedEntry> PeerPool::BatchResolve(const std::vector<std::string>& keys,
-                                                      bool include_descs) {
+                                                      bool include_descs, bool allow_file_refs) {
   std::shared_lock<std::shared_mutex> lifecycle_lock(lifecycle_mutex_);
   std::vector<PoolResolvedEntry> out(keys.size());
   if (keys.empty()) return out;
@@ -599,7 +599,7 @@ std::vector<PoolResolvedEntry> PeerPool::BatchResolve(const std::vector<std::str
     }
   }
   if (direct_backend != nullptr) {
-    auto resolved = direct_backend->BatchResolve(keys, include_descs);
+    auto resolved = direct_backend->BatchResolve(keys, include_descs, allow_file_refs);
     bool batch_busy = false;
     bool all_found = resolved.size() == keys.size();
     std::vector<bool> failed_but_owned(keys.size(), false);
@@ -661,7 +661,8 @@ std::vector<PoolResolvedEntry> PeerPool::BatchResolve(const std::vector<std::str
       attempted[index] |= static_cast<uint16_t>(1u << backend_id);
       if (!identity) backend_keys.push_back(keys[index]);
     }
-    auto results = backend->BatchResolve(identity ? keys : backend_keys, include_descs);
+    auto results =
+        backend->BatchResolve(identity ? keys : backend_keys, include_descs, allow_file_refs);
     for (size_t i = 0; i < indices.size() && i < results.size(); ++i) {
       const size_t index = indices[i];
       const ResolveOutcome outcome = EffectiveResolveOutcome(results[i]);