feat(server): support native CUDA IPC clients

pegaflow-core/src/backing/transfer_lock_guard.rs

@@ -81,11 +81,12 @@ mod tests {
 
     use pegaflow_proto::proto::engine::engine_server::{Engine, EngineServer};
     use pegaflow_proto::proto::engine::{
-        HealthRequest, HealthResponse, LoadRequest, LoadResponse, QueryBlocksForTransferRequest,
-        QueryBlocksForTransferResponse, QueryRequest, QueryResponse, RdmaHandshakeRequest,
-        RdmaHandshakeResponse, RegisterContextRequest, RegisterContextResponse, ReleaseRequest,
-        ReleaseResponse, ReleaseTransferLockResponse, SaveRequest, SaveResponse, SessionEvent,
-        SessionRequest, ShutdownRequest, ShutdownResponse, UnregisterRequest, UnregisterResponse,
+        FlushRequest, FlushResponse, HealthRequest, HealthResponse, LoadRequest, LoadResponse,
+        QueryBlocksForTransferRequest, QueryBlocksForTransferResponse, QueryRequest, QueryResponse,
+        RdmaHandshakeRequest, RdmaHandshakeResponse, RegisterContextRequest,
+        RegisterContextResponse, ReleaseRequest, ReleaseResponse, ReleaseTransferLockResponse,
+        SaveRequest, SaveResponse, SessionEvent, SessionRequest, ShutdownRequest, ShutdownResponse,
+        UnregisterRequest, UnregisterResponse,
     };
     use tokio_stream::wrappers::TcpListenerStream;
     use tonic::transport::Endpoint;
@@ -155,6 +156,12 @@ mod tests {
         ) -> Result<Response<ReleaseResponse>, Status> {
             Err(Status::unimplemented("stub"))
         }
+        async fn flush(
+            &self,
+            _request: Request<FlushRequest>,
+        ) -> Result<Response<FlushResponse>, Status> {
+            Err(Status::unimplemented("stub"))
+        }
         async fn unregister_context(
             &self,
             _request: Request<UnregisterRequest>,

pegaflow-core/src/internode/p2p_service.rs

@@ -20,12 +20,12 @@ use tonic::{Request, Response, Status, async_trait};
 
 use pegaflow_proto::proto::engine::engine_server::{Engine, EngineServer};
 use pegaflow_proto::proto::engine::{
-    HealthRequest, HealthResponse, LoadRequest, LoadResponse, QueryBlocksForTransferRequest,
-    QueryBlocksForTransferResponse, QueryRequest, QueryResponse, RdmaHandshakeRequest,
-    RdmaHandshakeResponse, RegisterContextRequest, RegisterContextResponse, ReleaseRequest,
-    ReleaseResponse, ReleaseTransferLockRequest, ReleaseTransferLockResponse, ResponseStatus,
-    SaveRequest, SaveResponse, SessionEvent, SessionRequest, ShutdownRequest, ShutdownResponse,
-    TransferBlockInfo, TransferSlotInfo, UnregisterRequest, UnregisterResponse,
+    FlushRequest, FlushResponse, HealthRequest, HealthResponse, LoadRequest, LoadResponse,
+    QueryBlocksForTransferRequest, QueryBlocksForTransferResponse, QueryRequest, QueryResponse,
+    RdmaHandshakeRequest, RdmaHandshakeResponse, RegisterContextRequest, RegisterContextResponse,
+    ReleaseRequest, ReleaseResponse, ReleaseTransferLockRequest, ReleaseTransferLockResponse,
+    ResponseStatus, SaveRequest, SaveResponse, SessionEvent, SessionRequest, ShutdownRequest,
+    ShutdownResponse, TransferBlockInfo, TransferSlotInfo, UnregisterRequest, UnregisterResponse,
 };
 
 use crate::{LayerBlock, PegaEngine};
@@ -260,6 +260,13 @@ impl Engine for P2pTransferService {
         Self::not_served("release")
     }
 
+    async fn flush(
+        &self,
+        _request: Request<FlushRequest>,
+    ) -> Result<Response<FlushResponse>, Status> {
+        Self::not_served("flush")
+    }
+
     async fn unregister_context(
         &self,
         _request: Request<UnregisterRequest>,

pegaflow-core/src/storage/mod.rs

@@ -5,8 +5,6 @@ pub(crate) mod transfer_lock;
 mod write_path;
 
 use bytesize::ByteSize;
-#[cfg(not(feature = "rdma"))]
-use log::warn;
 use log::{debug, info, warn};
 use std::collections::HashSet;
 use std::num::NonZeroU64;

pegaflow-proto/proto/engine.proto

@@ -46,6 +46,14 @@ message RegisterContextRequest {
   // Set by the connector for MLA; the engine derives per-layer page offsets
   // from the registered layouts. See spec.md.
   bool page_first = 18;
+  repeated CudaIpcTensor cuda_ipc_tensors = 19;
+}
+
+message CudaIpcTensor {
+  bytes handle = 1;
+  uint64 offset_bytes = 2;
+  uint64 size_bytes = 3;
+  uint64 block_stride_bytes = 4;
 }
 
 message RegisterContextResponse {
@@ -77,6 +85,7 @@ message LoadRequest {
   string load_state_shm = 4;
   repeated string layer_names = 5;
   repeated LeaseLoad loads = 6;
+  bool wait_for_completion = 7;
 }
 
 message LeaseLoad {
@@ -88,6 +97,12 @@ message LoadResponse {
   ResponseStatus status = 1;
 }
 
+message FlushRequest {}
+
+message FlushResponse {
+  ResponseStatus status = 1;
+}
+
 message QueryRequest {
   string instance_id = 1;
   repeated bytes block_hashes = 2;
@@ -203,6 +218,7 @@ service Engine {
   rpc RegisterContextBatch(RegisterContextRequest) returns (RegisterContextResponse);
   rpc Save(SaveRequest) returns (SaveResponse);
   rpc Load(LoadRequest) returns (LoadResponse);
+  rpc Flush(FlushRequest) returns (FlushResponse);
   // Prefetch query: check memory + trigger SSD prefetch or cross-node RDMA fetch for missing blocks
   rpc QueryPrefetch(QueryRequest) returns (QueryResponse);
   rpc Release(ReleaseRequest) returns (ReleaseResponse);

pegaflow-proto/src/lib.rs

@@ -7,3 +7,5 @@ pub mod proto {
         tonic::include_proto!("pegaflow");
     }
 }
+
+pub const VERSION: &str = concat!(env!("CARGO_PKG_VERSION"), "+native-cuda-ipc-v1");