[TENT] Add rule-based transport and device selection (#2079)

* [TENT] Add configuration-driven transport selector

Add TransportSelector for flexible, configuration-driven transport selection
policy while maintaining full backward compatibility.

Key features:
- Configuration-driven transport selection via JSON policy rules
- Support for segment_type filtering, device allocation, and transport priority
- Legacy mode option (use_legacy_transport_selection) for exact original behavior
- Default policies match original hardcoded behavior

Changes:
- Add TransportSelector class with SelectionContext, SelectionPolicy, SelectionResult
- Integrate TransportSelector into TransferEngineImpl
- Add legacy mode support to preserve original code path
- Restore TaskInfo fields (xport_priority, failover_count) for backward compatibility
- Add max_failover_attempts configuration option

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* [TENT] Add comprehensive unit tests for TransportSelector

Add transport_selector_test.cpp with test coverage for:
- Default policies matching original behavior (File/Memory segments)
- Transport type name parsing
- Legacy mode enable/disable
- Transport availability based on capabilities
- Priority offset for fallback scenarios
- Device mask handling
- NVLINK same-machine constraint
- ROCm memory type support
- GPU-to-GPU, CPU-to-CPU, CPU-to-GPU, GPU-to-CPU transfers

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* [TENT] Fix FakeTransport to use protected caps member

Fix compilation error by accessing Transport::caps (protected)
through helper methods instead of a separate public member.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix

* fix format issues

* Add priority-based rule

* Reformat

* Fix review comments

* Fix memory leak in endpoint_store_integration_test

When ibv_get_device_list returns a non-NULL list but num_devices == 0,
we need to call ibv_free_device_list before returning to avoid leaking
the allocated memory.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* Trigger CI

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: alogfans

docs/source/design/tent/transport-selector.md

@@ -0,0 +1,226 @@
+# TENT Transport Selector
+
+## Overview
+
+The Transport Selector is responsible for choosing the optimal transport and devices for each transfer request based on configuration policies.
+
+## Request Priority
+
+TENT uses a unified priority system across all components. See [QoS.md](qos.md) for detailed description of priority levels and their usage throughout the system.
+
+Quick reference:
+- `"high"` / `0` - High-priority requests (metadata, control, latency-sensitive)
+- `"medium"` / `1` - Medium-priority requests (interactive queries, serving)
+- `"low"` / `2` - Low-priority requests (bulk transfer, background jobs)
+
+## Configuration-Based Transport Selection
+
+Transport selection is driven by configuration with pattern-based rules.
+
+### Configuration Example
+
+```json
+{
+  "policy": [
+    {
+      "name": "high_prio_fast",
+      "segment_type": "memory",
+      "priority": "high",
+      "devices": ["mlx5_0", "mlx5_1", "mlx5_2"],
+      "transports": ["nvlink", "rdma", "shm"]
+    },
+    {
+      "name": "low_prio_slow",
+      "segment_type": "memory",
+      "priority": "low",
+      "devices": ["mlx5_0"],
+      "transports": ["rdma", "tcp"]
+    },
+    {
+      "name": "file_storage",
+      "segment_type": "file",
+      "transports": ["gds", "io_uring", "rdma"]
+    }
+  ]
+}
+```
+
+### Policy Fields
+
+| Field | Type | Required | Description |
+|-------|------|----------|-------------|
+| `name` | string | Yes | Policy identifier (for logging) |
+| `segment_type` | string | Yes | `"memory"` or `"file"` |
+| `priority` | string or int | No | Match only requests with this priority: `"high"` (0), `"medium"` (1), `"low"` (2) |
+| `devices` | array[string] | No | List of allowed device names (empty = all devices) |
+| `transports` | array[string] | No | Transport preference list (evaluated in order) |
+
+### Memory Type Filters
+
+For `memory` segments, you can filter by source/destination memory type:
+
+| Pattern | Matches |
+|---------|---------|
+| `"cuda"` | CUDA GPU memory |
+| `"cpu"` | CPU/host memory |
+| `"hip"` | ROCm/HIP GPU memory |
+| `"npu"` | Ascend NPU memory |
+| `"*"` | Any memory type |
+
+### Size Filters
+
+Restrict policies to specific transfer sizes:
+
+```json
+{
+  "name": "small_transfers",
+  "segment_type": "memory",
+  "min_size": 0,
+  "max_size": 1048576,
+  "transports": ["shm", "nvlink"]
+}
+```
+
+## Device Mask
+
+The `devices` field in a policy creates a bitmask that restricts which NICs can be used:
+
+```json
+{
+  "name": "use_nic_0_only",
+  "segment_type": "memory",
+  "devices": ["mlx5_0"],
+  "transports": ["rdma"]
+}
+```
+
+This is translated internally to a 64-bit bitmask where each bit represents one device:
+- `devices: ["mlx5_0"]` → `device_mask = 0x0001` (bit 0 set)
+- `devices: ["mlx5_1", "mlx5_2"]` → `device_mask = 0x0006` (bits 1 and 2 set)
+- `devices: []` (empty) → `device_mask = ~0ULL` (all devices)
+
+## Transport Fallback
+
+When multiple transports are listed in the `transports` array, they act as fallback options:
+
+```json
+{
+  "transports": ["rdma", "tcp"]
+}
+```
+
+This means:
+1. Try RDMA first
+2. If RDMA is unavailable or fails, fall back to TCP
+
+### Fallback with `transport_index`
+
+For programmatic control, the `transport_index` parameter selects which transport to use:
+
+```cpp
+// transport_index = 0 → First transport (rdma)
+// transport_index = 1 → Second transport (tcp)
+// transport_index = 2 → Third transport (if exists)
+auto result = selector.select(context, transports, transport_index);
+```
+
+## Default Behavior
+
+If no `policy` is configured, TENT falls back to original behavior:
+
+| Segment Type | Default Transport Order |
+|--------------|-------------------------|
+| File | GDS → IOURING → RDMA |
+| Memory | Uses `buffer_transports` order from buffer registration |
+
+## Complete Example
+
+```json
+{
+  "policy": [
+    {
+      "name": "high_priority_local",
+      "segment_type": "memory",
+      "same_machine": true,
+      "priority": "high",
+      "transports": ["nvlink", "shm"]
+    },
+    {
+      "name": "high_priority_remote",
+      "segment_type": "memory",
+      "same_machine": false,
+      "priority": "high",
+      "local_memory": "cuda",
+      "remote_memory": "cuda",
+      "devices": ["mlx5_0", "mlx5_1", "mlx5_2"],
+      "transports": ["rdma"]
+    },
+    {
+      "name": "medium_priority",
+      "segment_type": "memory",
+      "priority": "medium",
+      "transports": ["rdma"]
+    },
+    {
+      "name": "bulk_transfer",
+      "segment_type": "memory",
+      "priority": "low",
+      "min_size": 104857600,
+      "transports": ["rdma", "tcp"]
+    },
+    {
+      "name": "file_ops",
+      "segment_type": "file",
+      "transports": ["gds", "io_uring"]
+    }
+  ]
+}
+```
+
+## Unified Priority System
+
+The priority value propagates through multiple layers:
+
+```
+Request.priority (application layer)
+    ↓
+TransportSelector policy matching
+    ↓
+DeviceSelector allocation (QoS filtering)
+    ↓
+Worker thread scheduling queues
+```
+
+**Layer interactions**:
+
+1. **TransportSelector**: Policy's `priority` field filters which requests match
+2. **DeviceSelector**: `priority` parameter controls device eligibility (QoS mode)
+3. **Workers**: Separate queues per priority level with strict draining order
+
+See [QoS.md](qos.md) for details on worker scheduling and global slot coordination.
+
+## Data Flow
+
+```
+Request with priority
+    ↓
+TransportSelector.select(context, transports, transport_index)
+    ↓
+Match policy by:
+  - segment_type (file/memory)
+  - priority (exact match if specified in policy)
+  - location constraints
+  - size constraints
+    ↓
+Build device_mask from policy.devices
+    ↓
+Select transport from policy.transports[transport_index]
+    ↓
+Return SelectionResult { transport, device_mask }
+    ↓
+RdmaTransport.submitTransferTasks(batch, requests)
+    ↓
+DeviceSelector.allocate(..., request.priority, batch.device_mask)
+    ↓
+Worker scheduling (separate queues per priority)
+```

mooncake-transfer-engine/tent/config/transfer-engine.json

@@ -89,5 +89,18 @@
         "mnnvl": {
             "enable" : false
         }
-    }
+    },
+    "policy": [
+        {
+            "name": "default_memory",
+            "segment_type": "memory",
+            "devices": ["mlx5_0", "mlx5_2"],
+            "transports": ["nvlink", "rdma", "shm"]
+        },
+        {
+            "name": "file_storage",
+            "segment_type": "file",
+            "transports": ["gds", "io_uring", "rdma"]
+        }
+    ]
 }

