Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions docs/source/design/tent/tebench.md
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,9 @@ gpu_id + thread_id
**Transport (TENT only)**

* `--xport_type` : `rdma | shm | mnnvl | gds | iouring`
* `--tent_intent_type` : attach a standard transfer intent to every request,
such as `foreground_get`, `background_prefetch`, or `checkpoint`. This is
useful for validating intent-specific transport and QoS policy selection.

**Metadata service**

Expand Down
41 changes: 40 additions & 1 deletion docs/source/design/tent/transport-selector.md
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,9 @@ Transport selection is driven by configuration with pattern-based rules.
{
"policy": [
{
"name": "high_prio_fast",
"name": "foreground_get",
"segment_type": "memory",
"intent_type": "foreground_get",
"priority": "high",
"devices": ["mlx5_0", "mlx5_1", "mlx5_2"],
"transports": ["nvlink", "rdma", "shm"]
Expand Down Expand Up @@ -52,9 +53,46 @@ Transport selection is driven by configuration with pattern-based rules.
| `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) |
| `intent_type` | string or int | No | Match a standard transfer intent such as `"foreground_get"`, `"background_prefetch"`, `"migration"`, `"checkpoint"`, `"weight_loading"`, or `"staging_internal"` |
| `devices` | array[string] | No | List of allowed device names (empty = all devices) |
| `transports` | array[string] | No | Transport preference list (evaluated in order) |

### Intent-Based Policy Binding

`Request::intent_type` can select an intent-specific policy before transport,
device, QP-pool, and SL/TC resolution:

```json
{
"policy": [
{
"name": "foreground-kv",
"segment_type": "memory",
"intent_type": "foreground_get",
"qp_pool": "foreground",
"service_level": 3,
"traffic_class": 96,
"transports": ["rdma"]
},
{
"name": "memory-fallback",
"segment_type": "memory",
"transports": ["rdma", "tcp"]
}
]
}
```

Policies are evaluated in JSON order, so intent-specific entries should appear
before a catch-all entry. A policy without `intent_type` retains the historical
behavior and matches any intent. `INTENT_UNSPEC` therefore behaves exactly as
before with existing configurations.

An explicit `Request::policy_name` remains the strongest per-request override:
it selects the named policy by segment type and bypasses the policy's other
match filters, including `intent_type`. Invalid intent values cause that policy
entry to be skipped rather than silently converted into a catch-all rule.

### Memory Type Filters

For `memory` segments, you can filter by source/destination memory type:
Expand Down Expand Up @@ -228,6 +266,7 @@ TransportSelector.select(context, transports, transport_index)
Match policy by:
- segment_type (file/memory)
- intent_type (exact match if specified in policy)
- priority (exact match if specified in policy)
- location constraints
- size constraints
Expand Down
19 changes: 19 additions & 0 deletions mooncake-transfer-engine/benchmark/tent_backend.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,23 @@ static TransportType getTransportType(const std::string& xport_type) {
return UNSPEC;
}

static IntentType getIntentType(const std::string& intent_type) {
static const std::unordered_map<std::string, IntentType> kIntentTypes = {
{"unspec", IntentType::INTENT_UNSPEC},
{"intent_unspec", IntentType::INTENT_UNSPEC},
{"foreground_get", IntentType::FOREGROUND_GET},
{"background_prefetch", IntentType::BACKGROUND_PREFETCH},
{"migration", IntentType::MIGRATION},
{"checkpoint", IntentType::CHECKPOINT},
{"weight_loading", IntentType::WEIGHT_LOADING},
{"staging_internal", IntentType::STAGING_INTERNAL},
};
auto it = kIntentTypes.find(intent_type);
LOG_ASSERT(it != kIntentTypes.end())
<< "Invalid --tent_intent_type=" << intent_type;
return it->second;
}

int TENTBenchRunner::allocateBuffers() {
const auto total_buffer_size = XferBenchConfig::total_buffer_size;
const auto& seg_type = XferBenchConfig::seg_type;
Expand Down Expand Up @@ -201,6 +218,7 @@ TENTBenchRunner::TENTBenchRunner() {
engine_ = std::make_unique<TransferEngine>(loadConfig());
transport_hint_ = TransportSelector::parseTransportType(
XferBenchConfig::tent_transport_hint);
intent_type_ = getIntentType(XferBenchConfig::tent_intent_type);
allocateBuffers();
}

Expand Down Expand Up @@ -348,6 +366,7 @@ double TENTBenchRunner::runSingleTransfer(uint64_t local_addr,
entry.target_id = handle_;
entry.target_offset = target_addr + block_size * i;
entry.transport_hint = transport_hint_;
entry.intent_type = intent_type_;
requests.emplace_back(entry);
}
XferBenchTimer timer;
Expand Down
3 changes: 2 additions & 1 deletion mooncake-transfer-engine/benchmark/tent_backend.h
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@ class TENTBenchRunner : public BenchRunner {
SegmentID handle_;
SegmentInfo info_;
TransportType transport_hint_{UNSPEC};
IntentType intent_type_{IntentType::INTENT_UNSPEC};

std::vector<std::function<int(int)>> current_task_;
std::vector<std::thread> threads_;
Expand All @@ -99,4 +100,4 @@ class TENTBenchRunner : public BenchRunner {
} // namespace tent
} // namespace mooncake

#endif // TEV1_BACKEND_H
#endif // TEV1_BACKEND_H
6 changes: 6 additions & 0 deletions mooncake-transfer-engine/benchmark/utils.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,10 @@ DEFINE_string(
tent_transport_hint, "unspec",
"tent only: per-request transport_hint. "
"unspec|rdma|tcp|shm|nvlink|gds|io_uring|mnnvl|ascend|sunrise_link");
DEFINE_string(tent_intent_type, "unspec",
"tent only: intent_type attached to every benchmark request. "
"unspec|foreground_get|background_prefetch|migration|checkpoint|"
"weight_loading|staging_internal");

namespace mooncake {
namespace tent {
Expand All @@ -78,6 +82,7 @@ std::string XferBenchConfig::xport_type;
std::string XferBenchConfig::backend;
bool XferBenchConfig::notifi = false;
std::string XferBenchConfig::tent_transport_hint;
std::string XferBenchConfig::tent_intent_type;

int XferBenchConfig::local_gpu_id = 0;
int XferBenchConfig::target_gpu_id = 0;
Expand Down Expand Up @@ -106,6 +111,7 @@ void XferBenchConfig::loadFromFlags() {
backend = FLAGS_backend;
notifi = FLAGS_notifi;
tent_transport_hint = FLAGS_tent_transport_hint;
tent_intent_type = FLAGS_tent_intent_type;

local_gpu_id = FLAGS_local_gpu_id;
target_gpu_id = FLAGS_target_gpu_id;
Expand Down
1 change: 1 addition & 0 deletions mooncake-transfer-engine/benchmark/utils.h
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,7 @@ struct XferBenchConfig {
static std::string backend;
static bool notifi;
static std::string tent_transport_hint;
static std::string tent_intent_type;

static int local_gpu_id;
static int target_gpu_id;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,8 @@ struct SelectionContext {
int priority_level; // Request priority level (lower = more urgent)
std::optional<std::string>
policy_name; // Optional: bind to specific policy by name
IntentType intent_type{
IntentType::INTENT_UNSPEC}; // Business intent for policy matching
};

/**
Expand Down Expand Up @@ -125,6 +127,10 @@ struct SelectionPolicy {
// Named QP pool this policy's traffic should land on; parsed and stored for
// now, routing to be wired later. Unset = the current single "data QP".
std::optional<std::string> qp_pool;

// Optional business-intent filter. nullopt preserves the historical
// catch-all behavior; otherwise the request intent must match exactly.
std::optional<IntentType> intent_type;
};

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1016,6 +1016,7 @@ SelectionResult TransferEngineImpl::getTransportType(const Request& request,
ctx.priority_level =
request.priority; // Use request priority for selection
ctx.policy_name = request.policy_name; // Optional: bind to specific policy
ctx.intent_type = request.intent_type; // Business intent policy filter

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

While ctx.intent_type is correctly populated here, there is a critical gap in mergeRequests (and its sorting comparator) where intent_type is completely ignored.

If two requests with different intent_type values are submitted in the same batch, they can be merged into a single request. When merged, the second request's intent_type is lost, and its transfer will be executed under the first request's intent_type policy (including its QP pool, service level, traffic class, etc.).

To fix this, intent_type (and ideally policy_name as well) must be added to the sorting comparator and the can_merge lambda in mergeRequests to prevent merging requests with different intents.


if (desc->type == SegmentType::File) {
// File segment: use selector with empty buffer_transports
Expand Down
97 changes: 82 additions & 15 deletions mooncake-transfer-engine/tent/src/runtime/transport_selector.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,10 @@
#include "tent/runtime/platform.h"
#include "tent/thirdparty/nlohmann/json.h"

#include <algorithm>
#include <glog/logging.h>

#include <algorithm>
#include <cctype>
#include <cstdint>
#include <glog/logging.h>

namespace mooncake {
namespace tent {
Expand Down Expand Up @@ -63,6 +62,46 @@ static const std::string kMemoryTypeCuda = "cuda";
static const std::string kMemoryTypeNpu = "npu";
static const std::string kMemoryTypeWildcard = "*";

static const std::unordered_map<std::string, IntentType> kIntentTypeNameMap = {
{"intent_unspec", IntentType::INTENT_UNSPEC},
{"unspec", IntentType::INTENT_UNSPEC},
{"foreground_get", IntentType::FOREGROUND_GET},
{"background_prefetch", IntentType::BACKGROUND_PREFETCH},
{"migration", IntentType::MIGRATION},
{"checkpoint", IntentType::CHECKPOINT},
{"weight_loading", IntentType::WEIGHT_LOADING},
{"staging_internal", IntentType::STAGING_INTERNAL},
};

static std::optional<IntentType> parseIntentType(const json& value) {
if (value.is_string()) {
auto name = value.get<std::string>();
std::transform(name.begin(), name.end(), name.begin(),
[](unsigned char c) { return std::tolower(c); });
auto it = kIntentTypeNameMap.find(name);
if (it != kIntentTypeNameMap.end()) return it->second;
return std::nullopt;
}

if (value.is_number_unsigned()) {
const auto raw = value.get<uint64_t>();
if (raw <= static_cast<uint64_t>(IntentType::STAGING_INTERNAL)) {
return static_cast<IntentType>(raw);
}
return std::nullopt;
}

if (value.is_number_integer()) {
const auto raw = value.get<int64_t>();
if (raw >= static_cast<int64_t>(IntentType::INTENT_UNSPEC) &&
raw <= static_cast<int64_t>(IntentType::STAGING_INTERNAL)) {
return static_cast<IntentType>(raw);
}
}
Comment on lines +86 to +100

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The parsing logic for unsigned and signed integers can be simplified and made more consistent by using an else if structure and removing the redundant return std::nullopt; inside the first block.

    if (value.is_number_unsigned()) {
        const auto raw = value.get<uint64_t>();
        if (raw <= static_cast<uint64_t>(IntentType::STAGING_INTERNAL)) {
            return static_cast<IntentType>(raw);
        }
    } else if (value.is_number_integer()) {
        const auto raw = value.get<int64_t>();
        if (raw >= static_cast<int64_t>(IntentType::INTENT_UNSPEC) &&
            raw <= static_cast<int64_t>(IntentType::STAGING_INTERNAL)) {
            return static_cast<IntentType>(raw);
        }
    }


return std::nullopt;
}

std::string TransportSelector::transportTypeName(TransportType type) {
auto it = kTransportTypeNames.find(type);
if (it != kTransportTypeNames.end()) {
Expand All @@ -86,26 +125,34 @@ std::vector<SelectionPolicy> TransportSelector::getDefaultPolicies() {
{
"file_storage",
SegmentType::File,
std::nullopt, // same_machine doesn't matter for file
std::nullopt, // local_memory_pattern
std::nullopt, // remote_memory_pattern
std::nullopt, // min_size
std::nullopt, // max_size
std::nullopt, // priority
{}, // devices (empty = all devices)
{GDS, IOURING} // File segment priority (original: GDS -> IOURING)
std::nullopt, // same_machine doesn't matter for file
std::nullopt, // local_memory_pattern
std::nullopt, // remote_memory_pattern
std::nullopt, // min_size
std::nullopt, // max_size
std::nullopt, // priority
{}, // devices (empty = all devices)
{GDS, IOURING}, // File priority (original: GDS -> IOURING)
std::nullopt, // service_level
std::nullopt, // traffic_class
std::nullopt, // qp_pool
std::nullopt // intent_type
},
{
"memory_default",
SegmentType::Memory,
std::nullopt, // any machine
std::nullopt, // any local memory
std::nullopt, // any remote memory
std::nullopt, // any size
std::nullopt, // min_priority
std::nullopt, // min_size
std::nullopt, // max_size
std::nullopt, // priority
{}, // devices (empty = all devices)
{} // Empty priority = use buffer_transports order (original
// behavior)
{}, // Empty = use buffer_transports order
std::nullopt, // service_level
std::nullopt, // traffic_class
std::nullopt, // qp_pool
std::nullopt // intent_type
},
};
}
Expand Down Expand Up @@ -196,6 +243,19 @@ void TransportSelector::loadPolicies() {
policy.priority = std::nullopt;
}

// Parse the optional business-intent filter. An invalid value skips the
// entire policy instead of turning it into a catch-all rule, which
// would silently broaden its authorization scope.
if (policy_json.contains("intent_type")) {
auto intent = parseIntentType(policy_json["intent_type"]);
if (!intent.has_value()) {
LOG(WARNING)
<< "Skip policy " << policy.name << ": invalid intent_type";
continue;
}
policy.intent_type = *intent;
}

// Parse devices (optional)
if (policy_json.contains("devices")) {
for (const auto& device_name : policy_json["devices"]) {
Expand Down Expand Up @@ -351,6 +411,13 @@ bool TransportSelector::matchesPolicy(const SelectionPolicy& policy,
}
}

// Policies without an intent filter retain the historical catch-all
// behavior. Intent-specific policies require an exact match.
if (policy.intent_type.has_value() &&
context.intent_type != policy.intent_type.value()) {
return false;
}

return true;
}

Expand Down
Loading
Loading