14 changed files with 1630 additions and 465 deletions
--- a/docs/source/design/tent/qos.md
+++ b/docs/source/design/tent/qos.md
@ -0,0 +1,396 @@
+# TENT Quality of Service (QoS)
+
+## Overview
+
+TENT provides Quality of Service (QoS) support to ensure that high-priority requests receive preferential treatment in multi-tenant and multi-workload environments. This document describes TENT's QoS architecture and configuration.
+
+## Background
+
+In shared RDMA clusters, different types of transfers have different priority requirements:
+
+1. **Metadata and Control Messages**: Require low latency, small size
+2. **Interactive Queries**: Require low to medium latency, medium size
+3. **Bulk Data Transfer**: Can tolerate higher latency, large size
+
+Without QoS, low-priority bulk transfers can monopolize bandwidth and cause high tail latency for critical requests.
+
+TENT addresses this through:
+- **Per-worker priority queues** for intra-process isolation
+- **Global time-sliced coordination** for inter-process isolation
+- **Priority-aware device filtering** for NUMA-aware scheduling
+
+## Architecture
+
+### Priority Levels
+
+TENT supports three priority levels:
+
+| Priority | Value | Description | Use Cases |
+|----------|-------|-------------|-----------|
+| `PRIO_HIGH` | 0 | High-priority requests | Metadata, control messages, latency-sensitive operations |
+| `PRIO_MEDIUM` | 1 | Medium-priority requests | Interactive queries, serving workloads |
+| `PRIO_LOW` | 2 | Low-priority requests | Bulk data transfer, background jobs |
+
+### Per-Worker Priority Queues
+
+Each worker thread maintains separate queues for each priority level:
+
+```
+┌─────────────────────────────────────┐
+│         Worker Thread               │
+├─────────────────────────────────────┤
+│  PRIO_HIGH Queue     │              │
+│  PRIO_MEDIUM Queue   │              │
+│  PRIO_LOW Queue      │              │
+├─────────────────────────────────────┤
+│  Dequeue Priority: HIGH→MEDIUM→LOW   │
+└─────────────────────────────────────┘
+```
+
+**Scheduling Logic**:
+1. Always drain HIGH priority queue first
+2. Only process MEDIUM when HIGH is empty
+3. Only process LOW when both HIGH and MEDIUM are empty
+
+**Priority Promotion (Anti-Starvation)**:
+To prevent low-priority requests from starving indefinitely, TENT implements timeout-based priority promotion:
+- MEDIUM priority requests are promoted to HIGH after waiting too long
+- LOW priority requests are promoted to MEDIUM after waiting too long
+- Promotion checks run periodically (every 1ms by default)
+
+This ensures that:
+- High-priority requests normally never wait behind lower-priority work
+- Low-priority requests eventually get serviced even under continuous high-priority load
+
+### Global Slot Coordination
+
+For multi-process environments, TENT implements global time-sliced coordination using shared memory:
+
+```
+Time slices rotate every N milliseconds:
+
+Slot 0 (0-Nms):     Only HIGH priority requests allowed
+Slot 1 (N-2Nms):    MEDIUM + HIGH priority requests allowed  
+Slot 2 (2N-3Nms):   All priorities allowed
+...repeats...
+```
+
+**Default Configuration**: 2ms per slot (6ms full cycle)
+
+This mechanism ensures that:
+- High-priority requests get dedicated service windows
+- No process can monopolize bandwidth indefinitely
+- Fair access across process boundaries
+
+### Shared Memory Structure
+
+The global slot state is maintained in shared memory:
+
+```cpp
+struct SharedHeader {
+    uint64_t magic;              // Magic number for validation
+    int32_t version;             // Format version
+    std::atomic<int> current_slot;  // Current global slot (0, 1, or 2)
+    pthread_mutex_t global_mutex;    // For synchronization (robust)
+};
+```
+
+**Operations**:
+- Background thread rotates slot every N milliseconds
+- Workers check `canSend()` before processing requests
+- Only requests with priority ≤ slot level are processed
+
+## Configuration
+
+### Priority Filtering
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "enable_priority_filtering": true,
+      "local_rotation_interval_us": 200,
+      "priority_promotion_timeout_us": 10000
+    }
+  }
+}
+```
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `enable_priority_filtering` | bool | `true` | Enable priority-based device filtering |
+| `local_rotation_interval_us` | int | `200` | Local device priority rotation interval (microseconds) |
+| `priority_promotion_timeout_us` | int | `10000` | Timeout for priority promotion (microseconds) |
+
+### Global Coordination
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "slot_rotation_interval_ms": 2,
+      "shared_quota_shm_path": "/mooncake_rdma_slots"
+    }
+  }
+}
+```
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `slot_rotation_interval_ms` | int | `2` | Global slot rotation interval (milliseconds) |
+| `shared_quota_shm_path` | string | `""` | Shared memory path for multi-process coordination |
+
+**Note**: Leave `shared_quota_shm_path` empty to disable global coordination (single-process mode).
+
+## Usage Examples
+
+### Example 1: Latency-Critical Workload
+
+For workloads where high-priority requests must have minimal latency:
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "enable_priority_filtering": true,
+      "slot_rotation_interval_ms": 1,
+      "shared_quota_shm_path": "/mooncake_rdma_slots"
+    }
+  }
+}
+```
+
+**Effect**: High-priority requests get dedicated windows every 1ms.
+
+### Example 2: Single-Process Mode
+
+For single-process deployments where global coordination is not needed:
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "enable_priority_filtering": true,
+      "shared_quota_shm_path": ""
+    }
+  }
+}
+```
+
+**Effect**: Per-worker priority queues only, no cross-process coordination.
+
+### Example 3: Bulk-Friendly Configuration
+
+For workloads where low-priority bulk transfers should not be starved:
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "slot_rotation_interval_ms": 10,
+      "shared_quota_shm_path": "/mooncake_rdma_slots"
+    }
+  }
+}
+```
+
+**Effect**: Longer slots allow more low-priority work to complete.
+
+## Priority Assignment
+
+### Setting Request Priority
+
+Priority is assigned when creating `Request` objects. The default priority is `PRIO_HIGH`.
+
+### C++ API
+
+```cpp
+// In C++
+#include "tent/transfer_engine.h"
+
+using namespace mooncake::tent;
+
+// Create request with default priority (HIGH)
+Request req;
+req.opcode = Request::OpCode::READ;
+req.source = buffer;
+req.target_id = segment_id;
+req.target_offset = 0;
+req.length = size;
+// req.priority is PRIO_HIGH by default
+
+// Or specify priority explicitly
+req.priority = PRIO_MEDIUM;  // or PRIO_HIGH, PRIO_LOW
+
+// Submit the request
+engine.submitTransfer(batch_id, {req});
+```
+
+### Python API
+
+```python
+# In Python
+import tent
+
+# Create request with default priority (HIGH)
+req = tent.Request(
+    opcode=tent.OpCode.READ,
+    source=buffer_addr,
+    target_id=segment_id,
+    target_offset=0,
+    length=size
+)
+# req.priority is tent.PRIO_HIGH by default
+
+# Or specify priority in constructor
+req = tent.Request(
+    opcode=tent.OpCode.READ,
+    source=buffer_addr,
+    target_id=segment_id,
+    target_offset=0,
+    length=size,
+    priority=tent.PRIO_LOW  # or PRIO_HIGH, PRIO_MEDIUM
+)
+
+# Or set after creation
+req.priority = tent.PRIO_MEDIUM
+
+# Submit the request
+engine.submit_transfer(batch_id, [req])
+```
+
+### C API
+
+```c
+// In C
+#include "tent/transfer_engine.h"
+
+// Create request with priority
+tent_request_t req = {
+    .opcode = OPCODE_READ,
+    .source = buffer,
+    .target_id = segment_id,
+    .target_offset = 0,
+    .length = size,
+    .priority = 0  // 0=HIGH, 1=MEDIUM, 2=LOW
+};
+
+// Submit the request
+tent_submit(engine, batch_id, &req, 1);
+```
+
+## Performance Considerations
+
+### Trade-offs
+
+| Configuration | High-Priority Latency | Low-Priority Throughput | Fairness |
+|---------------|----------------------|------------------------|----------|
+| Short slot interval (1ms) | Excellent | Poor | High |
+| Default slot interval (2ms) | Good | Fair | High |
+| Long slot interval (10ms) | Fair | Good | Medium |
+| No global coordination | Variable | Excellent | Low (per-process only) |
+
+### Starvation Prevention
+
+TENT prevents starvation through two mechanisms:
+
+1. **Global slot mechanism**:
+   - **HIGH priority**: Never starved (always allowed in slot 0)
+   - **MEDIUM priority**: Never starved (allowed in slots 1 and 2)
+   - **LOW priority**: Never starved (always allowed in slot 2)
+
+2. **Priority promotion timeout**:
+   - Low-priority requests waiting longer than `priority_promotion_timeout_us` are promoted
+   - MEDIUM → HIGH promotion ensures medium priority gets service
+   - LOW → MEDIUM promotion ensures low priority eventually gets service
+   - Configurable via `priority_promotion_timeout_us` (default 10ms)
+
+### Tuning Guidelines
+
+1. **Start with default settings** (2ms slot interval)
+2. **Measure tail latency** for each priority level
+3. **Adjust slot interval** based on observations:
+   - If HIGH priority latency is too high: decrease interval
+   - If LOW priority throughput is too low: increase interval
+
+## Troubleshooting
+
+### Problem: High-priority requests have high latency
+
+**Symptoms**: `PRIO_HIGH` requests experiencing unexpected delays
+
+**Possible causes**:
+1. Slot interval too long
+2. Global coordination not enabled
+3. Worker threads blocked on LOW priority work
+
+**Solution**:
+```json
+{
+  "slot_rotation_interval_ms": 1,
+  "enable_priority_filtering": true
+}
+```
+
+### Problem: Low-priority transfers starved
+
+**Symptoms**: `PRIO_LOW` requests making no progress
+
+**Possible causes**:
+1. HIGH priority load is continuous
+2. Slot interval too short
+
+**Solution**: Increase slot interval to give LOW priority more time:
+```json
+{
+  "slot_rotation_interval_ms": 10
+}
+```
+
+### Problem: Shared memory creation fails
+
+**Symptoms**: Error messages about `/mooncake_rdma_slots`
+
+**Possible causes**:
+1. Permission issues (need write access to `/dev/shm`)
+2. Stale shared memory from previous run
+
+**Solution**:
+```bash
+# Remove stale shared memory
+rm -f /dev/shm/mooncake_rdma_slots
+
+# Or use a different path
+{
+  "shared_quota_shm_path": "/mooncake_rdma_slots_v2"
+}
+```
+
+## Monitoring
+
+### Traffic Statistics
+
+Monitor per-device traffic distribution:
+
+```cpp
+device_selector_->printTrafficStats();
+```
+
+Output example:
+```
+=== Device Traffic Statistics ===
+Dev 0: Total=10.5 GB, EWMA BW=45.23 Gbps, Inflight=0 bytes
+Dev 1: Total=8.2 GB, EWMA BW=42.18 Gbps, Inflight=0 bytes
+Dev 2: Total=0.5 GB, EWMA BW=38.91 Gbps, Inflight=0 bytes
+Dev 3: Total=0.3 GB, EWMA BW=39.12 Gbps, Inflight=0 bytes
+```
+
+### Priority Statistics
+
+Monitor queue depths for each priority level (requires instrumentation).
+
+## References
+
+- [TENT Overview](overview.md)
+- [TENT Slice Spraying](slice-spraying.md)
+- [TENT C++ API](cpp-api.md)
--- a/docs/source/design/tent/slice-spraying.md
+++ b/docs/source/design/tent/slice-spraying.md
@ -0,0 +1,398 @@
+# TENT Slice Spraying
+
+## Overview
+
+This document describes TENT's Slice Spraying mechanism, which enables efficient data movement in multi-rail RDMA environments through intelligent device selection and adaptive load balancing.
+
+## Background
+
+In multi-rail RDMA environments, naive round-robin striping leads to suboptimal performance because:
+
+1. **NUMA Effects**: Cross-NUMA access incurs additional latency and reduces effective bandwidth
+2. **Load Imbalance**: Static striping cannot adapt to dynamic load conditions
+3. **Heterogeneous Link Quality**: Different rails may have different effective bandwidth due to congestion or hardware characteristics
+
+TENT addresses these issues through:
+- **NUMA-aware device selection** with configurable penalties
+- **EWMA-based bandwidth estimation** for adaptive load balancing
+- **Dynamic multi-path allocation** for large transfers
+
+## Architecture
+
+### Device Selector
+
+The `DeviceSelector` component is responsible for choosing which RDMA device(s) to use for each transfer request. It operates in two modes:
+
+#### Baseline Mode (Round-Robin)
+
+When `enable_smart_scheduling = false`, the selector uses simple round-robin within the highest-priority device tier (typically local NUMA devices):
+
+```
+For each request:
+  1. Find first non-empty device tier (local NUMA preferred)
+  2. Select devices round-robin within that tier
+  3. Ignore lower-priority tiers
+```
+
+**Characteristics**:
+- Deterministic behavior
+- No runtime overhead for tracking
+- Consistent with original TE behavior
+- Does not adapt to load conditions
+
+#### Smart Mode (EWMA-Based Selection)
+
+When `enable_smart_scheduling = true`, the selector uses an EWMA-based algorithm:
+
+```
+For each request:
+  1. Calculate predicted completion time for each device:
+     predicted_time = (inflight_bytes + slice_bytes) / ewma_bandwidth
+
+  2. Apply NUMA penalty based on tier:
+     score = predicted_time × numa_tier_weights[tier]
+
+  3. Select device(s) with minimum score:
+     - Single slice: best device only
+     - Multiple slices: weighted distribution across devices
+
+  4. Update EWMA bandwidth on completion:
+     ewma_bandwidth = α × ewma_bandwidth + (1 - α) × observed_bandwidth
+     where α = bandwidth_learning_rate
+```
+
+**Characteristics**:
+- Adapts to changing load conditions
+- Prefers local NUMA devices
+- Spreads load across multiple rails
+- Higher runtime overhead
+
+### NUMA-Aware Selection
+
+Devices are organized into tiers based on NUMA distance:
+
+| Tier | Description | Default Penalty |
+|------|-------------|-----------------|
+| Rank 0 | Local NUMA | 1.0 (baseline) |
+| Rank 1 | Remote NUMA (tier 1) | 5.0 |
+| Rank 2 | Remote NUMA (tier 2) | 10.0 |
+
+The penalty is applied as a multiplier to predicted completion time, making remote devices less attractive unless local devices are heavily loaded.
+
+### EWMA Bandwidth Estimation
+
+Each device maintains an EWMA (Exponentially Weighted Moving Average) of its effective bandwidth:
+
+```
+initial_value = theoretical_bandwidth
+
+on_transfer_complete:
+  observed_bandwidth = transfer_size / transfer_time
+  ewma_bandwidth = α × ewma_bandwidth + (1 - α) × observed_bandwidth
+  ewma_bandwidth = clamp(ewma_bandwidth,
+                        0.1 × theoretical,
+                        10.0 × theoretical)
+```
+
+where `α = bandwidth_learning_rate`.
+
+**Note on terminology**: The EWMA formula uses α as the coefficient for the old value. Therefore:
+- **Lower α** (closer to 0) → more weight on new observations → **faster adaptation**
+- **Higher α** (closer to 1) → more weight on old value → **slower adaptation**
+
+Examples:
+- α = 0: `ewma_bandwidth = observed_bandwidth` (full adaptation, always use new value)
+- α = 1: `ewma_bandwidth = ewma_bandwidth` (no learning, never update)
+- α = 0.01: `ewma_bandwidth = 0.01 × old + 0.99 × new` (default, gradual adaptation)
+
+The EWMA provides:
+- **Memory**: Recent observations have more influence than old ones
+- **Stability**: Smooths out transient fluctuations
+- **Adaptability**: Tracks gradual changes in link quality
+
+### Multi-Path Allocation
+
+For large transfers, TENT distributes slices across multiple devices:
+
+**Single Path** (small requests):
+- All slices go to the single best device
+- Minimizes coordination overhead
+
+**Multi Path** (large requests):
+- **Normal mode** (99% of calls): Slices distributed proportionally to device capacity
+  - Each device gets: `(device_weight / total_weight) × num_slices`
+  - Remaining slices assigned to best device
+- **Probe mode** (1% of calls, every 100th call): Slices distributed round-robin
+  - Purpose: Ensure all devices are continuously sampled for EWMA updates
+  - Prevents EWMA starvation for less-used devices
+
+### Request Flow
+
+```
+┌──────────────┐
+│ Application  │
+└──────┬───────┘
+       │ submitTransfer()
+       ▼
+┌──────────────────────────────────────┐
+│  RdmaTransport::submitTransferTasks  │
+│  - Split large requests into slices   │
+│  - Call DeviceSelector for allocation │
+│  - Only if num_slices >= max_slice_count/2 │
+└──────┬───────────────────────────────┘
+       │
+       ▼
+┌──────────────────────────────────────┐
+│     DeviceSelector::allocate         │
+│  ┌────────────────────────────────┐  │
+│  │ smart_selection_enabled?       │  │
+│  └────┬──────────────────────┬────┘  │
+│       │ Yes                  │ No     │
+│       ▼                     ▼        │
+│  ┌─────────┐          ┌─────────┐   │
+│  │  Smart  │          │ Baseline│   │
+│  │  Mode   │          │   Mode  │   │
+│  └────┬────┘          └────┬────┘   │
+│       │                    │         │
+│       └────────┬───────────┘         │
+│                ▼                     │
+│  ┌────────────────────────────────┐  │
+│  │  Return slice_dev_ids          │  │
+│  └────────────────────────────────┘  │
+└──────────────────────────────────────┘
+```
+
+## Configuration
+
+All slice spraying parameters are configurable via the configuration file:
+
+### Core Scheduling
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "enable_smart_scheduling": true
+    }
+  }
+}
+```
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `enable_smart_scheduling` | bool | `true` | Enable EWMA-based selection (false = round-robin) |
+
+### NUMA Penalties
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "numa_penalties": [1.0, 5.0, 10.0]
+    }
+  }
+}
+```
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `numa_penalties` | array[float] | `[1.0, 5.0, 10.0]` | Penalty multipliers for each NUMA tier |
+
+**Guidelines**:
+- Higher values = stronger preference for local devices
+- Set all to `1.0` to disable NUMA awareness
+- Increase remote penalties if cross-NUMA latency is high
+
+### Bandwidth Estimation
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "bandwidth_learning_rate": 0.01,
+      "ewma_min_bandwidth_multiplier": 0.1,
+      "ewma_max_bandwidth_multiplier": 10.0
+    }
+  }
+}
+```
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `bandwidth_learning_rate` | float | `0.01` | EWMA learning rate (0.0 = full adaptation, 1.0 = no learning) |
+| `ewma_min_bandwidth_multiplier` | float | `0.1` | Minimum bandwidth as fraction of theoretical |
+| `ewma_max_bandwidth_multiplier` | float | `10.0` | Maximum bandwidth as fraction of theoretical |
+
+**Guidelines**:
+- Lower α (e.g., 0.001) → faster adaptation, more volatile → responds quickly to changes
+- Higher α (e.g., 0.1) → slower adaptation, more stable → smooths out transient fluctuations
+- Default α = 0.01 provides balanced adaptation
+- Multipliers constrain EWMA to reasonable range [0.1×, 10.0×] of theoretical bandwidth
+
+### Device Selection Scoring
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "score_jitter_range": 1e-9,
+      "score_epsilon": 1e-12
+    }
+  }
+}
+```
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `score_jitter_range` | float | `1e-9` | Random jitter range to avoid deterministic selection |
+| `score_epsilon` | float | `1e-12` | Small value to prevent division by zero |
+
+### Bandwidth Constants
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "default_bandwidth_gbps": 400.0,
+      "min_bandwidth_gbps": 10.0,
+      "max_bandwidth_gbps": 800.0
+    }
+  }
+}
+```
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `default_bandwidth_gbps` | float | `400.0` | Default NIC bandwidth when topology info unavailable |
+| `min_bandwidth_gbps` | float | `10.0` | Minimum valid NIC bandwidth (Gbps) |
+| `max_bandwidth_gbps` | float | `800.0` | Maximum valid NIC bandwidth (Gbps) |
+
+**Notes**:
+- These constants define the valid range and default for device bandwidth
+- Used in EWMA calculations and theoretical bandwidth estimation
+- If a device's reported bandwidth is outside [min, max], default_bandwidth is used
+
+## Usage Examples
+
+### Example 1: Latency-Sensitive Workload
+
+For latency-sensitive queries where local NUMA access is critical:
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "enable_smart_scheduling": true,
+      "numa_penalties": [1.0, 100.0, 1000.0],
+      "bandwidth_learning_rate": 0.001
+    }
+  }
+}
+```
+
+**Effect**: Strongly prefers local devices, slow adaptation for stability.
+
+### Example 2: Bulk Data Transfer
+
+For bulk transfers where throughput is more important than latency:
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "enable_smart_scheduling": true,
+      "numa_penalties": [1.0, 2.0, 3.0],
+      "bandwidth_learning_rate": 0.1
+    }
+  }
+}
+```
+
+**Effect**: Allows cross-NUMA transfers, fast adaptation to load.
+
+### Example 3: Baseline Mode
+
+For deterministic performance matching original TE:
+
+```json
+{
+  "transports": {
+    "rdma": {
+      "enable_smart_scheduling": false
+    }
+  }
+}
+```
+
+**Effect**: Round-robin within local NUMA tier, no adaptation, minimal overhead.
+
+## Performance Considerations
+
+### Overhead Comparison
+
+| Mode | CPU Overhead | Adaptability | NUMA Awareness |
+|------|--------------|--------------|----------------|
+| Baseline | Minimal | None | Tier-based (static) |
+| Smart | Moderate | EWMA-based | Dynamic + penalty |
+
+### When to Use Each Mode
+
+**Use Baseline Mode when**:
+- Workload is uniform and predictable
+- Deterministic performance is required
+- CPU overhead must be minimized
+- All devices are in same NUMA node
+
+**Use Smart Mode when**:
+- Workload is heterogeneous
+- Link quality varies over time
+- NUMA effects are significant
+- Maximum throughput is desired
+
+### Tuning Guidelines
+
+1. **Start with baseline mode** to establish performance baseline
+2. **Enable smart mode** with conservative parameters:
+   - `numa_penalties = [1.0, 2.0, 5.0]`
+   - `bandwidth_learning_rate = 0.01`
+3. **Monitor performance** and adjust based on observations:
+   - If cross-NUMA transfers are too frequent: increase remote penalties
+   - If adaptation is too slow (EWMA not keeping up with load changes): decrease α
+   - If performance is unstable (too much fluctuation): increase α
+
+## Troubleshooting
+
+### Problem: All requests go to cross-NUMA devices
+
+**Symptoms**: Poor performance, high latency
+
+**Diagnosis**:
+```cpp
+device_selector_->printTrafficStats();
+```
+
+**Solution**: Check `numa_penalties` configuration. Ensure local devices have lowest penalty (1.0).
+
+### Problem: Performance worse than baseline
+
+**Symptoms**: Smart mode slower than baseline mode
+
+**Possible causes**:
+1. Learning rate too high (volatile decisions)
+2. NUMA penalties too low (not preferring local)
+3. Score jitter too large (too much randomness)
+
+**Solution**: Use more conservative:
+```json
+{
+  "bandwidth_learning_rate": 0.001,
+  "numa_penalties": [1.0, 10.0, 100.0],
+  "score_jitter_range": 1e-12
+}
+```
+
+## References
+
+- [TENT Overview](overview.md)
+- [TENT QoS](qos.md)
+- [TENT C++ API](cpp-api.md)
--- a/mooncake-transfer-engine/tent/include/tent/common/types.h
+++ b/mooncake-transfer-engine/tent/include/tent/common/types.h
@ -28,6 +28,11 @@ namespace tent {
 using BatchID = uint64_t;
 using SegmentID = uint64_t;

+// QoS priority levels
+static constexpr uint8_t PRIO_HIGH = 0;
+static constexpr uint8_t PRIO_MEDIUM = 1;
+static constexpr uint8_t PRIO_LOW = 2;
+
 struct Notification {
    std::string name;
    std::string msg;
@ -44,6 +49,8 @@ struct Request {
    SegmentID target_id;
    uint64_t target_offset;
    size_t length;
+    int priority =
+        PRIO_HIGH;  // Request priority (PRIO_HIGH, PRIO_MEDIUM, PRIO_LOW)
 };

 enum TransferStatusEnum {
--- a/mooncake-transfer-engine/tent/include/tent/transfer_engine.h
+++ b/mooncake-transfer-engine/tent/include/tent/transfer_engine.h
@ -38,6 +38,7 @@ struct tent_request {
    tent_segment_id_t target_id;
    uint64_t target_offset;
    uint64_t length;
+    int priority; /* Request priority (0=HIGH, 1=MEDIUM, 2=LOW) */
 };

 typedef struct tent_request tent_request_t;
--- a/mooncake-transfer-engine/tent/include/tent/transport/rdma/quota.h
+++ b/mooncake-transfer-engine/tent/include/tent/transport/rdma/quota.h
@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-#ifndef TENT_QUOTA_H
-#define TENT_QUOTA_H
+#ifndef TENT_SELECTOR_H
+#define TENT_SELECTOR_H

 #include <atomic>
 #include <vector>
@ -30,47 +30,83 @@
 namespace mooncake {
 namespace tent {

-class SharedQuotaManager;
+// Bandwidth constants (Gbps)
+static constexpr double kDefaultBwGbps = 400.0;
+static constexpr double kMinBwGbps = 10.0;
+static constexpr double kMaxBwGbps = 800.0;
+
+class SharedSlotManager;

 /**
- * @brief DeviceQuota implements NIC selection based on adaptive feedback.
+ * @brief DeviceSelector implements NIC selection with two modes:
 *
- * Each NIC maintains a smoothed estimate of its average service time,
- * updated after each request completes. The allocator predicts the total
- * completion time of each NIC as:
+ * 1. Baseline mode (smart_selection_enabled=false): Simple round-robin
+ *    - Deterministic, no load tracking
+ *    - All devices used equally
 *
- *     predicted_time = (active_bytes / bandwidth) + avg_service_time
+ * 2. Smart mode (smart_selection_enabled=true): EWMA-based selection
+ *    - Tracks global inflight bytes per device
+ *    - Learns effective bandwidth via EWMA
+ *    - Selects device with minimal predicted completion time
+ *    - Supports multi-path for large requests
 *
- * and selects the NIC with the smallest predicted_time.
+ * Selection formula:
+ *     predicted_time = (inflight + slice_bytes) / ewma_bandwidth
 *
- * The estimator is updated using exponential smoothing:
- *
- *     avg_service_time <- (1 - alpha) * avg_service_time + alpha *
- * observed_time
+ * EWMA update:
+ *     ewma_bandwidth <- alpha * ewma_bandwidth + (1 - alpha) *
+ * observed_bandwidth
 */
-class DeviceQuota {
+class DeviceSelector {
   public:
+    // Candidate device for allocation
+    struct Candidate {
+        int dev_id;
+        double score;
+        bool is_cross_numa;
+    };
+
    struct DeviceInfo {
        int dev_id;
        double bw_gbps;
        int numa_id;
        uint64_t padding0[5];
-        std::atomic<uint64_t> active_bytes{0};
+        std::atomic<uint64_t> inflight_bytes{0};
        uint64_t padding1[7];
-        std::atomic<uint64_t> diffusion_active_bytes{0};
+        std::atomic<double> ewma_bandwidth_bps{50e9};
        uint64_t padding2[7];
-        std::atomic<double> beta0{0.0};  // Fixed latency (PCIe, setup)
-        uint64_t padding3[7];
-        std::atomic<double> beta1{1.0};  // Effective bandwidth correction
-        uint64_t padding4[7];
+        std::atomic<uint64_t> total_bytes{0};
+        uint64_t padding3[5];
+
+        uint64_t getInflightBytes() const {
+            return inflight_bytes.load(std::memory_order_relaxed);
+        }
+
+        void addInflight(uint64_t bytes) {
+            inflight_bytes.fetch_add(bytes, std::memory_order_relaxed);
+        }
+
+        void releaseInflight(uint64_t bytes) {
+            inflight_bytes.fetch_sub(bytes, std::memory_order_relaxed);
+        }
+
+        double getEwmaBandwidth() const {
+            return ewma_bandwidth_bps.load(std::memory_order_relaxed);
+        }
+
+        double getTheoreticalBandwidth() const {
+            if (bw_gbps >= kMinBwGbps && bw_gbps <= kMaxBwGbps)
+                return bw_gbps * 1e9 / 8.0;
+            return kDefaultBwGbps * 1e9 / 8.0;
+        }
    };

   public:
-    DeviceQuota() = default;
-    ~DeviceQuota() = default;
+    DeviceSelector() = default;
+    ~DeviceSelector() = default;

-    DeviceQuota(const DeviceQuota &) = delete;
-    DeviceQuota &operator=(const DeviceQuota &) = delete;
+    DeviceSelector(const DeviceSelector &) = delete;
+    DeviceSelector &operator=(const DeviceSelector &) = delete;

    Status loadTopology(std::shared_ptr<Topology> &local_topology);

@ -78,46 +114,111 @@ class DeviceQuota {

    Status enableSharedQuota(const std::string &shm_name);

+    std::shared_ptr<SharedSlotManager> getSharedSlotManager() const {
+        return slot_manager_;
+    }
+
+    // Allocate devices for a request (new API)
+    // slice_bytes: pre-calculated slice size from rdma_transport to ensure
+    // consistency
+    Status allocate(uint64_t total_length, uint32_t num_slices,
+                    uint64_t slice_bytes, const std::string &location,
+                    std::vector<int> &slice_dev_ids, int priority = PRIO_HIGH,
+                    uint64_t device_mask = ~0ULL);
+
    Status allocate(uint64_t length, const std::string &location,
                    int &chosen_dev_id);

    Status release(int dev_id, uint64_t length, double latency);

-    void setDiffusionActiveBytes(int dev_id, uint64_t value) {
-        devices_[dev_id].diffusion_active_bytes.store(
-            value, std::memory_order_relaxed);
+    void updateTrafficStats(int dev_id, uint64_t length) {
+        auto it = devices_.find(dev_id);
+        if (it != devices_.end()) {
+            it->second.total_bytes.fetch_add(length, std::memory_order_relaxed);
+        }
    }

-    uint64_t getActiveBytes(int dev_id) {
-        return devices_[dev_id].active_bytes.load(std::memory_order_relaxed);
+    void setSmartSelection(bool enable) { smart_selection_enabled_ = enable; }
+    bool getSmartSelection() const { return smart_selection_enabled_; }
+
+    void setLearningRate(double alpha) {
+        sched_params_.bandwidth_learning_rate = std::clamp(alpha, 0.0, 1.0);
    }

-    void setLearningRate(double alpha) { alpha_ = std::clamp(alpha, 0.0, 1.0); }
+    int getDeviceRank(const std::string &location, int dev_id) const;

-    void setLocalWeight(double local_weight) {
-        local_weight_ = std::clamp(local_weight, 0.0, 1.0);
+    void printTrafficStats();
+
+    void fillDevicePriorities();
+    int getDevicePriority(int dev_id) const;
+
+    struct SchedulingParams {
+        // NUMA tier penalties (rank 0 = local, should be smallest)
+        double numa_tier_weights[Topology::DevicePriorityRanks] = {1.0, 5.0,
+                                                                   10.0};
+
+        // EWMA bandwidth learning rate (0.0 = full adaptation, 1.0 = no
+        // learning)
+        double bandwidth_learning_rate = 0.01;
+
+        // Enable priority-based filtering
+        bool enable_priority_filtering = true;
+
+        // Local device priority rotation interval (microseconds)
+        uint64_t local_rotation_interval_us = 200;
+
+        // Score random jitter range (to avoid deterministic selection)
+        double score_jitter_range = 1e-9;
+
+        // Epsilon for division by zero protection
+        double score_epsilon = 1e-12;
+
+        // EWMA bandwidth bounds (multiplier of theoretical bandwidth)
+        double ewma_min_multiplier = 0.1;   // 10% of theoretical
+        double ewma_max_multiplier = 10.0;  // 1000% of theoretical
+
+        // Default bandwidth (Gbps) when topology info unavailable
+        double default_bandwidth_gbps = 400.0;  // Default NIC bandwidth
+        double min_bandwidth_gbps = 10.0;       // Minimum valid NIC bandwidth
+        double max_bandwidth_gbps = 800.0;      // Maximum valid NIC bandwidth
+
+        // Shared slot rotation interval (milliseconds)
+        int slot_rotation_interval_ms = 2;
+
+        std::vector<int> device_base_priorities;
+    };
+
+    void setSchedulingParams(const SchedulingParams &params) {
+        sched_params_ = params;
    }

-    void setDiffusionInterval(uint64_t msec) {
-        diffusion_interval_ = msec * 1000000ull;
+    const SchedulingParams &getSchedulingParams() const {
+        return sched_params_;
    }

-    void setCrossNumaAccess(bool enable = true) { allow_cross_numa_ = enable; }
-
   private:
    std::shared_ptr<Topology> local_topology_;
    std::unordered_map<int, DeviceInfo> devices_;
-    mutable std::shared_mutex rwlock_;
-    bool allow_cross_numa_ = false;
-    double alpha_ = 0.01;
-    double local_weight_ = 0.9;
-    uint64_t diffusion_interval_ = 10 * 1000000ull;
-    std::shared_ptr<SharedQuotaManager> shared_quota_;
-    bool enable_quota_ = true;
-    bool update_quota_params_ = true;
+    std::shared_ptr<SharedSlotManager> slot_manager_;
+    bool smart_selection_enabled_ = true;
+    SchedulingParams sched_params_;
+
+    Status buildCandidates(const Topology::MemEntry *entry,
+                           uint64_t slice_bytes, uint64_t device_mask,
+                           std::vector<Candidate> &candidates,
+                           int request_priority = PRIO_HIGH);
+
+    void selectSinglePath(const std::vector<Candidate> &candidates,
+                          uint32_t num_slices, uint64_t total_length,
+                          std::vector<int> &slice_dev_ids);
+
+    void selectMultiPath(const std::vector<Candidate> &candidates,
+                         uint32_t num_slices, uint64_t total_length,
+                         std::vector<int> &slice_dev_ids,
+                         bool probe_mode = false);
 };

 }  // namespace tent
 }  // namespace mooncake

-#endif  // TENT_QUOTA_H
+#endif  // TENT_SELECTOR_H
--- a/mooncake-transfer-engine/tent/include/tent/transport/rdma/shared_quota.h
+++ b/mooncake-transfer-engine/tent/include/tent/transport/rdma/shared_quota.h
@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-#ifndef TENT_SHARED_QUOTA_H
-#define TENT_SHARED_QUOTA_H
+#ifndef TENT_SHARED_SLOT_H
+#define TENT_SHARED_SLOT_H

 #include "quota.h"
 #include "tent/common/status.h"
@ -31,60 +31,54 @@
 #include <vector>
 #include <iostream>
 #include <errno.h>
+#include <thread>

 namespace mooncake {
 namespace tent {

-static constexpr int MAX_DEVICES = 64;
-static constexpr int MAX_PID_SLOTS = 256;
-static constexpr uint64_t SHM_MAGIC = 0x2025082772805202ULL;
-static constexpr int SHM_VERSION = 1;
+static constexpr uint64_t SHM_MAGIC = 0x2025082772805203ULL;
+static constexpr int SHM_VERSION = 10;

-struct PidUsage {
-    pid_t pid;                     // 0 == free slot
-    volatile uint64_t used_bytes;  // local used bytes reported by this pid
-    uint8_t reserved[56];          // padding -> total 64B
-};
-
-struct SharedDeviceEntry {
-    char dev_name[56];  // NUL-terminated device name, empty means unused
-    volatile uint64_t active_bytes;
-    PidUsage pid_usages[MAX_PID_SLOTS];
-};
+// Time slice configuration for process-level coordination
+// Slot 0:            HIGH only
+// Slot 1:            MEDIUM + HIGH
+// Slot 2:            ALL (LOW + MEDIUM + HIGH)
+// Then repeat
+static constexpr int NUM_SLOTS = PRIO_LOW + 1;

 struct SharedHeader {
    uint64_t magic;
    int32_t version;
-    int32_t num_devices;
+    std::atomic<int> current_slot;
    pthread_mutex_t global_mutex;
-    SharedDeviceEntry devices[MAX_DEVICES];
 };

-class DeviceQuota;
-class SharedQuotaManager {
+class DeviceSelector;
+class SharedSlotManager {
   public:
-    explicit SharedQuotaManager(DeviceQuota* local_quota);
-    ~SharedQuotaManager();
+    explicit SharedSlotManager(DeviceSelector* local_quota);
+    ~SharedSlotManager();

    Status attach(const std::string& shm_name);
    Status detach();

-    Status diffusion();
+    // Check if current process can send (global slot)
+    // Returns true if given priority is allowed in current global slot
+    bool canSend(int priority = PRIO_HIGH);
+
+    // Set slot duration in milliseconds (must be > 0)
+    void setRotationIntervalMs(int ms) { rotation_interval_ms_ = ms; }
+    int getRotationIntervalMs() const { return rotation_interval_ms_; }

   private:
-    Status attachProcess();
-    Status detachProcess();
-
-   private:
-    PidUsage* findOrCreatePidSlotLocked(int dev_id, pid_t pid);
-    PidUsage* findPidSlotLocked(int dev_id, pid_t pid);
-    int findDeviceIdByNameLocked(const std::string& dev_name);
+    void startBackgroundThread();
+    void stopBackgroundThread();
+    void backgroundThreadLoop();
    Status initializeHeader();
    Status initMutex(pthread_mutex_t* m);
-    void reclaimDeadPidsInternal();
-    static bool isPidAlive(pid_t pid);
-    int lock();
-    int unlock();
+
+    // Check if a priority is allowed in the given slot
+    bool isPriorityAllowedInSlot(int priority, int slot) const;

   private:
    std::string name_;
@ -92,10 +86,15 @@ class SharedQuotaManager {
    int fd_;
    size_t size_;
    bool created_;
-    DeviceQuota* local_quota_;
+    DeviceSelector* device_selector_;
+    int rotation_interval_ms_ = 2;  // Default: 2ms per slot
+
+    // Background thread
+    std::thread background_thread_;
+    std::atomic<bool> background_running_;
 };

 }  // namespace tent
 }  // namespace mooncake

-#endif  // TENT_SHARED_QUOTA_H
+#endif  // TENT_SHARED_SLOT_H
--- a/mooncake-transfer-engine/tent/include/tent/transport/rdma/slice.h
+++ b/mooncake-transfer-engine/tent/include/tent/transport/rdma/slice.h
@ -96,6 +96,7 @@ struct RdmaSlice {
    // WorkerContext::rails stores values via unique_ptr, so rehashes do
    // not invalidate the pointee.
    RailMonitor* rail_monitor = nullptr;
+    int priority = PRIO_HIGH;
 };

 static inline void updateSliceStatus(RdmaSlice* slice,
--- a/mooncake-transfer-engine/tent/include/tent/transport/rdma/workers.h
+++ b/mooncake-transfer-engine/tent/include/tent/transport/rdma/workers.h
@ -29,11 +29,14 @@
 #include "rail_monitor.h"
 #include "tent/common/utils/os.h"
 #include "tent/common/concurrent/bounded_mpsc_queue.h"
+#include "tent/common/types.h"

 namespace mooncake {
 namespace tent {

 class RdmaTransport;
+class DeviceSelector;
+
 class Workers {
   public:
    static constexpr size_t kCapacity = 1024 * 8;
@ -54,6 +57,8 @@ class Workers {

    Status cancel(RdmaSliceList &slice_list);

+    DeviceSelector *getDeviceSelector() const { return device_selector_.get(); }
+
   private:
    using Task = std::function<void()>;

@ -176,9 +181,11 @@ class Workers {
        PerfMetric inflight_lat;
    };

+    static constexpr int kNumPriorityLevels = PRIO_LOW + 1;
+
    struct WorkerContext {
        std::thread thread;
-        BoundedSliceQueue queue;
+        BoundedSliceQueue queues[kNumPriorityLevels];  // Priority queues
        GroupedRequests requests;
        std::unordered_set<RdmaSlice *> inflight_slice_set;
        std::atomic<int64_t> inflight_slices = 0;
@ -187,18 +194,25 @@ class Workers {
        std::condition_variable cv;
        volatile bool in_suspend = false;

+        // Next time to check for priority promotions (nanoseconds)
+        uint64_t next_promotion_check_ns = 0;
+
        // Values are held via unique_ptr so that map rehashing does not
        // invalidate pointers into RailMonitor stored on in-flight slices
        // (see RdmaSlice::rail_monitor).
        std::unordered_map<std::string, std::unique_ptr<RailMonitor>> rails;
        PerfMetricSummary perf;
-        uint64_t padding[16];
+        uint64_t padding[15];
    };

+    // Promote timed-out low priority requests to higher priority queues
+    void promoteTimedOutRequests(WorkerContext &worker);
+
    WorkerContext *worker_context_;
    uint64_t slice_timeout_ns_;
+    uint64_t priority_promotion_timeout_ns_;  // Timeout for priority promotion

-    std::unique_ptr<DeviceQuota> device_quota_;
+    std::unique_ptr<DeviceSelector> device_selector_;
    bool always_tier1_ = false;
 };
 }  // namespace tent
--- a/mooncake-transfer-engine/tent/src/python/pybind.cpp
+++ b/mooncake-transfer-engine/tent/src/python/pybind.cpp
@ -259,6 +259,11 @@ PYBIND11_MODULE(tent, m) {
    m.attr("LOCAL_SEGMENT_ID") = py::int_(LOCAL_SEGMENT_ID);
    m.attr("kWildcardLocation") = py::str(kWildcardLocation);

+    // Priority constants
+    m.attr("PRIO_HIGH") = py::int_(PRIO_HIGH);
+    m.attr("PRIO_MEDIUM") = py::int_(PRIO_MEDIUM);
+    m.attr("PRIO_LOW") = py::int_(PRIO_LOW);
+
    // -------------------------------------------------------------------------
    // Enums
    // -------------------------------------------------------------------------
@ -314,17 +319,19 @@ PYBIND11_MODULE(tent, m) {
        .def(py::init<>())
        .def(py::init([](Request::OpCode opcode, uint64_t source,
                         uint64_t target_id, uint64_t target_offset,
-                         size_t length) {
+                         size_t length, int priority) {
                 Request r;
                 r.opcode = opcode;
                 r.source = U64ToPtr(source);
                 r.target_id = target_id;
                 r.target_offset = target_offset;
                 r.length = length;
+                 r.priority = priority;
                 return r;
             }),
             py::arg("opcode"), py::arg("source"), py::arg("target_id"),
-             py::arg("target_offset"), py::arg("length"))
+             py::arg("target_offset"), py::arg("length"),
+             py::arg("priority") = PRIO_HIGH)
        .def_property(
            "opcode", [](const Request& r) { return r.opcode; },
            [](Request& r, Request::OpCode op) { r.opcode = op; })
@ -333,7 +340,8 @@ PYBIND11_MODULE(tent, m) {
            [](Request& r, uint64_t addr) { r.source = U64ToPtr(addr); })
        .def_readwrite("target_id", &Request::target_id)
        .def_readwrite("target_offset", &Request::target_offset)
-        .def_readwrite("length", &Request::length);
+        .def_readwrite("length", &Request::length)
+        .def_readwrite("priority", &Request::priority);

    py::class_<TransferStatus>(m, "TransferStatus")
        .def(py::init<>())
--- a/mooncake-transfer-engine/tent/src/transfer_engine_c.cpp
+++ b/mooncake-transfer-engine/tent/src/transfer_engine_c.cpp
@ -216,6 +216,7 @@ int tent_submit(tent_engine_t engine, tent_batch_id_t batch_id,
        req_list[index].target_id = entries[index].target_id;
        req_list[index].target_offset = entries[index].target_offset;
        req_list[index].length = entries[index].length;
+        req_list[index].priority = entries[index].priority;
    }
    auto status = CAST(engine)->submitTransfer(batch_id, req_list);
    if (!status.ok()) {
@ -241,6 +242,7 @@ int tent_submit_notif(tent_engine_t engine, tent_batch_id_t batch_id,
        req_list[index].target_id = entries[index].target_id;
        req_list[index].target_offset = entries[index].target_offset;
        req_list[index].length = entries[index].length;
+        req_list[index].priority = entries[index].priority;
    }
    mooncake::tent::Notification notifi;
    notifi.name = name;
--- a/mooncake-transfer-engine/tent/src/transport/rdma/quota.cpp
+++ b/mooncake-transfer-engine/tent/src/transport/rdma/quota.cpp
@ -15,175 +15,346 @@
 #include "tent/transport/rdma/quota.h"
 #include "tent/transport/rdma/shared_quota.h"
 #include "tent/common/utils/random.h"
+#include "tent/common/utils/os.h"

-#include <assert.h>
-#include <unordered_set>
+#include <algorithm>
+#include <iostream>
+#include <iomanip>

 namespace mooncake {
 namespace tent {
-Status DeviceQuota::loadTopology(std::shared_ptr<Topology>& local_topology) {
+Status DeviceSelector::loadTopology(std::shared_ptr<Topology>& local_topology) {
    local_topology_ = local_topology;
-    std::unordered_set<int> used_numa_id;
    for (size_t dev_id = 0; dev_id < local_topology->getNicCount(); ++dev_id) {
        auto entry = local_topology->getNicEntry(dev_id);
-        if (entry->type != Topology::NIC_RDMA) continue;
+        if (!entry || entry->type != Topology::NIC_RDMA) continue;
        DeviceInfo& info = devices_[dev_id];
        info.dev_id = dev_id;
-        info.bw_gbps = 200.0;
+        info.bw_gbps = kDefaultBwGbps;
        info.numa_id = entry->numa_node;
-        used_numa_id.insert(entry->numa_node);
+        info.ewma_bandwidth_bps.store(info.getTheoreticalBandwidth(),
+                                      std::memory_order_relaxed);
    }
-    if (used_numa_id.size() == 1) allow_cross_numa_ = true;
+    // Initialize device base priorities after all devices are loaded
+    fillDevicePriorities();
    return Status::OK();
 }

-Status DeviceQuota::enableSharedQuota(const std::string& shm_name) {
-    shared_quota_ = std::make_shared<SharedQuotaManager>(this);
-    auto status = shared_quota_->attach(shm_name);
-    if (!status.ok()) shared_quota_.reset();
+Status DeviceSelector::enableSharedQuota(const std::string& shm_name) {
+    slot_manager_ = std::make_shared<SharedSlotManager>(this);
+    slot_manager_->setRotationIntervalMs(
+        sched_params_.slot_rotation_interval_ms);
+    auto status = slot_manager_->attach(shm_name);
+    if (!status.ok()) slot_manager_.reset();
    return status;
 }

-struct TlsDeviceInfo {
-    uint64_t active_bytes{0};
-    double beta0{0.0};
-    double beta1{1.0};
-};
-
-thread_local std::unordered_map<int, TlsDeviceInfo> tl_device_info;
-
-Status DeviceQuota::allocate(uint64_t length, const std::string& location,
-                             int& chosen_dev_id) {
+Status DeviceSelector::allocate(uint64_t total_length, uint32_t num_slices,
+                                uint64_t slice_bytes,
+                                const std::string& location,
+                                std::vector<int>& slice_dev_ids, int priority,
+                                uint64_t device_mask) {
+    slice_dev_ids.clear();
+    slice_dev_ids.reserve(num_slices);
    auto entry = local_topology_->getMemEntry(location);
    if (!entry) return Status::InvalidArgument("Unknown location" LOC_MARK);

-    if (!enable_quota_) {
-        thread_local int id = 0;
+    if (!smart_selection_enabled_) {
+        // Baseline mode: consistent with original TE behavior
+        // Use devices from the first non-empty rank only
+        thread_local uint64_t tl_rr_counter = 0;
        for (size_t rank = 0; rank < Topology::DevicePriorityRanks; ++rank) {
-            auto& list = entry->device_list[rank];
-            if (list.empty()) continue;
-            chosen_dev_id = list[id % list.size()];
-            id++;
+            thread_local std::vector<int> tl_eligible;
+            tl_eligible.clear();
+            for (int dev_id : entry->device_list[rank]) {
+                if (!devices_.count(dev_id)) continue;
+                if ((device_mask & (1ULL << dev_id)) == 0) continue;
+                tl_eligible.push_back(dev_id);
+            }
+            if (tl_eligible.empty()) continue;
+
+            // Found first non-empty rank, do round-robin within this rank
+            uint64_t offset = 0;
+            for (uint32_t i = 0; i < num_slices; ++i) {
+                int dev_id = tl_eligible[tl_rr_counter % tl_eligible.size()];
+                tl_rr_counter++;
+                slice_dev_ids.push_back(dev_id);
+                uint64_t this_slice_bytes =
+                    std::min(slice_bytes, total_length - offset);
+                offset += this_slice_bytes;
+                devices_[dev_id].total_bytes.fetch_add(
+                    this_slice_bytes, std::memory_order_relaxed);
+            }
            return Status::OK();
        }
-        return Status::DeviceNotFound("no eligible devices for " + location);
+        return Status::DeviceNotFound("no eligible devices");
    }

-    static constexpr double penalty[] = {1.0, 3.0, 10.0};
-    const double w = local_weight_;
-    std::unordered_map<int, double> score_map;
-    bool found_device = false;
-    double best_score = std::numeric_limits<double>::infinity();
-    for (size_t rank = 0; rank < Topology::DevicePriorityRanks; ++rank) {
-        if (rank == Topology::DevicePriorityRanks - 1 && !allow_cross_numa_ &&
-            found_device)
-            continue;
-        for (int dev_id : entry->device_list[rank]) {
-            if (!devices_.count(dev_id)) continue;
-            auto& dev = devices_[dev_id];
-            auto& tl_dev = tl_device_info[dev_id];
-            uint64_t overall_active_bytes =
-                dev.diffusion_active_bytes.load(std::memory_order_relaxed) +
-                dev.active_bytes.load(std::memory_order_relaxed);
-            double weighted_active = w * tl_dev.active_bytes +
-                                     (1.0 - w) * overall_active_bytes + length;
-            double beta0_g = dev.beta0.load(std::memory_order_relaxed);
-            double beta1_g = dev.beta1.load(std::memory_order_relaxed);
-            double beta0 = w * tl_dev.beta0 + (1.0 - w) * beta0_g;
-            double beta1 = w * tl_dev.beta1 + (1.0 - w) * beta1_g;
-            double bw = dev.bw_gbps * 1e9 / 8;
-            double predicted_time = (weighted_active / bw) * beta1 + beta0;
-            score_map[dev_id] = penalty[rank] * predicted_time;
-            best_score = std::min(best_score, score_map[dev_id]);
-            found_device = true;
-        }
+    std::vector<DeviceSelector::Candidate> tl_candidates;
+    Status status = buildCandidates(entry, slice_bytes, device_mask,
+                                    tl_candidates, priority);
+    if (!status.ok()) return status;
+    if (num_slices == 1) {
+        selectSinglePath(tl_candidates, num_slices, total_length,
+                         slice_dev_ids);
+    } else {
+        // Probe mode: every 100th call uses round-robin distribution
+        // to ensure all devices are sampled for EWMA updates
+        thread_local uint64_t tl_call_count = 0;
+        bool probe_mode = ((++tl_call_count % 100) == 0);
+        selectMultiPath(tl_candidates, num_slices, total_length, slice_dev_ids,
+                        probe_mode);
    }
-
-    if (!found_device) {
-        return Status::DeviceNotFound("no eligible devices for " + location);
-    }
-
-    std::vector<int> filtered;
-    for (const auto& [dev_id, score] : score_map) {
-        if (score <= best_score * 1.05) filtered.push_back(dev_id);
-    }
-
-    std::sort(filtered.begin(), filtered.end(), [&](int a, int b) {
-        if (std::abs(score_map[a] - score_map[b]) > 1e-9)
-            return score_map[a] < score_map[b];
-        return a < b;
-    });
-
-    thread_local size_t rr_index = 0;
-    chosen_dev_id = filtered[rr_index % filtered.size()];
-    rr_index++;
-
-    tl_device_info[chosen_dev_id].active_bytes += length;
-    if (local_weight_ < 1 - 1e-6)
-        devices_[chosen_dev_id].active_bytes.fetch_add(
-            length, std::memory_order_relaxed);
    return Status::OK();
 }

-Status DeviceQuota::release(int dev_id, uint64_t length, double latency) {
-    if (!enable_quota_) return Status::OK();
+int DeviceSelector::getDeviceRank(const std::string& location,
+                                  int dev_id) const {
+    auto entry = local_topology_->getMemEntry(location);
+    if (!entry) return 0;
+    for (size_t rank = 0; rank < Topology::DevicePriorityRanks; ++rank) {
+        for (int id : entry->device_list[rank]) {
+            if (id == dev_id) return static_cast<int>(rank);
+        }
+    }
+    return 0;
+}
+
+Status DeviceSelector::buildCandidates(const Topology::MemEntry* entry,
+                                       uint64_t slice_bytes,
+                                       uint64_t device_mask,
+                                       std::vector<Candidate>& candidates,
+                                       int request_priority) {
+    // Helper lambda to add candidate device
+    // Score formula: predicted_time × numa_penalty + random_jitter
+    // Lower score = better candidate
+    auto add_candidate = [&](int dev_id, size_t rank) {
+        auto& dev = devices_[dev_id];
+        uint64_t inflight = dev.getInflightBytes();
+        double ewma_bw = dev.getEwmaBandwidth();
+        double predicted_time =
+            static_cast<double>(inflight + slice_bytes) / ewma_bw;
+        double rank_penalty = sched_params_.numa_tier_weights[rank];
+        double score = predicted_time * rank_penalty;
+        score +=
+            (SimpleRandom::Get().next(10) * sched_params_.score_jitter_range);
+        bool is_cross_numa = (rank > 0);
+        Candidate c;
+        c.dev_id = dev_id;
+        c.score = score;
+        c.is_cross_numa = is_cross_numa;
+        candidates.push_back(c);
+    };
+
+    // First pass: filter by device priority (QoS filtering)
+    for (size_t rank = 0; rank < Topology::DevicePriorityRanks; ++rank) {
+        for (int dev_id : entry->device_list[rank]) {
+            if (!devices_.count(dev_id)) continue;
+            if ((device_mask & (1ULL << dev_id)) == 0) continue;
+            // QoS: Get device's current priority slot (local, per-process)
+            // Device accepts request if dev_priority >= request_priority
+            int dev_priority = PRIO_LOW;  // Default: accept all
+            if (sched_params_.enable_priority_filtering) {
+                dev_priority = getDevicePriority(dev_id);
+            }
+            if (dev_priority < request_priority) continue;
+            add_candidate(dev_id, rank);
+        }
+    }
+
+    // If no devices after filtering, fallback to all devices
+    if (candidates.empty()) {
+        for (size_t rank = 0; rank < Topology::DevicePriorityRanks; ++rank) {
+            for (int dev_id : entry->device_list[rank]) {
+                if (!devices_.count(dev_id)) continue;
+                if ((device_mask & (1ULL << dev_id)) == 0) continue;
+                add_candidate(dev_id, rank);
+            }
+        }
+    }
+
+    if (candidates.empty()) {
+        return Status::DeviceNotFound("no eligible devices");
+    }
+
+    std::sort(
+        candidates.begin(), candidates.end(),
+        [this](const Candidate& a, const Candidate& b) {
+            if (std::abs(a.score - b.score) > sched_params_.score_jitter_range)
+                return a.score < b.score;
+            return a.dev_id < b.dev_id;
+        });
+    return Status::OK();
+}
+
+void DeviceSelector::selectSinglePath(const std::vector<Candidate>& candidates,
+                                      uint32_t num_slices,
+                                      uint64_t total_length,
+                                      std::vector<int>& slice_dev_ids) {
+    if (candidates.empty()) return;
+
+    const Candidate& best = candidates[0];
+    int dev_id = best.dev_id;
+    auto& dev = devices_[dev_id];
+
+    dev.addInflight(total_length);
+    dev.total_bytes.fetch_add(total_length, std::memory_order_relaxed);
+
+    for (uint32_t i = 0; i < num_slices; ++i) {
+        slice_dev_ids.push_back(dev_id);
+    }
+}
+
+void DeviceSelector::selectMultiPath(const std::vector<Candidate>& candidates,
+                                     uint32_t num_slices, uint64_t total_length,
+                                     std::vector<int>& slice_dev_ids,
+                                     bool probe_mode) {
+    if (candidates.empty()) return;
+    uint64_t slice_bytes = (total_length + num_slices - 1) / num_slices;
+    if (probe_mode) {
+        // Probe mode: round-robin distribution to ensure all devices are
+        // sampled Activates every 100th call to prevent EWMA starvation
+        for (uint32_t i = 0; i < num_slices; ++i) {
+            const Candidate& c = candidates[i % candidates.size()];
+            slice_dev_ids.push_back(c.dev_id);
+            devices_[c.dev_id].addInflight(slice_bytes);
+            devices_[c.dev_id].total_bytes.fetch_add(slice_bytes,
+                                                     std::memory_order_relaxed);
+        }
+    } else {
+        // Normal mode: weighted distribution based on inverse score
+        // Lower score → higher weight → more slices
+        double total_weight = 0.0;
+        double max_weight = -1.0;
+        int best_dev_idx = -1;
+        for (size_t i = 0; i < candidates.size(); ++i) {
+            double w =
+                1.0 / (candidates[i].score + sched_params_.score_epsilon);
+            total_weight += w;
+            if (w > max_weight) {
+                max_weight = w;
+                best_dev_idx = static_cast<int>(i);
+            }
+        }
+        if (best_dev_idx == -1 || num_slices == 0 || total_weight <= 0.0)
+            return;
+        uint32_t remaining_slices = num_slices;
+        for (size_t i = 0; i < candidates.size(); ++i) {
+            double w =
+                1.0 / (candidates[i].score + sched_params_.score_epsilon);
+            uint32_t assigned =
+                static_cast<uint32_t>((w / total_weight) * num_slices);
+            if (assigned > 0) {
+                if (assigned > remaining_slices) assigned = remaining_slices;
+                remaining_slices -= assigned;
+                const Candidate& c = candidates[i];
+                for (uint32_t s = 0; s < assigned; ++s) {
+                    slice_dev_ids.push_back(c.dev_id);
+                }
+                uint64_t total_assigned_bytes =
+                    static_cast<uint64_t>(slice_bytes) * assigned;
+                devices_[c.dev_id].addInflight(total_assigned_bytes);
+                devices_[c.dev_id].total_bytes.fetch_add(
+                    total_assigned_bytes, std::memory_order_relaxed);
+            }
+        }
+        if (remaining_slices > 0) {
+            const Candidate& c = candidates[best_dev_idx];
+            for (uint32_t s = 0; s < remaining_slices; ++s) {
+                slice_dev_ids.push_back(c.dev_id);
+            }
+            uint64_t total_assigned_bytes =
+                static_cast<uint64_t>(slice_bytes) * remaining_slices;
+            devices_[c.dev_id].addInflight(total_assigned_bytes);
+            devices_[c.dev_id].total_bytes.fetch_add(total_assigned_bytes,
+                                                     std::memory_order_relaxed);
+        }
+    }
+}
+
+Status DeviceSelector::allocate(uint64_t length, const std::string& location,
+                                int& chosen_dev_id) {
+    std::vector<int> slice_dev_ids;
+    Status status = allocate(length, 1, length, location, slice_dev_ids, ~0ULL);
+    if (!status.ok()) return status;
+    if (slice_dev_ids.empty()) {
+        return Status::DeviceNotFound("allocation failed");
+    }
+    chosen_dev_id = slice_dev_ids[0];
+    return Status::OK();
+}
+
+Status DeviceSelector::release(int dev_id, uint64_t length, double latency) {
    auto it = devices_.find(dev_id);
    if (it == devices_.end())
        return Status::InvalidArgument("device not found");

    auto& dev = it->second;
-    auto& tl_dev = tl_device_info[dev_id];
+    dev.releaseInflight(length);

-    if (local_weight_ < 1 - 1e-6)
-        dev.active_bytes.fetch_sub(length, std::memory_order_relaxed);
-    tl_dev.active_bytes -= length;
-
-    if (!update_quota_params_) return Status::OK();
-
-    double bw = dev.bw_gbps * 1e9 / 8;
-    double theory_time = static_cast<double>(length) / bw;
-    double obs_time = latency;
-
-    const double w = local_weight_;
-    double beta0_g = dev.beta0.load(std::memory_order_relaxed);
-    double beta1_g = dev.beta1.load(std::memory_order_relaxed);
-    double beta0 = w * tl_dev.beta0 + (1.0 - w) * beta0_g;
-    double beta1 = w * tl_dev.beta1 + (1.0 - w) * beta1_g;
-
-    double pred_time = beta0 + beta1 * theory_time;
-    double err = obs_time - pred_time;
-    double rel_err = (pred_time > 1e-9) ? (err / pred_time) : 0.0;
-
-    double adapt_alpha = alpha_;
-    if (std::abs(err) > 0.05 * pred_time)
-        adapt_alpha = std::min(1.0, alpha_ * 5.0);
-
-    double delta0 = adapt_alpha * err;
-    double delta1 = adapt_alpha * rel_err;
-
-    double new_beta0_l = tl_dev.beta0 + w * delta0;
-    double new_beta1_l = tl_dev.beta1 * (1.0 + w * delta1);
-    tl_dev.beta0 = std::clamp(new_beta0_l, 0.0, 5e-4);
-    tl_dev.beta1 = std::clamp(new_beta1_l, 0.5, 20.0);
-
-    if (local_weight_ < 1 - 1e-6) {
-        double new_beta0_g = beta0_g + (1.0 - w) * delta0;
-        double new_beta1_g = beta1_g * (1.0 + (1.0 - w) * delta1);
-        dev.beta0.store(std::clamp(new_beta0_g, 0.0, 5e-4),
-                        std::memory_order_relaxed);
-        dev.beta1.store(std::clamp(new_beta1_g, 0.5, 20.0),
-                        std::memory_order_relaxed);
-        if (shared_quota_) {
-            thread_local uint64_t tl_last_ts = 0;
-            uint64_t now = getCurrentTimeInNano();
-            if (now - tl_last_ts > diffusion_interval_) {
-                tl_last_ts = now;
-                return shared_quota_->diffusion();
-            }
-        }
+    if (!smart_selection_enabled_) {
+        return Status::OK();
    }
+
+    // Update EWMA bandwidth: new = α × old + (1-α) × observed
+    // α = 0: always use observed (full adaptation)
+    // α = 1: never update (no learning)
+    double observed_bw = static_cast<double>(length) / latency;
+    double current_ewma = dev.getEwmaBandwidth();
+
+    double alpha = sched_params_.bandwidth_learning_rate;
+    double new_ewma = alpha * current_ewma + (1.0 - alpha) * observed_bw;
+
+    // Clamp to [min_multiplier, max_multiplier] of theoretical bandwidth
+    double theoretical_bw = dev.getTheoreticalBandwidth();
+    new_ewma = std::max(
+        sched_params_.ewma_min_multiplier * theoretical_bw,
+        std::min(sched_params_.ewma_max_multiplier * theoretical_bw, new_ewma));
+
+    dev.ewma_bandwidth_bps.store(new_ewma, std::memory_order_relaxed);
+
    return Status::OK();
 }

+void DeviceSelector::printTrafficStats() {
+    std::cout << "=== Device Traffic Statistics ===" << std::endl;
+    for (const auto& [dev_id, dev] : devices_) {
+        uint64_t total = dev.total_bytes.load(std::memory_order_relaxed);
+        double ewma_bw_gbps = dev.getEwmaBandwidth() / 1e9 * 8.0;
+        uint64_t inflight = dev.getInflightBytes();
+        std::cout << "Dev " << dev_id << ": "
+                  << "Total=" << (total / 1024.0 / 1024.0 / 1024.0) << " GB, "
+                  << "EWMA BW=" << std::fixed << std::setprecision(2)
+                  << ewma_bw_gbps << " Gbps, "
+                  << "Inflight=" << inflight << " bytes" << std::endl;
+    }
+}
+
+void DeviceSelector::fillDevicePriorities() {
+    sched_params_.device_base_priorities.clear();
+    for (const auto& [dev_id, dev] : devices_) {
+        sched_params_.device_base_priorities.push_back(dev_id);
+    }
+}
+
+int DeviceSelector::getDevicePriority(int dev_id) const {
+    if (!sched_params_.enable_priority_filtering) return 0;
+    auto it = std::find(sched_params_.device_base_priorities.begin(),
+                        sched_params_.device_base_priorities.end(), dev_id);
+    if (it == sched_params_.device_base_priorities.end()) return 0;
+    size_t base_index =
+        std::distance(sched_params_.device_base_priorities.begin(), it);
+    size_t num_devices = sched_params_.device_base_priorities.size();
+    if (sched_params_.local_rotation_interval_us > 0 && num_devices > 0) {
+        uint64_t now = getCurrentTimeInNano();
+        uint64_t offset_us = now / 1000;
+        size_t rotation_offset =
+            (offset_us / sched_params_.local_rotation_interval_us) %
+            num_devices;
+        base_index = (base_index + rotation_offset) % num_devices;
+    }
+    return static_cast<int>(base_index);
+}
+
 }  // namespace tent
 }  // namespace mooncake
--- a/mooncake-transfer-engine/tent/src/transport/rdma/rdma_transport.cpp
+++ b/mooncake-transfer-engine/tent/src/transport/rdma/rdma_transport.cpp
@ -14,6 +14,7 @@

 #include "tent/transport/rdma/rdma_transport.h"
 #include "tent/transport/rdma/ibv_loader.h"
+#include "tent/transport/rdma/quota.h"

 #include <glog/logging.h>
 #include <sys/mman.h>
@ -326,16 +327,14 @@ Status RdmaTransport::submitTransferTasks(

    const size_t default_block_size = params_->workers.block_size;
    const int num_workers = params_->workers.num_workers;
-    const int num_devices = (size_t)local_topology_->getNicCount();
    std::vector<RdmaSliceList> slice_lists(num_workers);
    std::vector<RdmaSlice*> slice_tails(num_workers, nullptr);
    auto enqueue_ts = getCurrentTimeInNano();

+    // Distribute starting worker across threads to avoid contention
    static std::atomic<int> g_caller_threads(0);
    thread_local int tl_caller_id = g_caller_threads.fetch_add(1);
-    bool enable_spray =
-        g_caller_threads.load(std::memory_order_relaxed) <= num_workers;
-    int submit_slices = 0;
+    int next_worker_idx = tl_caller_id;
    for (auto& request : request_list) {
        auto opcode = request.opcode;
        auto type = Platform::getLoader().getMemoryType(request.source);
@ -367,8 +366,26 @@ Status RdmaTransport::submitTransferTasks(
        uint64_t block_size = roundup(
            (request.length + num_slices - 1) / num_slices, default_block_size);

-        num_slices = std::max<uint64_t>(
-            1, std::min<uint64_t>(num_slices, max_slice_count));
+        std::vector<int> slice_dev_ids;
+        // Only if a single request is enough, we perform aggregated allocation
+        if (num_slices >= max_slice_count / 2) {
+            std::string source_location = kWildcardLocation;
+            auto source_locations =
+                Platform::getLoader().getLocation(request.source, 1, true);
+            if (!source_locations.empty()) {
+                source_location = source_locations[0].location;
+            }
+            auto device_selector = workers_->getDeviceSelector();
+            if (device_selector) {
+                auto status = device_selector->allocate(
+                    request.length, static_cast<uint32_t>(num_slices),
+                    block_size, source_location, slice_dev_ids);
+                if (!status.ok() || slice_dev_ids.empty()) {
+                    LOG(WARNING) << "Device quota allocation failed: "
+                                 << status.message();
+                }
+            }
+        }

        uint64_t offset = 0;
        for (uint64_t slice_idx = 0; slice_idx < num_slices; ++slice_idx) {
@ -384,17 +401,17 @@ Status RdmaTransport::submitTransferTasks(
            slice->word = PENDING;
            slice->next = nullptr;
            slice->enqueue_ts = enqueue_ts;
+            slice->priority = request.priority;  // Copy priority from request
            task->num_slices++;
            task->ref();  // Each slice holds a reference to the task
+            if (slice_idx < slice_dev_ids.size())
+                slice->source_dev_id = slice_dev_ids[slice_idx];
            offset += length;
-            int part_id =
-                ((enable_spray ? submit_slices : static_cast<int>(slice_idx)) /
-                 num_devices) %
-                num_workers;
+            int part_id = next_worker_idx % num_workers;
            auto& list = slice_lists[part_id];
            auto& tail = slice_tails[part_id];
            list.num_slices++;
-            submit_slices++;
+            next_worker_idx++;
            if (list.first) {
                tail->next = slice;
                tail = slice;
@ -407,7 +424,7 @@ Status RdmaTransport::submitTransferTasks(
    for (int i = 0; i < num_workers; ++i) {
        if (slice_lists[i].first) {
            rdma_batch->slice_chain.push_back(slice_lists[i].first);
-            workers_->submit(slice_lists[i], (tl_caller_id + i) % num_workers);
+            workers_->submit(slice_lists[i], i);
        }
    }
    return Status::OK();
--- a/mooncake-transfer-engine/tent/src/transport/rdma/shared_quota.cpp
+++ b/mooncake-transfer-engine/tent/src/transport/rdma/shared_quota.cpp
@ -14,94 +14,43 @@

 #include "tent/transport/rdma/shared_quota.h"
 #include "tent/common/utils/os.h"
+#include "tent/common/types.h"
+
+#include <glog/logging.h>
+#include <unistd.h>

 namespace mooncake {
 namespace tent {
-SharedQuotaManager::SharedQuotaManager(DeviceQuota* local_quota)
+SharedSlotManager::SharedSlotManager(DeviceSelector* device_selector)
    : hdr_(nullptr),
      fd_(-1),
      size_(sizeof(SharedHeader)),
      created_(false),
-      local_quota_(local_quota) {}
+      device_selector_(device_selector),
+      background_running_(false) {}

-SharedQuotaManager::~SharedQuotaManager() { detach(); }
+SharedSlotManager::~SharedSlotManager() { detach(); }

-Status SharedQuotaManager::attach(const std::string& shm_name) {
-    name_ = shm_name;
-
-    // Open or create shared memory (mode 0666)
-    fd_ = shm_open(name_.c_str(), O_RDWR | O_CREAT, 0666);
-    if (fd_ < 0) {
-        return Status::InternalError("shm_open failed: " +
-                                     std::string(strerror(errno)));
-    }
-
-    // Ensure size
-    if (ftruncate(fd_, static_cast<off_t>(size_)) != 0) {
-        int e = errno;
-        close(fd_);
-        fd_ = -1;
-        return Status::InternalError("ftruncate failed: " +
-                                     std::string(strerror(e)));
-    }
-
-    // mmap
-    void* ptr =
-        mmap(nullptr, size_, PROT_READ | PROT_WRITE, MAP_SHARED, fd_, 0);
-    if (ptr == MAP_FAILED) {
-        int e = errno;
-        close(fd_);
-        fd_ = -1;
-        return Status::InternalError("mmap failed: " +
-                                     std::string(strerror(e)));
-    }
-
-    hdr_ = reinterpret_cast<SharedHeader*>(ptr);
-
-    if (hdr_->magic != SHM_MAGIC || hdr_->version != SHM_VERSION) {
-        created_ = true;
-        Status s = initializeHeader();
-        if (!s.ok()) {
-            munmap(ptr, size_);
-            close(fd_);
-            hdr_ = nullptr;
-            fd_ = -1;
-            return s;
-        }
-    } else {
-        created_ = false;
-    }
-
-    Status s = attachProcess();
-    if (!s.ok()) return s;
-    return Status::OK();
+bool SharedSlotManager::isPriorityAllowedInSlot(int priority, int slot) const {
+    return priority <= slot;
 }

-Status SharedQuotaManager::detach() {
-    if (hdr_) {
-        detachProcess();
-        munmap(hdr_, size_);
-        hdr_ = nullptr;
-    }
-    if (fd_ >= 0) {
-        close(fd_);
-        fd_ = -1;
-    }
-    return Status::OK();
-}
+Status SharedSlotManager::initializeHeader() {
+    hdr_->magic = 0;
+    hdr_->version = 0;
+    hdr_->current_slot.store(0, std::memory_order_relaxed);

-Status SharedQuotaManager::initializeHeader() {
-    memset(hdr_, 0, size_);
    Status s = initMutex(&hdr_->global_mutex);
-    if (!s.ok()) {
-        return s;
-    }
+    if (!s.ok()) return s;
+
    hdr_->version = SHM_VERSION;
    hdr_->magic = SHM_MAGIC;
+    hdr_->current_slot.store(0, std::memory_order_release);
+
    return Status::OK();
 }

-Status SharedQuotaManager::initMutex(pthread_mutex_t* m) {
+Status SharedSlotManager::initMutex(pthread_mutex_t* m) {
    pthread_mutexattr_t attr;
    if (pthread_mutexattr_init(&attr) != 0) {
        return Status::InternalError("pthread_mutexattr_init failed");
@ -124,162 +73,115 @@ Status SharedQuotaManager::initMutex(pthread_mutex_t* m) {
    return Status::OK();
 }

-// attempt to acquire global lock; handle EOWNERDEAD
-int SharedQuotaManager::lock() {
-    if (!hdr_) return EINVAL;
-    int rc = pthread_mutex_lock(&hdr_->global_mutex);
-    if (rc == 0) return 0;
-    if (rc == EOWNERDEAD) {
-        // make consistent so others can continue
-#if defined(PTHREAD_MUTEX_ROBUST)
-        int rc2 = pthread_mutex_consistent(&hdr_->global_mutex);
-        if (rc2 != 0) {
-            return rc2;
+Status SharedSlotManager::attach(const std::string& shm_name) {
+    name_ = shm_name;
+
+    fd_ = shm_open(name_.c_str(), O_RDWR | O_CREAT, 0666);
+    if (fd_ < 0) {
+        return Status::InternalError("shm_open failed: " +
+                                     std::string(std::strerror(errno)));
+    }
+
+    if (ftruncate(fd_, static_cast<off_t>(size_)) != 0) {
+        int e = errno;
+        close(fd_);
+        fd_ = -1;
+        return Status::InternalError("ftruncate failed: " +
+                                     std::string(std::strerror(e)));
+    }
+
+    void* ptr =
+        mmap(nullptr, size_, PROT_READ | PROT_WRITE, MAP_SHARED, fd_, 0);
+    if (ptr == MAP_FAILED) {
+        int e = errno;
+        close(fd_);
+        fd_ = -1;
+        return Status::InternalError("mmap failed: " +
+                                     std::string(std::strerror(e)));
+    }
+
+    hdr_ = reinterpret_cast<SharedHeader*>(ptr);
+
+    if (hdr_->magic != SHM_MAGIC || hdr_->version != SHM_VERSION) {
+        created_ = true;
+        Status s = initializeHeader();
+        if (!s.ok()) {
+            munmap(ptr, size_);
+            close(fd_);
+            hdr_ = nullptr;
+            fd_ = -1;
+            return s;
        }
-#endif
-        // We hold the lock now — repair data if needed
-        reclaimDeadPidsInternal();
-        return 0;
-    }
-    return rc;
-}
-
-int SharedQuotaManager::unlock() {
-    if (!hdr_) return EINVAL;
-    return pthread_mutex_unlock(&hdr_->global_mutex);
-}
-
-void SharedQuotaManager::reclaimDeadPidsInternal() {
-    if (!hdr_) return;
-    // Assumes caller holds lock (but we also call from lock() on EOWNERDEAD)
-    for (int i = 0; i < hdr_->num_devices; ++i) {
-        // skip unused slots (dev_name empty)
-        if (hdr_->devices[i].dev_name[0] == '\0') continue;
-        SharedDeviceEntry& dev = hdr_->devices[i];
-        for (int s = 0; s < MAX_PID_SLOTS; ++s) {
-            pid_t p = dev.pid_usages[s].pid;
-            if (p == 0) continue;
-            if (!isPidAlive(p)) {
-                // zero out slot — we'll recompute active_bytes in diffusion
-                dev.pid_usages[s].pid = 0;
-                dev.pid_usages[s].used_bytes = 0;
-            }
-        }
-    }
-}
-
-bool SharedQuotaManager::isPidAlive(pid_t pid) {
-    if (pid <= 0) return false;
-    int r = kill(pid, 0);
-    if (r == 0) return true;
-    if (errno == ESRCH) return false;
-    return true;  // other errors (EPERM) -> treat as alive
-}
-
-int SharedQuotaManager::findDeviceIdByNameLocked(const std::string& dev_name) {
-    if (!hdr_) return -1;
-    for (int i = 0; i < hdr_->num_devices; ++i) {
-        if (hdr_->devices[i].dev_name[0] == '\0') continue;
-        if (strncmp(hdr_->devices[i].dev_name, dev_name.c_str(),
-                    sizeof(hdr_->devices[i].dev_name)) == 0)
-            return i;
-    }
-    return -1;
-}
-
-PidUsage* SharedQuotaManager::findOrCreatePidSlotLocked(int dev_id, pid_t pid) {
-    if (!hdr_) return nullptr;
-    if (dev_id < 0 || dev_id >= hdr_->num_devices) return nullptr;
-    SharedDeviceEntry& dev = hdr_->devices[dev_id];
-    PidUsage* empty = nullptr;
-    for (int s = 0; s < MAX_PID_SLOTS; ++s) {
-        if (dev.pid_usages[s].pid == pid) return &dev.pid_usages[s];
-        if (dev.pid_usages[s].pid == 0 && empty == nullptr)
-            empty = &dev.pid_usages[s];
-    }
-    if (empty) {
-        empty->pid = pid;
-        empty->used_bytes = 0;
-    }
-    return empty;
-}
-
-PidUsage* SharedQuotaManager::findPidSlotLocked(int dev_id, pid_t pid) {
-    if (!hdr_) return nullptr;
-    if (dev_id < 0 || dev_id >= hdr_->num_devices) return nullptr;
-    SharedDeviceEntry& dev = hdr_->devices[dev_id];
-    for (int s = 0; s < MAX_PID_SLOTS; ++s) {
-        if (dev.pid_usages[s].pid == pid) return &dev.pid_usages[s];
-    }
-    return nullptr;
-}
-
-Status SharedQuotaManager::attachProcess() {
-    if (!hdr_) return Status::InvalidArgument("not attached");
-
-    int rc = lock();
-    if (rc != 0) {
-        return Status::InternalError("failed to lock shared mutex: " +
-                                     std::string(strerror(rc)));
+    } else {
+        created_ = false;
    }

-    auto topo = local_quota_->getTopology();
-    for (size_t i = 0; i < topo->getNicCount(); ++i) {
-        if (topo->getNicType(i) != Topology::NIC_RDMA) continue;
-        auto dev_name = topo->getNicName(i);
-        if (findDeviceIdByNameLocked(dev_name) >= 0) continue;
-        int empty_idx = -1;
-        for (int j = 0; j < MAX_DEVICES; ++j) {
-            if (hdr_->devices[j].dev_name[0] == '\0') {
-                empty_idx = j;
-                break;
-            }
-        }
-        if (empty_idx < 0) continue;
-        strncpy(hdr_->devices[empty_idx].dev_name, dev_name.c_str(), 56);
-        hdr_->devices[empty_idx].active_bytes = 0;
-    }
+    startBackgroundThread();

-    int count = 0;
-    for (int i = 0; i < MAX_DEVICES; ++i)
-        if (hdr_->devices[i].dev_name[0] != '\0') ++count;
-    hdr_->num_devices = count;
-
-    unlock();
    return Status::OK();
 }

-Status SharedQuotaManager::detachProcess() { return Status::OK(); }
+Status SharedSlotManager::detach() {
+    stopBackgroundThread();

-Status SharedQuotaManager::diffusion() {
-    if (!hdr_) return Status::InvalidArgument("not attached");
-    pid_t pid = getpid();
-    int rc = lock();
-    if (rc != 0)
-        return Status::InternalError("lock failed: " +
-                                     std::string(strerror(rc)));
-    for (int d = 0; d < hdr_->num_devices; ++d) {
-        std::string dev_name = hdr_->devices[d].dev_name;
-        auto dev_id = local_quota_->getTopology()->getNicId(dev_name);
-        if (dev_name.empty() || dev_id < 0) continue;
-        PidUsage* slot = findOrCreatePidSlotLocked(dev_id, pid);
-        if (!slot) {
-            unlock();
-            return Status::InternalError("no free pid slot for device");
-        }
-        auto used_bytes = local_quota_->getActiveBytes(dev_id);
-        slot->used_bytes = used_bytes;
-        uint64_t sum = 0;
-        for (int s = 0; s < MAX_PID_SLOTS; ++s)
-            sum += hdr_->devices[d].pid_usages[s].used_bytes;
-        uint64_t diffusion_active_bytes =
-            sum < used_bytes ? 0 : sum - used_bytes;
-        hdr_->devices[d].active_bytes = sum;
-        local_quota_->setDiffusionActiveBytes(dev_id, diffusion_active_bytes);
+    if (hdr_) {
+        munmap(hdr_, size_);
+        hdr_ = nullptr;
    }
-    unlock();
+
+    if (fd_ >= 0) {
+        close(fd_);
+        fd_ = -1;
+    }
+
    return Status::OK();
 }

+bool SharedSlotManager::canSend(int priority) {
+    if (!hdr_) return true;
+
+    // Get current global slot
+    int current_slot = hdr_->current_slot.load(std::memory_order_acquire);
+
+    // Check if the given priority is allowed in current global slot
+    // Slot 0: HIGH only (priority 0 <= 0)
+    // Slot 1: HIGH + MEDIUM (priority 0 or 1 <= 1)
+    // Slot 2: ALL (priority 0, 1, or 2 <= 2)
+    return isPriorityAllowedInSlot(priority, current_slot);
+}
+
+void SharedSlotManager::startBackgroundThread() {
+    if (background_running_.exchange(true)) return;
+
+    background_thread_ = std::thread([this]() { backgroundThreadLoop(); });
+}
+
+void SharedSlotManager::stopBackgroundThread() {
+    if (!background_running_.exchange(false)) return;
+
+    if (background_thread_.joinable()) {
+        background_thread_.join();
+    }
+}
+
+// Background thread: advance global slot periodically
+void SharedSlotManager::backgroundThreadLoop() {
+    const uint64_t SLEEP_INTERVAL_US = 1000;  // 1ms
+
+    while (background_running_.load(std::memory_order_relaxed)) {
+        // Calculate base slot from time
+        uint64_t now = getCurrentTimeInNano();
+        uint64_t base_slot = now / (rotation_interval_ms_ * 1000000ull);
+
+        pthread_mutex_lock(&hdr_->global_mutex);
+        // Update global slot
+        int global_slot = static_cast<int>(base_slot % NUM_SLOTS);
+        hdr_->current_slot.store(global_slot, std::memory_order_release);
+        pthread_mutex_unlock(&hdr_->global_mutex);
+
+        usleep(SLEEP_INTERVAL_US);
+    }
+}
+
 }  // namespace tent
 }  // namespace mooncake
--- a/mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp
+++ b/mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp
@ -19,7 +19,7 @@
 #include <cassert>

 #include "tent/transport/rdma/endpoint_store.h"
-#include "tent/transport/rdma/rail_monitor.h"
+#include "tent/transport/rdma/shared_quota.h"
 #include "tent/common/utils/ip.h"
 #include "tent/common/utils/string_builder.h"
 #include "tent/common/utils/os.h"
@ -45,23 +45,108 @@ RailMonitor& getOrCreateRail(

 Workers::Workers(RdmaTransport* transport)
    : transport_(transport), num_workers_(0), running_(false) {
-    device_quota_ = std::make_unique<DeviceQuota>();
-    device_quota_->loadTopology(transport_->local_topology_);
+    device_selector_ = std::make_unique<DeviceSelector>();
+    device_selector_->loadTopology(transport_->local_topology_);
    auto& conf = transport_->conf_;
+
+    // ============================================================
+    // Core Scheduling Configuration
+    // ============================================================
+
+    // Enable/disable smart scheduling (false = simple round-robin)
+    bool enable_smart_scheduling =
+        conf->get("transports/rdma/enable_smart_scheduling", true);
+    device_selector_->setSmartSelection(enable_smart_scheduling);
+
+    // ============================================================
+    // NUMA Distance Penalties
+    // Higher values = higher penalty for cross-NUMA access
+    // Format: [local_numa, remote_numa1, remote_numa2, ...]
+    // ============================================================
+    DeviceSelector::SchedulingParams params;
+
+    auto numa_penalties =
+        conf->get("transports/rdma/numa_penalties", std::vector<double>{});
+    if (numa_penalties.size() == Topology::DevicePriorityRanks) {
+        for (size_t i = 0; i < Topology::DevicePriorityRanks; ++i) {
+            params.numa_tier_weights[i] = numa_penalties[i];
+        }
+    }
+
+    // ============================================================
+    // Bandwidth Estimation (EWMA)
+    // ============================================================
+
+    // Learning rate: 0.0 = full adaptation, 1.0 = no adaptation
+    params.bandwidth_learning_rate =
+        conf->get("transports/rdma/bandwidth_learning_rate", 0.01);
+
+    // EWMA bounds as multipliers of theoretical bandwidth
+    params.ewma_min_multiplier =
+        conf->get("transports/rdma/ewma_min_bandwidth_multiplier", 0.1);
+    params.ewma_max_multiplier =
+        conf->get("transports/rdma/ewma_max_bandwidth_multiplier", 10.0);
+
+    // ============================================================
+    // Device Selection Scoring
+    // ============================================================
+
+    // Random jitter to avoid deterministic selection
+    params.score_jitter_range =
+        conf->get("transports/rdma/score_jitter_range", 1e-9);
+
+    // Small value to prevent division by zero
+    params.score_epsilon = conf->get("transports/rdma/score_epsilon", 1e-12);
+
+    // ============================================================
+    // Priority-Based Filtering
+    // ============================================================
+
+    params.enable_priority_filtering =
+        conf->get("transports/rdma/enable_priority_filtering", true);
+
+    // Local device priority rotation interval (microseconds)
+    params.local_rotation_interval_us =
+        conf->get("transports/rdma/local_rotation_interval_us", 200);
+
+    // ============================================================
+    // Priority Promotion (Anti-Starvation)
+    // ============================================================
+
+    // Timeout after which low-priority requests get promoted (nanoseconds)
+    // Default: 10ms (10000000 ns)
+    priority_promotion_timeout_ns_ =
+        conf->get("transports/rdma/priority_promotion_timeout_us", 10000) *
+        1000ull;
+
+    // ============================================================
+    // Global Slot Coordination (Multi-Process)
+    // ============================================================
+
+    params.slot_rotation_interval_ms =
+        conf->get("transports/rdma/slot_rotation_interval_ms", 2);
+
+    // ============================================================
+    // Bandwidth Constants (Gbps)
+    // ============================================================
+
+    params.default_bandwidth_gbps =
+        conf->get("transports/rdma/default_bandwidth_gbps", 400.0);
+    params.min_bandwidth_gbps =
+        conf->get("transports/rdma/min_bandwidth_gbps", 10.0);
+    params.max_bandwidth_gbps =
+        conf->get("transports/rdma/max_bandwidth_gbps", 800.0);
+
+    device_selector_->setSchedulingParams(params);
+
+    // ============================================================
+    // Shared Memory Configuration
+    // ============================================================
+
    auto shared_quota_shm_path =
        conf->get("transports/rdma/shared_quota_shm_path", "");
    if (!shared_quota_shm_path.empty())
-        device_quota_->enableSharedQuota(shared_quota_shm_path);
-    auto cross_numa_access =
-        conf->get("transports/rdma/cross_numa_access", false);
-    device_quota_->setCrossNumaAccess(cross_numa_access);
-    auto local_weight = conf->get("transports/rdma/local_weight", 1.0);
-    device_quota_->setLocalWeight(local_weight);
-    auto learning_rate = conf->get("transports/rdma/learning_rate", 0.1);
-    device_quota_->setLearningRate(learning_rate);
-    auto diffusion_interval =
-        conf->get("transports/rdma/diffusion_interval", 10);
-    device_quota_->setDiffusionInterval(diffusion_interval);
+        device_selector_->enableSharedQuota(shared_quota_shm_path);
 }

 Workers::~Workers() {
@ -118,7 +203,14 @@ Status Workers::submit(RdmaSliceList& slice_list, int worker_id) {
        }
    }
    auto& worker = worker_context_[worker_id];
-    worker.queue.push(slice_list);
+
+    // Get priority from first slice (all slices in list have same priority)
+    int priority = PRIO_HIGH;
+    if (slice_list.first && slice_list.first->task) {
+        priority = slice_list.first->priority;
+    }
+
+    worker.queues[priority].push(slice_list);
    if (!worker.inflight_slices.fetch_add(slice_list.num_slices)) {
        std::lock_guard<std::mutex> lock(worker.mutex);
        if (worker.in_suspend) worker.cv.notify_all();
@ -212,7 +304,20 @@ void Workers::disableEndpoint(RdmaSlice* slice) {
 void Workers::asyncPostSend() {
    auto& worker = worker_context_[tl_wid];
    std::vector<RdmaSliceList> result;
-    worker.queue.pop(result);
+
+    auto shared_quota =
+        device_selector_ ? device_selector_->getSharedSlotManager() : nullptr;
+
+    // Promote timed-out low priority requests
+    promoteTimedOutRequests(worker);
+
+    // Priority selection: HIGH -> MEDIUM -> LOW
+    for (int prio = PRIO_HIGH; prio < kNumPriorityLevels; ++prio) {
+        if (shared_quota && !shared_quota->canSend(prio)) continue;
+        worker.queues[prio].pop(result);
+        if (!result.empty()) break;
+    }
+
    for (auto& slice_list : result) {
        if (slice_list.num_slices == 0) continue;
        auto slice = slice_list.first;
@ -283,6 +388,49 @@ void Workers::asyncPostSend() {
    }
 }

+void Workers::promoteTimedOutRequests(WorkerContext& worker) {
+    uint64_t current_ts = getCurrentTimeInNano();
+    if (current_ts < worker.next_promotion_check_ns) return;
+
+    // Set next check time (1ms from now)
+    worker.next_promotion_check_ns = current_ts + 1000000ull;
+
+    // Check MEDIUM -> HIGH promotion
+    std::vector<RdmaSliceList> promoted;
+    worker.queues[PRIO_MEDIUM].pop(promoted);
+    if (!promoted.empty()) {
+        auto* slice = promoted.front().first;
+        if (slice && slice->enqueue_ts > 0 &&
+            (current_ts - slice->enqueue_ts) >=
+                priority_promotion_timeout_ns_) {
+            for (auto& slice_list : promoted) {
+                worker.queues[PRIO_HIGH].push(slice_list);
+            }
+            return;
+        }
+        for (auto& slice_list : promoted) {
+            worker.queues[PRIO_MEDIUM].push(slice_list);
+        }
+    }
+
+    // Check LOW -> MEDIUM promotion
+    worker.queues[PRIO_LOW].pop(promoted);
+    if (!promoted.empty()) {
+        auto* slice = promoted.front().first;
+        if (slice && slice->enqueue_ts > 0 &&
+            (current_ts - slice->enqueue_ts) >=
+                priority_promotion_timeout_ns_) {
+            for (auto& slice_list : promoted) {
+                worker.queues[PRIO_MEDIUM].push(slice_list);
+            }
+            return;
+        }
+        for (auto& slice_list : promoted) {
+            worker.queues[PRIO_LOW].push(slice_list);
+        }
+    }
+}
+
 void Workers::asyncPollCq() {
    auto& worker = worker_context_[tl_wid];
    const static size_t kPollCount = 64;
@ -328,8 +476,8 @@ void Workers::asyncPollCq() {
            double inflight_lat = (poll_ts - slice->submit_ts) / 1000.0;
            double overall_lat_sec = (poll_ts - slice->enqueue_ts) / 1e9;
            if (slice->retry_count == 0) {
-                device_quota_->release(slice->source_dev_id, slice->length,
-                                       overall_lat_sec);
+                device_selector_->release(slice->source_dev_id, slice->length,
+                                          overall_lat_sec);
            }
            if (slice->word != PENDING) continue;
            if (!ep) {
@ -551,7 +699,7 @@ Status Workers::selectOptimalDevice(RouteHint& source, RouteHint& target,
                                    RdmaSlice* slice) {
    auto& worker = worker_context_[tl_wid];
    if (slice->source_dev_id < 0) {
-        CHECK_STATUS(device_quota_->allocate(
+        CHECK_STATUS(device_selector_->allocate(
            slice->length, source.buffer->location, slice->source_dev_id));
    }