diff --git a/src/umbp/distributed/metrics/activity_summary.cpp b/src/umbp/distributed/metrics/activity_summary.cpp index 4bc44bfe2..746b52111 100644 --- a/src/umbp/distributed/metrics/activity_summary.cpp +++ b/src/umbp/distributed/metrics/activity_summary.cpp @@ -133,6 +133,8 @@ std::vector ActivitySummary::Render(const ActivitySnapshot& now, Delta(now.transitions.offloaded_bytes, before.transitions.offloaded_bytes); const uint64_t moves_ok = Delta(now.transitions.succeeded, before.transitions.succeeded); const uint64_t moves_failed = Delta(now.transitions.failed, before.transitions.failed); + const uint64_t moves_retried = Delta(now.transitions.retried, before.transitions.retried); + const uint64_t moves_dropped = Delta(now.transitions.dropped, before.transitions.dropped); const ReCacheActivity& rc_now = now.recache; const ReCacheActivity& rc_old = before.recache; @@ -157,8 +159,9 @@ std::vector ActivitySummary::Render(const ActivitySnapshot& now, std::snprintf(window.data(), window.size(), "%.1fs", window_seconds); const bool anything = Moved(total) || promoted != 0 || demoted != 0 || moves_failed != 0 || - rc_admitted != 0 || rc_dropped != 0 || rc_prefetched != 0 || - rc_fragmented != 0 || rc_already != 0; + moves_retried != 0 || moves_dropped != 0 || rc_admitted != 0 || + rc_dropped != 0 || rc_prefetched != 0 || rc_fragmented != 0 || + rc_already != 0; std::vector lines; @@ -205,11 +208,14 @@ std::vector ActivitySummary::Render(const ActivitySnapshot& now, // they are printed as a pair even when one of them is zero: "promote 0 B" // next to a large demote is the shape of a tier graph that only sheds. head += " · promote " + FormatBytes(promoted) + " · demote " + FormatBytes(demoted); - if (moves_ok != 0 || moves_failed != 0) { - std::array moves{}; - std::snprintf(moves.data(), moves.size(), " (%llu moved, %llu failed)", - static_cast(moves_ok), - static_cast(moves_failed)); + if (moves_ok != 0 || moves_failed != 0 || moves_retried != 0 || moves_dropped != 0) { + // dropped = evicted copies lost for good; retried = demotions re-queued. + std::array moves{}; + std::snprintf( + moves.data(), moves.size(), " (%llu moved, %llu failed, %llu retried, %llu dropped)", + static_cast(moves_ok), static_cast(moves_failed), + static_cast(moves_retried), + static_cast(moves_dropped)); head += moves.data(); } lines.push_back(std::move(head)); diff --git a/src/umbp/distributed/pool/peer_pool.cpp b/src/umbp/distributed/pool/peer_pool.cpp index 9127f5ef6..b8eb8ec98 100644 --- a/src/umbp/distributed/pool/peer_pool.cpp +++ b/src/umbp/distributed/pool/peer_pool.cpp @@ -34,9 +34,23 @@ #include #include +#include "umbp/common/env_time.h" + namespace mori::umbp { + namespace { +// How long a retryable eviction demotion keeps retrying before its copy is +// dropped. Time, not a count: the faults it rides out last a wall-clock +// interval, and a short queue would burn a count in milliseconds. +std::chrono::milliseconds EvictDemotionRetryBudget() { + static const auto v = GetEnvMilliseconds("UMBP_EVICT_DEMOTION_RETRY_MS", + std::chrono::milliseconds(30000), /*min_allowed=*/0); + return v; +} +// Pause before a lone retry, so waiting out pressure does not spin the worker. +constexpr std::chrono::milliseconds kEvictRetryIdlePause{20}; + // One watermark scan queues at most this many keys; the worker tops the queue // up after each completed offload, so a tier still drains all the way down to // its low watermark without ever queueing its whole key set at once. @@ -90,6 +104,7 @@ PeerPool::PeerPool(BackendRegistry* backends, std::unique_ptr policy break; } } + evict_demotion_retry_budget_ = EvictDemotionRetryBudget(); transition_worker_ = std::thread(&PeerPool::TransitionWorkerLoop, this); } } @@ -117,7 +132,8 @@ void PeerPool::DropEvictedCopyLocked(const TransitionJob& job) { if (migrating_keys_.count(job.key) != 0) return; auto* source = backends_ == nullptr ? nullptr : backends_->Get(job.source_backend_id); if (source == nullptr) return; - source->Evict({job.key}); + const auto evicted = source->Evict({job.key}); + if (!evicted.empty() && evicted.front().bytes_freed > 0) ++transition_metrics_.dropped; RepairAfterDropLocked(job.key); } @@ -324,17 +340,27 @@ void PeerPool::TransitionWorkerLoop() { transition_queue_.pop_front(); } bool retry = false; + bool evict_retry = false; { std::unique_lock lock(operation_mutex_); if (tier_graph_ != nullptr) { const auto& tier = tier_graph_->NodeAt(job.source_tier); const bool watermark_driven = WatermarkDrivenLocked(job); if (!watermark_driven || tier_graph_->AtOrAbove(job.source_tier, tier.low_watermark)) { + if (job.evict && job.first_attempt == std::chrono::steady_clock::time_point{}) { + job.first_attempt = std::chrono::steady_clock::now(); + } const auto result = RunTransition(lock, job); - if (job.evict && !result.success) DropEvictedCopyLocked(job); + // The evicted copy is the only one: retry what can clear, and drop it + // only once the budget is spent or the target can never take it. + evict_retry = + job.evict && !result.success && result.retryable && + std::chrono::steady_clock::now() - job.first_attempt < evict_demotion_retry_budget_; + if (evict_retry) ++transition_metrics_.retried; + if (job.evict && !result.success && !evict_retry) DropEvictedCopyLocked(job); const bool still_pressured = watermark_driven && tier_graph_->AtOrAbove(job.source_tier, tier.low_watermark); - retry = still_pressured && !result.success && job.attempts < 3; + retry = (still_pressured && !result.success && job.attempts < 3) || evict_retry; // Keep the drain going without waiting for the next commit batch to // notice the tier is still above its low watermark. The scan skips // whatever is still queued, so this tops the batch back up rather @@ -346,7 +372,14 @@ void PeerPool::TransitionWorkerLoop() { } } } - if (retry) { + if (evict_retry) { + bool idle = false; + { + std::lock_guard lock(transition_mutex_); + idle = transition_queue_.empty(); + } + if (idle) std::this_thread::sleep_for(kEvictRetryIdlePause); + } else if (retry) { std::this_thread::sleep_for(std::chrono::milliseconds(uint64_t{1} << job.attempts)); } { @@ -354,7 +387,7 @@ void PeerPool::TransitionWorkerLoop() { queued_transition_keys_.erase(job.key); const size_t source_tier = job.source_tier; if (retry && !stop_transition_worker_) { - ++job.attempts; + if (job.attempts < UINT8_MAX) ++job.attempts; queued_transition_keys_.insert(job.key); transition_queue_.push_back(std::move(job)); transition_cv_.notify_one(); diff --git a/src/umbp/distributed/pool/tier_transition.cpp b/src/umbp/distributed/pool/tier_transition.cpp index 9350de8cf..cf926160b 100644 --- a/src/umbp/distributed/pool/tier_transition.cpp +++ b/src/umbp/distributed/pool/tier_transition.cpp @@ -99,14 +99,19 @@ TierTransitionResult TierTransitionExecutor::Execute( if (!BuildTransfers(source, resolved, target, allocation, &transfers) || !transfer_engine_->Transfer(transfers, nullptr)) { target->BatchAbort({allocation.slot_id}); + result.retryable = true; continue; } auto committed = target->BatchCommit({CommitRequest{allocation.slot_id, key}}); if (committed.empty() || !committed.front().success) { target->BatchAbort({allocation.slot_id}); + result.retryable = true; continue; } } else if (allocation.outcome != AllocateOutcome::kSuccessAlreadyExists) { + // No space is pressure (staging full of reads, eviction catching up); + // a shape failure (kFailed) stays final. + if (allocation.outcome == AllocateOutcome::kFailedNoSpace) result.retryable = true; continue; } @@ -114,10 +119,12 @@ TierTransitionResult TierTransitionExecutor::Execute( if (!target->AcquireMigrationRead(key, &target_read) || target_read.size != resolved.size) { if (target_read.found) target->ReleaseMigrationRead(key); if (allocation.outcome == AllocateOutcome::kSuccessAllocated) target->Evict({key}); + result.retryable = true; continue; } result.success = true; + result.retryable = false; result.target_backend_id = target_backend_id; result.bytes_moved = resolved.size; source->ReleaseMigrationRead(key); diff --git a/src/umbp/distributed/pool_client.cpp b/src/umbp/distributed/pool_client.cpp index 2f150f263..5605fb010 100644 --- a/src/umbp/distributed/pool_client.cpp +++ b/src/umbp/distributed/pool_client.cpp @@ -5171,6 +5171,10 @@ void PoolClient::PublishComponentMetrics() { {{"outcome", "succeeded"}}, tiers.succeeded); sample(MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS, MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS_HELP, {{"outcome", "failed"}}, tiers.failed); + sample(MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS, MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS_HELP, + {{"outcome", "dropped"}}, tiers.dropped); + sample(MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS, MORI_UMBP_METRIC_CLIENT_TIER_TRANSITIONS_HELP, + {{"outcome", "retried"}}, tiers.retried); sample(MORI_UMBP_METRIC_CLIENT_TIER_OFFLOADED_BYTES, MORI_UMBP_METRIC_CLIENT_TIER_OFFLOADED_BYTES_HELP, {}, tiers.offloaded_bytes); sample(MORI_UMBP_METRIC_CLIENT_TIER_PROMOTED_BYTES, diff --git a/src/umbp/doc/runtime-env-vars.md b/src/umbp/doc/runtime-env-vars.md index ad2ed0d17..0cdc58c09 100644 --- a/src/umbp/doc/runtime-env-vars.md +++ b/src/umbp/doc/runtime-env-vars.md @@ -61,6 +61,7 @@ Read by the **master process** (`bin/master_main.cpp` via | `UMBP_HEARTBEAT_INTERVAL_DIVISOR` | `2` | count | Recommended client heartbeat interval = `heartbeat_ttl / divisor`. `min_allowed=1` guards against div-by-zero. Read by the master and echoed in `RegisterClientResponse.heartbeat_interval_ms`. | | `UMBP_EVICTKEY_DEADLINE_MS` | `1000` | ms | Per-call gRPC deadline applied to outbound `EvictKey` RPCs from `MasterPeerStubPool`. | | `UMBP_EVICT_MAX_BYTES_PER_ROUND` | `0` | bytes | Most bytes one eviction round frees from one `(node, tier)`. `0` = its whole overage above the low watermark. Set it to throttle eviction, e.g. how much a demoting tier copies downstream per round. | +| `UMBP_EVICT_DEMOTION_RETRY_MS` | `30000` | ms | How long a peer retries an eviction demotion that failed in a way that can clear (no space at the target, e.g. an SSD staging arena full of reads, or a copy/commit error) before it drops the evicted copy so the tier still frees. Counted from the first attempt. `0` = drop on the first such failure. | | `UMBP_HIT_INDEX_TTL_SEC` | `7200` | sec | External KV hit-count entry TTL. A hash with no counted match for longer than this is removed from the hit index. | | `UMBP_HIT_INDEX_GC_INTERVAL_SEC` | `60` | sec | External KV hit-count GC sweep interval. | | `UMBP_HIT_QUERY_MAX_BATCH` | `4096` | count | Maximum hashes accepted by one `GetExternalKvHitCounts` request. Oversized requests return gRPC `INVALID_ARGUMENT`; the server does not truncate. | diff --git a/src/umbp/include/umbp/distributed/pool/peer_pool.h b/src/umbp/include/umbp/distributed/pool/peer_pool.h index 689082484..15161ce99 100644 --- a/src/umbp/include/umbp/distributed/pool/peer_pool.h +++ b/src/umbp/include/umbp/distributed/pool/peer_pool.h @@ -156,6 +156,10 @@ class PeerPool { LocalEvictionMetrics LocalEvictionStats() const; // Test seam: one synchronous pass. Returns keys dropped or queued. size_t RunLocalEvictionOnceForTest() { return LocalEvictionPass(); } + // UMBP_EVICT_DEMOTION_RETRY_MS override; set it before queueing work. + void SetEvictDemotionRetryBudgetForTest(std::chrono::milliseconds budget) { + evict_demotion_retry_budget_ = budget; + } // Victims for a backend over its watermark, coldest first, before the // medium's own order: (tier, logical tier, bytes) -> offers, or nullopt when @@ -177,7 +181,9 @@ class PeerPool { uint32_t source_backend_id = BackendRegistry::kMaxBackends; LogicalTierGraph::TierIndex source_tier = 0; uint8_t attempts = 0; - bool evict = false; // from Evict(): ignores watermarks; drops on failure + // An eviction demotion's retries are bounded by time from its first attempt. + std::chrono::steady_clock::time_point first_attempt{}; + bool evict = false; // from Evict(): ignores watermarks; drops once retries run out }; // What the copy phase needs, resolved while the pool lock is still held. @@ -296,6 +302,7 @@ class PeerPool { std::vector queued_by_tier_; bool stop_transition_worker_ = false; std::thread transition_worker_; + std::chrono::milliseconds evict_demotion_retry_budget_{30000}; // transition worker only // Local watermark eviction. local_evict_mutex_ guards local_evict_, // stop_local_evict_ and the worker's start. diff --git a/src/umbp/include/umbp/distributed/pool/tier_transition.h b/src/umbp/include/umbp/distributed/pool/tier_transition.h index dd36e0de9..93e2ccfde 100644 --- a/src/umbp/include/umbp/distributed/pool/tier_transition.h +++ b/src/umbp/include/umbp/distributed/pool/tier_transition.h @@ -43,6 +43,9 @@ struct TierTransitionResult { uint64_t bytes_moved = 0; uint64_t bytes_freed = 0; bool source_draining = false; + // Failed in a way that can clear (no space, or a copy/commit/read-back + // error), so a later attempt may succeed. False for a shape failure. + bool retryable = false; }; // Executes one peer-local copy or move. Placement policy and scheduling remain diff --git a/src/umbp/include/umbp/distributed/types.h b/src/umbp/include/umbp/distributed/types.h index bc3e95b44..b76991b2c 100644 --- a/src/umbp/include/umbp/distributed/types.h +++ b/src/umbp/include/umbp/distributed/types.h @@ -80,6 +80,8 @@ struct TierTransitionMetrics { uint64_t attempted = 0; uint64_t succeeded = 0; uint64_t failed = 0; + uint64_t dropped = 0; // evicted copies whose demotion never landed (lost keys) + uint64_t retried = 0; // eviction demotions re-queued after a retryable failure uint64_t offloaded_bytes = 0; uint64_t promoted_bytes = 0; }; diff --git a/tests/cpp/umbp/distributed/test_activity_summary.cpp b/tests/cpp/umbp/distributed/test_activity_summary.cpp index 2d701a189..420759c6d 100644 --- a/tests/cpp/umbp/distributed/test_activity_summary.cpp +++ b/tests/cpp/umbp/distributed/test_activity_summary.cpp @@ -135,10 +135,12 @@ TEST(ActivitySummary, ReportsPromotionAndDemotionAsAPairEvenWhenOneIsZero) { second.transitions.offloaded_bytes = 8ULL * 1024 * 1024; second.transitions.succeeded = 4; second.transitions.failed = 1; + second.transitions.retried = 3; + second.transitions.dropped = 2; const std::string all = Joined(summary.Render(second, 60.0)); EXPECT_NE(all.find("promote 0 B"), std::string::npos) << all; EXPECT_NE(all.find("demote 8.00 MiB"), std::string::npos) << all; - EXPECT_NE(all.find("1 failed"), std::string::npos) << all; + EXPECT_NE(all.find("1 failed, 3 retried, 2 dropped"), std::string::npos) << all; } TEST(ActivitySummary, SurfacesLoadMissesWhichBytesAloneHide) { diff --git a/tests/cpp/umbp/distributed/test_peer_pool.cpp b/tests/cpp/umbp/distributed/test_peer_pool.cpp index 8882d0fce..04f670015 100644 --- a/tests/cpp/umbp/distributed/test_peer_pool.cpp +++ b/tests/cpp/umbp/distributed/test_peer_pool.cpp @@ -176,6 +176,37 @@ class NoSpaceBackend final : public MockBackend { } }; +// LocalCopyEngine whose next `failures` copies fail. +class FailingEngine final : public TransferEngine { + public: + explicit FailingEngine(int failures) : failures_(failures) {} + void FailNext(int n) { failures_.store(n); } + + const char* Name() const override { return "failing"; } + TransferRef RegisterMemory(void* base, size_t size, mori::io::MemoryLocationType loc, + int device) override { + return inner_.RegisterMemory(base, size, loc, device); + } + void Deregister(const TransferRef& ref) override { inner_.Deregister(ref); } + bool CanHandle(const TransferRef& src, const TransferRef& dst) const override { + return inner_.CanHandle(src, dst); + } + TransferPlanSet Plan(const std::vector& items) const override { + return inner_.Plan(items); + } + std::unique_ptr Submit(std::vector plans) override { + if (failures_.load() > 0) { + failures_.fetch_sub(1); + return nullptr; // the copy never happened + } + return inner_.Submit(std::move(plans)); + } + + private: + LocalCopyEngine inner_; + std::atomic failures_; +}; + class BusyResolveBackend final : public MockBackend { public: explicit BusyResolveBackend(TierType tier) : MockBackend(tier) {} @@ -810,6 +841,7 @@ TEST(PeerPool, DemotionWithNoRoomDownstreamDropsTheCopy) { auto compiled = LogicalTierGraph::Compile(HotColdTiers(PoolOffloadTrigger::kOnEvict), registry); ASSERT_TRUE(compiled.ok()) << compiled.error; PeerPool pool(®istry, MakeTieredPlacementPolicy(compiled.graph), &engine); + pool.SetEvictDemotionRetryBudgetForTest(std::chrono::milliseconds{50}); // Fill cold's only page directly. auto* cold = registry.Get("cold"); auto filler = cold->BatchAllocate({AllocateRequest{"filler", kPageSize}}).front(); @@ -824,6 +856,65 @@ TEST(PeerPool, DemotionWithNoRoomDownstreamDropsTheCopy) { EXPECT_FALSE(cold->Contains("k")); EXPECT_TRUE(cold->Contains("filler")); EXPECT_FALSE(pool.PlacementBackend("k").has_value()); + EXPECT_EQ(pool.TransitionMetrics().dropped, 1u); +} + +// A demotion copy that fails is retried, since the evicted copy is the only +// one; it is dropped only once the retry budget runs out, so the tier frees. +TEST(PeerPool, AFailedDemotionIsRetriedUntilItsBudgetRunsOut) { + constexpr uint64_t kPageSize = 4096; + FailingEngine engine(/*failures=*/0); + BackendRegistry registry; + RegisterPageBackend(®istry, &engine, "hot", TierType::DRAM, kPageSize, 4); + RegisterPageBackend(®istry, &engine, "cold", TierType::SSD, kPageSize, 4); + auto compiled = LogicalTierGraph::Compile(HotColdTiers(PoolOffloadTrigger::kOnEvict), registry); + ASSERT_TRUE(compiled.ok()) << compiled.error; + PeerPool pool(®istry, MakeTieredPlacementPolicy(compiled.graph), &engine); + CommitKey(&pool, "a"); + CommitKey(&pool, "b"); + + engine.FailNext(2); // the first two copies fail; the third lands + auto evicted = pool.Evict({PoolEvictRequest{"a", TierType::DRAM}}, PoolEvictMode::kReclaim); + EXPECT_EQ(evicted.front().outcome, PoolEvictOutcome::kDemotionQueued); + ASSERT_TRUE(WaitForKeyGone(®istry, "hot", "a")); + EXPECT_TRUE(registry.Get("cold")->Contains("a")) << "dropped, not demoted"; + EXPECT_EQ(pool.TransitionMetrics().dropped, 0u); + EXPECT_GE(pool.TransitionMetrics().failed, 2u) << "the failed attempts still count"; + EXPECT_GE(pool.TransitionMetrics().retried, 2u); + + pool.SetEvictDemotionRetryBudgetForTest(std::chrono::milliseconds{50}); + engine.FailNext(1000000); // every copy fails + pool.Evict({PoolEvictRequest{"b", TierType::DRAM}}, PoolEvictMode::kReclaim); + ASSERT_TRUE(WaitForKeyGone(®istry, "hot", "b")); + EXPECT_FALSE(registry.Get("cold")->Contains("b")); + EXPECT_FALSE(pool.PlacementBackend("b").has_value()); + EXPECT_EQ(pool.TransitionMetrics().dropped, 1u); +} + +// No space at the target (an SSD whose staging is full of reads) is waited +// out, not a reason to drop the only copy. +TEST(PeerPool, ADemotionWaitsOutATargetWithNoRoomForNow) { + constexpr uint64_t kPageSize = 4096; + LocalCopyEngine engine; + BackendRegistry registry; + RegisterPageBackend(®istry, &engine, "hot", TierType::DRAM, kPageSize, 4); + RegisterPageBackend(®istry, &engine, "cold", TierType::SSD, kPageSize, 1); + auto compiled = LogicalTierGraph::Compile(HotColdTiers(PoolOffloadTrigger::kOnEvict), registry); + ASSERT_TRUE(compiled.ok()) << compiled.error; + PeerPool pool(®istry, MakeTieredPlacementPolicy(compiled.graph), &engine); + auto* cold = registry.Get("cold"); + auto filler = cold->BatchAllocate({AllocateRequest{"filler", kPageSize}}).front(); + ASSERT_EQ(filler.outcome, AllocateOutcome::kSuccessAllocated); // cold's only page, pending + CommitKey(&pool, "k"); + + pool.Evict({PoolEvictRequest{"k", TierType::DRAM}}, PoolEvictMode::kReclaim); + std::this_thread::sleep_for(std::chrono::milliseconds{100}); + EXPECT_TRUE(registry.Get("hot")->Contains("k")) << "dropped while the target was only busy"; + + cold->BatchAbort({filler.slot_id}); // the pressure clears + ASSERT_TRUE(WaitForKeyGone(®istry, "hot", "k")); + EXPECT_TRUE(cold->Contains("k")); + EXPECT_EQ(pool.TransitionMetrics().dropped, 0u); } TEST(PeerPool, WatermarkMigratesCommittedKey) {