diff --git a/docs/source/design/tent/tebench.md b/docs/source/design/tent/tebench.md index c25745b5bd..e4dd877ad8 100644 --- a/docs/source/design/tent/tebench.md +++ b/docs/source/design/tent/tebench.md @@ -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** diff --git a/docs/source/design/tent/transport-selector.md b/docs/source/design/tent/transport-selector.md index d1a90e0877..e4ba13e2ec 100644 --- a/docs/source/design/tent/transport-selector.md +++ b/docs/source/design/tent/transport-selector.md @@ -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"] @@ -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: @@ -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 diff --git a/mooncake-transfer-engine/benchmark/tent_backend.cpp b/mooncake-transfer-engine/benchmark/tent_backend.cpp index 3aa24db0d4..030205d0a2 100644 --- a/mooncake-transfer-engine/benchmark/tent_backend.cpp +++ b/mooncake-transfer-engine/benchmark/tent_backend.cpp @@ -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 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; @@ -201,6 +218,7 @@ TENTBenchRunner::TENTBenchRunner() { engine_ = std::make_unique(loadConfig()); transport_hint_ = TransportSelector::parseTransportType( XferBenchConfig::tent_transport_hint); + intent_type_ = getIntentType(XferBenchConfig::tent_intent_type); allocateBuffers(); } @@ -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; diff --git a/mooncake-transfer-engine/benchmark/tent_backend.h b/mooncake-transfer-engine/benchmark/tent_backend.h index 1648ad9269..3f5dafcc59 100644 --- a/mooncake-transfer-engine/benchmark/tent_backend.h +++ b/mooncake-transfer-engine/benchmark/tent_backend.h @@ -87,6 +87,7 @@ class TENTBenchRunner : public BenchRunner { SegmentID handle_; SegmentInfo info_; TransportType transport_hint_{UNSPEC}; + IntentType intent_type_{IntentType::INTENT_UNSPEC}; std::vector> current_task_; std::vector threads_; @@ -99,4 +100,4 @@ class TENTBenchRunner : public BenchRunner { } // namespace tent } // namespace mooncake -#endif // TEV1_BACKEND_H \ No newline at end of file +#endif // TEV1_BACKEND_H diff --git a/mooncake-transfer-engine/benchmark/utils.cpp b/mooncake-transfer-engine/benchmark/utils.cpp index d907a2bd68..a1db881871 100644 --- a/mooncake-transfer-engine/benchmark/utils.cpp +++ b/mooncake-transfer-engine/benchmark/utils.cpp @@ -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 { @@ -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; @@ -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; diff --git a/mooncake-transfer-engine/benchmark/utils.h b/mooncake-transfer-engine/benchmark/utils.h index 6c862cd6c5..cbd254a5c7 100644 --- a/mooncake-transfer-engine/benchmark/utils.h +++ b/mooncake-transfer-engine/benchmark/utils.h @@ -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; diff --git a/mooncake-transfer-engine/tent/include/tent/runtime/transport_selector.h b/mooncake-transfer-engine/tent/include/tent/runtime/transport_selector.h index a419e58d69..99d0929f54 100644 --- a/mooncake-transfer-engine/tent/include/tent/runtime/transport_selector.h +++ b/mooncake-transfer-engine/tent/include/tent/runtime/transport_selector.h @@ -82,6 +82,8 @@ struct SelectionContext { int priority_level; // Request priority level (lower = more urgent) std::optional policy_name; // Optional: bind to specific policy by name + IntentType intent_type{ + IntentType::INTENT_UNSPEC}; // Business intent for policy matching }; /** @@ -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 qp_pool; + + // Optional business-intent filter. nullopt preserves the historical + // catch-all behavior; otherwise the request intent must match exactly. + std::optional intent_type; }; /** diff --git a/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp b/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp index 6bee9e27aa..844ecce47e 100644 --- a/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp +++ b/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp @@ -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 if (desc->type == SegmentType::File) { // File segment: use selector with empty buffer_transports diff --git a/mooncake-transfer-engine/tent/src/runtime/transport_selector.cpp b/mooncake-transfer-engine/tent/src/runtime/transport_selector.cpp index 046af38c2e..6b5616e0ad 100644 --- a/mooncake-transfer-engine/tent/src/runtime/transport_selector.cpp +++ b/mooncake-transfer-engine/tent/src/runtime/transport_selector.cpp @@ -17,11 +17,10 @@ #include "tent/runtime/platform.h" #include "tent/thirdparty/nlohmann/json.h" -#include -#include - #include #include +#include +#include namespace mooncake { namespace tent { @@ -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 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 parseIntentType(const json& value) { + if (value.is_string()) { + auto name = value.get(); + 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(); + if (raw <= static_cast(IntentType::STAGING_INTERNAL)) { + return static_cast(raw); + } + return std::nullopt; + } + + if (value.is_number_integer()) { + const auto raw = value.get(); + if (raw >= static_cast(IntentType::INTENT_UNSPEC) && + raw <= static_cast(IntentType::STAGING_INTERNAL)) { + return static_cast(raw); + } + } + + return std::nullopt; +} + std::string TransportSelector::transportTypeName(TransportType type) { auto it = kTransportTypeNames.find(type); if (it != kTransportTypeNames.end()) { @@ -86,14 +125,18 @@ std::vector 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", @@ -101,11 +144,15 @@ std::vector TransportSelector::getDefaultPolicies() { 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 }, }; } @@ -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"]) { @@ -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; } diff --git a/mooncake-transfer-engine/tent/tests/transport_selector_test.cpp b/mooncake-transfer-engine/tent/tests/transport_selector_test.cpp index 4fb0f39018..277e0567d7 100644 --- a/mooncake-transfer-engine/tent/tests/transport_selector_test.cpp +++ b/mooncake-transfer-engine/tent/tests/transport_selector_test.cpp @@ -14,6 +14,8 @@ #include +#include +#include #include #include @@ -750,6 +752,224 @@ TEST(TransportSelectorTest, PolicyQpPoolEmptyOrNonStringIsUnset) { selector.select(ctx, transports, /*index=*/0).qp_pool.has_value()); } +// Intent-specific policies bind Request::intent_type to transport and +// link-layer QoS selection. Policies are first-match, so the specific entry is +// deliberately placed before the catch-all fallback. +TEST(TransportSelectorTest, IntentSpecificPolicyIsSelected) { + auto conf = std::make_shared(); + json foreground; + foreground["name"] = "foreground"; + foreground["segment_type"] = "memory"; + foreground["intent_type"] = "foreground_get"; + foreground["transports"] = {"rdma"}; + foreground["service_level"] = 3; + foreground["traffic_class"] = 96; + foreground["qp_pool"] = "foreground"; + json fallback; + fallback["name"] = "fallback"; + fallback["segment_type"] = "memory"; + fallback["transports"] = {"tcp"}; + conf->set("policy", json::array({foreground, fallback})); + + TransportSelector selector(conf); + std::array, kSupportedTransportTypes> + transports{}; + transports[RDMA] = std::make_shared(RDMA); + transports[TCP] = std::make_shared(TCP); + static_cast(transports[RDMA].get())->setDramToDram(true); + static_cast(transports[TCP].get())->setDramToDram(true); + std::vector buffer_transports = {RDMA, TCP}; + + SelectionContext ctx; + ctx.segment_type = SegmentType::Memory; + ctx.same_machine = false; + ctx.local_memory_type = MTYPE_CPU; + ctx.remote_memory_type = MTYPE_CPU; + ctx.transfer_size = 4096; + ctx.priority_level = PRIO_HIGH; + ctx.buffer_transports = &buffer_transports; + ctx.intent_type = IntentType::FOREGROUND_GET; + + auto result = selector.select(ctx, transports); + EXPECT_EQ(result.transport, RDMA); + EXPECT_EQ(result.service_level, 3); + EXPECT_EQ(result.traffic_class, 96); + EXPECT_EQ(result.qp_pool, "foreground"); +} + +TEST(TransportSelectorTest, IntentMismatchFallsThroughToCatchAll) { + auto conf = std::make_shared(); + json foreground; + foreground["name"] = "foreground"; + foreground["segment_type"] = "memory"; + foreground["intent_type"] = "foreground_get"; + foreground["transports"] = {"rdma"}; + json fallback; + fallback["name"] = "fallback"; + fallback["segment_type"] = "memory"; + fallback["transports"] = {"tcp"}; + conf->set("policy", json::array({foreground, fallback})); + + TransportSelector selector(conf); + std::array, kSupportedTransportTypes> + transports{}; + transports[RDMA] = std::make_shared(RDMA); + transports[TCP] = std::make_shared(TCP); + static_cast(transports[RDMA].get())->setDramToDram(true); + static_cast(transports[TCP].get())->setDramToDram(true); + std::vector buffer_transports = {RDMA, TCP}; + + SelectionContext ctx; + ctx.segment_type = SegmentType::Memory; + ctx.same_machine = false; + ctx.local_memory_type = MTYPE_CPU; + ctx.remote_memory_type = MTYPE_CPU; + ctx.transfer_size = 4096; + ctx.priority_level = PRIO_LOW; + ctx.buffer_transports = &buffer_transports; + ctx.intent_type = IntentType::CHECKPOINT; + + EXPECT_EQ(selector.select(ctx, transports).transport, TCP); +} + +TEST(TransportSelectorTest, PolicyWithoutIntentMatchesAnyIntent) { + auto conf = std::make_shared(); + json policy; + policy["name"] = "legacy"; + policy["segment_type"] = "memory"; + policy["transports"] = {"rdma"}; + conf->set("policy", json::array({policy})); + + TransportSelector selector(conf); + std::array, kSupportedTransportTypes> + transports{}; + transports[RDMA] = std::make_shared(RDMA); + static_cast(transports[RDMA].get())->setDramToDram(true); + std::vector buffer_transports = {RDMA}; + + SelectionContext ctx; + ctx.segment_type = SegmentType::Memory; + ctx.same_machine = false; + ctx.local_memory_type = MTYPE_CPU; + ctx.remote_memory_type = MTYPE_CPU; + ctx.transfer_size = 4096; + ctx.priority_level = PRIO_LOW; + ctx.buffer_transports = &buffer_transports; + ctx.intent_type = IntentType::CHECKPOINT; + + EXPECT_EQ(selector.select(ctx, transports).transport, RDMA); +} + +TEST(TransportSelectorTest, NumericIntentValueIsAccepted) { + auto conf = std::make_shared(); + json policy; + policy["name"] = "checkpoint"; + policy["segment_type"] = "memory"; + policy["intent_type"] = static_cast(IntentType::CHECKPOINT); + policy["transports"] = {"tcp"}; + conf->set("policy", json::array({policy})); + + TransportSelector selector(conf); + std::array, kSupportedTransportTypes> + transports{}; + transports[TCP] = std::make_shared(TCP); + static_cast(transports[TCP].get())->setDramToDram(true); + std::vector buffer_transports = {TCP}; + + SelectionContext ctx; + ctx.segment_type = SegmentType::Memory; + ctx.same_machine = false; + ctx.local_memory_type = MTYPE_CPU; + ctx.remote_memory_type = MTYPE_CPU; + ctx.transfer_size = 4096; + ctx.priority_level = PRIO_LOW; + ctx.buffer_transports = &buffer_transports; + ctx.intent_type = IntentType::CHECKPOINT; + + EXPECT_EQ(selector.select(ctx, transports).transport, TCP); +} + +TEST(TransportSelectorTest, InvalidIntentPolicyIsSkipped) { + auto conf = std::make_shared(); + json bad_name; + bad_name["name"] = "bad-name"; + bad_name["segment_type"] = "memory"; + bad_name["intent_type"] = "not_an_intent"; + bad_name["transports"] = {"rdma"}; + json bad_number; + bad_number["name"] = "bad-number"; + bad_number["segment_type"] = "memory"; + bad_number["intent_type"] = 999; + bad_number["transports"] = {"rdma"}; + json bad_type; + bad_type["name"] = "bad-type"; + bad_type["segment_type"] = "memory"; + bad_type["intent_type"] = true; + bad_type["transports"] = {"rdma"}; + json bad_unsigned; + bad_unsigned["name"] = "bad-unsigned"; + bad_unsigned["segment_type"] = "memory"; + bad_unsigned["intent_type"] = std::numeric_limits::max(); + bad_unsigned["transports"] = {"rdma"}; + json fallback; + fallback["name"] = "fallback"; + fallback["segment_type"] = "memory"; + fallback["transports"] = {"tcp"}; + conf->set("policy", json::array({bad_name, bad_number, bad_type, + bad_unsigned, fallback})); + + TransportSelector selector(conf); + std::array, kSupportedTransportTypes> + transports{}; + transports[RDMA] = std::make_shared(RDMA); + transports[TCP] = std::make_shared(TCP); + static_cast(transports[RDMA].get())->setDramToDram(true); + static_cast(transports[TCP].get())->setDramToDram(true); + std::vector buffer_transports = {RDMA, TCP}; + + SelectionContext ctx; + ctx.segment_type = SegmentType::Memory; + ctx.same_machine = false; + ctx.local_memory_type = MTYPE_CPU; + ctx.remote_memory_type = MTYPE_CPU; + ctx.transfer_size = 4096; + ctx.priority_level = PRIO_HIGH; + ctx.buffer_transports = &buffer_transports; + ctx.intent_type = IntentType::FOREGROUND_GET; + + EXPECT_EQ(selector.select(ctx, transports).transport, TCP); +} + +TEST(TransportSelectorTest, ExplicitPolicyNameOverridesIntentFilter) { + auto conf = std::make_shared(); + json policy; + policy["name"] = "operator-override"; + policy["segment_type"] = "memory"; + policy["intent_type"] = "checkpoint"; + policy["transports"] = {"tcp"}; + conf->set("policy", json::array({policy})); + + TransportSelector selector(conf); + std::array, kSupportedTransportTypes> + transports{}; + transports[TCP] = std::make_shared(TCP); + static_cast(transports[TCP].get())->setDramToDram(true); + std::vector buffer_transports = {TCP}; + + SelectionContext ctx; + ctx.segment_type = SegmentType::Memory; + ctx.same_machine = false; + ctx.local_memory_type = MTYPE_CPU; + ctx.remote_memory_type = MTYPE_CPU; + ctx.transfer_size = 4096; + ctx.priority_level = PRIO_HIGH; + ctx.buffer_transports = &buffer_transports; + ctx.intent_type = IntentType::FOREGROUND_GET; + ctx.policy_name = "operator-override"; + + EXPECT_EQ(selector.select(ctx, transports).transport, TCP); +} + } // namespace } // namespace tent } // namespace mooncake