mooncake-transfer-engine/tent/include/tent/runtime/transfer_engine_impl.h

@@ -31,6 +31,7 @@
 #include "tent/common/status.h"
 #include "tent/common/types.h"
 #include "tent/common/concurrent/thread_local_storage.h"
+#include "tent/runtime/transport_selector.h"
 
 namespace mooncake {
 namespace tent {
@@ -49,14 +50,15 @@ class ProxyManager;
 struct TaskInfo {
     TransportType type{UNSPEC};
     int sub_task_id{-1};
-    bool derived{false};  // merged by other tasks
-    int xport_priority{0};
+    bool derived{false};          // merged by other tasks
+    int xport_priority{0};        // transport priority (for fallback)
+    int failover_count{0};        // number of failover attempts
+    uint64_t device_mask{~0ULL};  // Device mask for quota allocation
     Request request;
     bool staging{false};
     TransferStatusEnum status{TransferStatusEnum::PENDING};
     volatile TransferStatusEnum staging_status{TransferStatusEnum::PENDING};
     std::chrono::steady_clock::time_point start_time{};  // For latency tracking
-    int failover_count{0};  // Number of cross-transport failover attempts
 };
 
 class TransferEngineImpl {
@@ -179,7 +181,8 @@ class TransferEngineImpl {
 
     Status lazyFreeBatch();
 
-    TransportType getTransportType(const Request& request, int priority = 0);
+    SelectionResult getTransportType(const Request& request,
+                                     int transport_index = 0);
 
     std::vector<TransportType> getSupportedTransports(
         TransportType request_type);
@@ -189,8 +192,8 @@ class TransferEngineImpl {
     void updateTaskStatusFromPoll(Batch* batch, size_t task_id,
                                   TransferStatus& task_status);
 
-    TransportType resolveTransport(const Request& req, int priority,
-                                   bool invalidate_on_fail = true);
+    SelectionResult resolveTransport(const Request& req, int transport_index,
+                                     bool invalidate_on_fail = true);
 
     Status loadTransports();
 
@@ -220,6 +223,7 @@ class TransferEngineImpl {
     std::shared_ptr<Config> conf_;
     std::shared_ptr<ControlService> metadata_;
     std::shared_ptr<Topology> topology_;
+    std::unique_ptr<TransportSelector> transport_selector_;
     bool available_;
 
     std::array<std::shared_ptr<Transport>, kSupportedTransportTypes>

mooncake-transfer-engine/tent/include/tent/runtime/transport.h

@@ -45,9 +45,10 @@ struct Capabilities {
 class Transport {
    public:
     struct SubBatch {
-        SubBatch() {}
+        SubBatch() : device_mask(~0ULL) {}
         virtual ~SubBatch() {}
         virtual size_t size() const = 0;
+        uint64_t device_mask;  // Device mask for transport selection
     };
 
     using SubBatchRef = SubBatch *;