diff --git a/benchmark/MulticlockScaling.hpp b/benchmark/MulticlockScaling.hpp index 0f2c70a..ab5daa0 100644 --- a/benchmark/MulticlockScaling.hpp +++ b/benchmark/MulticlockScaling.hpp @@ -233,7 +233,7 @@ inline int runClockScaling(int argc, char** argv) { "allocated_ingress_bytes,peak_ingress_bytes,allocated_staging_bytes,peak_staging_" "records," "native_buffer_peak_bytes,file_bytes"; - const std::array, 16> fields{ + const std::array, 21> fields{ {{"sweeps", &sender::ClockSchedulerProfile::sweeps}, {"idle_sweeps", &sender::ClockSchedulerProfile::idle_sweeps}, {"retirement_ns", &sender::ClockSchedulerProfile::retirement_ns}, @@ -249,7 +249,12 @@ inline int runClockScaling(int argc, char** argv) { {"allowance_waits", &sender::ClockSchedulerProfile::allowance_waits}, {"dependency_waits", &sender::ClockSchedulerProfile::dependency_waits}, {"completion_loads", &sender::ClockSchedulerProfile::completion_loads}, - {"coordinator_sweeps", &sender::ClockSchedulerProfile::sweeps}}}; + {"coordinator_sweeps", &sender::ClockSchedulerProfile::sweeps}, + {"migration_plans", &sender::ClockSchedulerProfile::migration_plans}, + {"migration_planning_ns", &sender::ClockSchedulerProfile::migration_planning_ns}, + {"migration_handoff_ns", &sender::ClockSchedulerProfile::migration_handoff_ns}, + {"migration_feedback_good", &sender::ClockSchedulerProfile::migration_feedback_good}, + {"migration_feedback_bad", &sender::ClockSchedulerProfile::migration_feedback_bad}}}; for (const auto& [name, field] : fields) { (void)field; std::cout << ",sample_" << name; diff --git a/src/sender/core/ClockSchedulerProfile.hpp b/src/sender/core/ClockSchedulerProfile.hpp index e3ec862..82bc910 100644 --- a/src/sender/core/ClockSchedulerProfile.hpp +++ b/src/sender/core/ClockSchedulerProfile.hpp @@ -17,6 +17,10 @@ struct alignas(64) ClockSchedulerProfile { uint64_t tick_ns = 0, bridge_ns = 0, wait_ns = 0; uint64_t cluster_polls = 0, bridge_polls = 0, cluster_ticks = 0, bridge_commits = 0; uint64_t allowance_waits = 0, dependency_waits = 0, completion_loads = 0; + // Unlike sweep samples above, these count every migration planning attempt. + // Handoff time is request-to-commit wall elapsed, including overlapped work. + uint64_t migration_plans = 0, migration_planning_ns = 0, migration_handoff_ns = 0; + uint64_t migration_feedback_good = 0, migration_feedback_bad = 0; }; namespace detail { diff --git a/src/sender/core/TickSimulation.hpp b/src/sender/core/TickSimulation.hpp index 296be77..023a1e9 100644 --- a/src/sender/core/TickSimulation.hpp +++ b/src/sender/core/TickSimulation.hpp @@ -540,7 +540,7 @@ private: }; DynamicRuntimeCostEstimate dynamicUnitRuntimeCost_(size_t unit, double fallback) const; DynamicRuntimeCostEstimate dynamicClusterRuntimeCost_(size_t cluster); - DynamicRuntimeCostEstimate dynamicClockActorCost_(size_t actor); + DynamicRuntimeCostEstimate dynamicClockActorCost_(size_t actor, bool window = false); /** * Topology-only cluster-aware placement (no cost profiling). Used as @@ -783,6 +783,7 @@ private: void recordClockWaitSample_(size_t worker, const BlockedClusterInfo& blocker, SimTime edge_time, uint64_t elapsed_ns); void finishClockMigrationRun_(); + void recordClockMigrationHandoff_(); void rebuildThreadUnitsFromClusterOwners_(); bool maybeRequestEpochFreeMigration_(uint64_t cycle); void serviceEpochFreeMigration_(size_t worker_thread); diff --git a/src/sender/core/TickSimulationClockMigration.cpp b/src/sender/core/TickSimulationClockMigration.cpp index 476d32a..27c0af4 100644 --- a/src/sender/core/TickSimulationClockMigration.cpp +++ b/src/sender/core/TickSimulationClockMigration.cpp @@ -104,6 +104,15 @@ void TickSimulation::recordClockWaitSample_(size_t worker, const BlockedClusterI void TickSimulation::initializeClockMigration_() { auto& runtime = *clock_parallel_; const size_t clusters = clusters_.numClusters(); + // Calibrate only the timer overhead; never execute a model tick here. + uint64_t timer = UINT64_MAX; + auto before = detail::MigrationBenefit::now(); + for (unsigned i = 0; i < 32; ++i) { + const auto after = detail::MigrationBenefit::now(); + timer = std::min(timer, after - before); + before = after; + } + runtime.migration_benefit.timer_ns = timer; initDynamicMigrationRuntime_(); for (size_t worker = 0; worker < runtime.worker_bridges.size(); ++worker) for (const size_t b : runtime.worker_bridges[worker]) @@ -154,7 +163,8 @@ void TickSimulation::initializeClockMigration_() { std::memory_order_relaxed); } -TickSimulation::DynamicRuntimeCostEstimate TickSimulation::dynamicClockActorCost_(size_t actor) { +TickSimulation::DynamicRuntimeCostEstimate TickSimulation::dynamicClockActorCost_(size_t actor, + bool window) { DynamicRuntimeCostEstimate estimate; if (actor < clusters_.numClusters()) { estimate = dynamicClusterRuntimeCost_(actor); @@ -171,12 +181,68 @@ TickSimulation::DynamicRuntimeCostEstimate TickSimulation::dynamicClockActorCost : static_cast( clock_parallel_->bridges[actor - clusters_.numClusters()]->lanes.size()); } + if (window) { + auto& windows = clock_parallel_->migration_benefit.windows; + windows.resize(unit_ptrs_.size() + clock_parallel_->bridges.size()); + const double fallback = estimate.cost; + estimate = {}; + estimate.ready = true; + estimate.samples = UINT64_MAX; + const auto consume = [&](size_t slot, detail::MigrationBenefit::Sample sample) { + auto& value = windows[slot]; + value.observe(sample); + estimate.cost += value.cost; + estimate.samples = std::min(estimate.samples, value.samples); + estimate.ready &= value.ready; + }; + if (actor < clusters_.numClusters()) { + for (size_t u : clusters_.clusters[actor]) { + consume(u, + {dynamic_unit_active_sample_time_ns_[u].load(std::memory_order_relaxed), + dynamic_unit_active_sample_count_[u].load(std::memory_order_relaxed), + dynamic_unit_inactive_sample_time_ns_[u].load(std::memory_order_relaxed), + dynamic_unit_inactive_sample_count_[u].load(std::memory_order_relaxed), + dynamic_unit_observed_cycles_[u].load(std::memory_order_relaxed), + dynamic_unit_observed_active_ticks_[u].load(std::memory_order_relaxed)}); + } + } else { + // The numerator includes both phases, the denominator participating + // edges; use transaction count for confidence only. + const auto transactions = + cluster_active_sample_count_[actor].load(std::memory_order_relaxed); + const auto edges = cluster_sample_count_[actor].load(std::memory_order_relaxed); + consume(unit_ptrs_.size() + actor - clusters_.numClusters(), + {cluster_sample_time_ns_[actor].load(std::memory_order_relaxed), edges, 0, 0, + transactions, transactions}); + } + if (!estimate.ready) estimate.cost = fallback; + } estimate.cost *= clock_parallel_->actor_rates.at(actor); return estimate; } +void TickSimulation::recordClockMigrationHandoff_() { + auto& runtime = *clock_parallel_; + const auto requested = runtime.migration_requested_ns.load(std::memory_order_relaxed); + if (requested) { + const auto elapsed = detail::MigrationBenefit::now() - requested; + runtime.migration_handoff_ns.store(elapsed, std::memory_order_relaxed); + runtime.migration_handoff_total_ns.fetch_add(elapsed, std::memory_order_relaxed); + } +} + void TickSimulation::finishClockMigrationRun_() { if (!config_.enable_dynamic_rebalance) return; + if (config_.profile_clock_scheduler && !clock_scheduler_profile_.empty()) { + const auto& benefit = clock_parallel_->migration_benefit; + auto& profile = clock_scheduler_profile_.front(); + profile.migration_plans = benefit.planning_calls; + profile.migration_planning_ns = benefit.planning_total_ns; + profile.migration_handoff_ns = + clock_parallel_->migration_handoff_total_ns.load(std::memory_order_relaxed); + profile.migration_feedback_good = benefit.feedback_good; + profile.migration_feedback_bad = benefit.feedback_bad; + } // Workers have joined. A fence beyond this call's admission/stop boundary // must not leave a pending request blocking settlement or the next run. const size_t actor = migration_request_.cluster.load(std::memory_order_relaxed); diff --git a/src/sender/core/TickSimulationClockParallel.cpp b/src/sender/core/TickSimulationClockParallel.cpp index 5138708..44e21d0 100644 --- a/src/sender/core/TickSimulationClockParallel.cpp +++ b/src/sender/core/TickSimulationClockParallel.cpp @@ -135,6 +135,23 @@ uint64_t TickSimulation::runClockEpochFree_(uint64_t max_batches, std::optional< const auto token = stop_source_->get_token(); auto* trace = clock_trace_ && clock_trace_->parallelActive() ? clock_trace_.get() : nullptr; const bool dynamic = config_.enable_dynamic_rebalance; + double batches_per_reference_cycle = 0; + if (dynamic) { + for (auto* domain : runtime.indexed_domains) + batches_per_reference_cycle += + static_cast(domain->clock->period().denominator()) / + domain->clock->period().numerator() / config_.tick_frequency_hz; + runtime.migration_benefit.startRun(dynamicMigrationCycle_(), + detail::MigrationBenefit::now()); + runtime.migration_requested_ns.store(0, std::memory_order_relaxed); + } + if (dynamic) { + runtime.migration_max_batches = max_batches; + runtime.migration_window = window_limit; + runtime.migration_limit = limit ? clockRebalanceCycle_(*limit) : UINT64_MAX; + runtime.migration_batch_rate = batches_per_reference_cycle; + runtime.migration_completed_batches.store(0, std::memory_order_relaxed); + } if (dynamic) { epoch_free_dynamic_runtime_active_.store(true, std::memory_order_release); resetDynamicSchedulerMarkers_(); @@ -188,9 +205,11 @@ uint64_t TickSimulation::runClockEpochFree_(uint64_t max_batches, std::optional< --runtime.pending_size; progress = true; } - if (dynamic && progress) + if (dynamic && progress) { runtime.rebalance_cycle.store(clockRebalanceCycle_(clock_time_), std::memory_order_release); + runtime.migration_completed_batches.store(completed, std::memory_order_relaxed); + } // Retirement observes both unit and bridge completion with acquire. // Publish without waiting: actors sharing this worker must keep running // while the backend drains its bounded observation window. diff --git a/src/sender/core/TickSimulationClockRuntime.hpp b/src/sender/core/TickSimulationClockRuntime.hpp index 9089e08..c7b131e 100644 --- a/src/sender/core/TickSimulationClockRuntime.hpp +++ b/src/sender/core/TickSimulationClockRuntime.hpp @@ -5,6 +5,7 @@ #include #include "TickSimulation.hpp" +#include "sender/schedule/MigrationBenefit.hpp" namespace chronon::sender { @@ -95,6 +96,19 @@ struct TickSimulation::ClockParallelRuntime { // never incomparable actor-local cycles or calendar batch counts. std::atomic rebalance_cycle{0}; std::vector actor_rates; + detail::MigrationBenefit migration_benefit; // Exclusive planner, or joined workers. + std::atomic migration_completed_batches{0}; + uint64_t migration_max_batches = 0, migration_window = 0, migration_limit = UINT64_MAX; + double migration_batch_rate = 1; + double migrationHorizon(uint64_t cycle) const { + const auto completed = migration_completed_batches.load(std::memory_order_relaxed); + const auto left = migration_max_batches - std::min(completed, migration_max_batches); + const auto batches = left - std::min(left, migration_window); + return std::min(double(batches) / migration_batch_rate, + double(migration_limit - std::min(cycle, migration_limit))); + } + std::atomic migration_requested_ns{0}, migration_handoff_ns{0}; + std::atomic migration_handoff_total_ns{0}; }; } // namespace chronon::sender diff --git a/src/sender/core/TickSimulationDynamicRebalance.cpp b/src/sender/core/TickSimulationDynamicRebalance.cpp index 6bed522..155d306 100644 --- a/src/sender/core/TickSimulationDynamicRebalance.cpp +++ b/src/sender/core/TickSimulationDynamicRebalance.cpp @@ -140,6 +140,7 @@ void TickSimulation::serviceEpochFreeMigration_(size_t worker_thread) { dynamic_thread_no_ready_wait_ns_[t].store(0, std::memory_order_relaxed); } + if (clock_mode_) recordClockMigrationHandoff_(); ++rebalance_count_; cycles_since_last_actual_rebalance_ = 0; migration_request_.state.store(static_cast(MigrationRequestState::Committed), diff --git a/src/sender/core/TickSimulationPlanning.cpp b/src/sender/core/TickSimulationPlanning.cpp index 02fae07..e39bbb0 100644 --- a/src/sender/core/TickSimulationPlanning.cpp +++ b/src/sender/core/TickSimulationPlanning.cpp @@ -1,5 +1,5 @@ // SPDX-License-Identifier: MPL-2.0 -#include "TickSimulation.hpp" +#include "TickSimulationClockRuntime.hpp" #include "sender/schedule/PreparedTopologyCost.hpp" #include "sender/schedule/SmallPlacementCost.hpp" @@ -21,7 +21,7 @@ struct TickSimulation::PlanningScratch { epoch_free_cost::ObjectiveSummary fallback; std::vector thread_cost; std::vector thread_active_clusters, assignment, source_threads; - std::vector cluster_cost_ready; + std::vector cluster_cost_ready, thread_cost_ready; std::vector thread_floor_wait, thread_dep_wait, thread_no_ready_wait, cluster_blocked_wait, cluster_blocker_wait; }; @@ -63,6 +63,50 @@ bool TickSimulation::maybeRequestEpochFreeMigration_(uint64_t cycle) { const uint64_t planning_generation = cluster_assignment_generation_.load(std::memory_order_acquire); + auto* benefit = clock_mode_ ? &clock_parallel_->migration_benefit : nullptr; + const uint64_t planning_started = benefit ? detail::MigrationBenefit::now() : 0; + const uint64_t benefit_interval = std::max(interval, 4 * detail::kDynamicTickSampleInterval); + bool published = false, assessed = false; + struct BenefitGuard { + detail::MigrationBenefit* benefit; + std::atomic& next; + uint64_t start, cycle, interval; + bool& published; + bool& assessed; + ~BenefitGuard() { + if (!benefit) return; + const uint64_t elapsed = detail::MigrationBenefit::now() - start; + benefit->planning_ns = std::max(double(elapsed), benefit->planning_ns * 0.875); + ++benefit->planning_calls; + benefit->planning_total_ns += elapsed; + if (!published && !benefit->pending) { + if (assessed) benefit->defer(); + // Compare the next candidate against recent physical progress, + // not a lifetime rate that could hide a workload phase change. + benefit->window_cycle = cycle; + benefit->window_ns = start; + } + const uint64_t check = + detail::MigrationBenefit::add(cycle, benefit->interval(interval)); + auto old = next.load(std::memory_order_relaxed); + while (old < check && + !next.compare_exchange_weak(old, check, std::memory_order_relaxed)) { + } + } + } benefit_guard{benefit, next_dynamic_rebalance_check_cycle_, + planning_started, gate_cycle, + benefit_interval, published, + assessed}; + if (benefit) { + benefit->handoff_ns = + std::max(benefit->handoff_ns, + double(clock_parallel_->migration_handoff_ns.load(std::memory_order_relaxed))); + if (!benefit->feedback(cycle, planning_started, planning_generation, benefit_interval)) + return false; + if (clock_parallel_->migrationHorizon(cycle) < benefit_interval) return false; + benefit->confidence.resize(dynamic_runtime_cluster_count_); + } + const size_t num_threads = thread_units_.size(); const size_t num_clusters = dynamic_runtime_cluster_count_; auto& planning = schedulerScratch_().planning; @@ -89,18 +133,23 @@ bool TickSimulation::maybeRequestEpochFreeMigration_(uint64_t cycle) { assignment.assign(num_clusters, 0); auto& cluster_cost_ready = scratch.cluster_cost_ready; cluster_cost_ready.assign(num_clusters, 0); + auto& thread_cost_ready = scratch.thread_cost_ready; + thread_cost_ready.assign(num_threads, 1); for (size_t c = 0; c < num_clusters; ++c) { const auto estimate = - clock_mode_ ? dynamicClockActorCost_(c) : dynamicClusterRuntimeCost_(c); + clock_mode_ ? dynamicClockActorCost_(c, true) : dynamicClusterRuntimeCost_(c); cluster_cost[c] = estimate.cost; - cluster_cost_ready[c] = estimate.ready ? 1 : 0; + cluster_cost_ready[c] = benefit ? benefit->confidence[c].observe( + estimate.cost, estimate.samples, estimate.ready) + : estimate.ready; size_t owner = cluster_runtime_owner_[c].load(std::memory_order_acquire); if (owner >= num_threads) owner = 0; assignment[c] = owner; if (owner < num_threads) { thread_cost[owner] += cluster_cost[c]; + thread_cost_ready[owner] &= cluster_cost_ready[c]; if (cluster_cost[c] > 0.0) { ++thread_active_clusters[owner]; } @@ -119,6 +168,10 @@ bool TickSimulation::maybeRequestEpochFreeMigration_(uint64_t cycle) { dynamic_cluster_blocker_wait_ns_[c].load(std::memory_order_relaxed); } + assessed = !benefit || (std::any_of(thread_cost_ready.begin(), thread_cost_ready.end(), + [](uint8_t ready) { return ready != 0; }) && + std::any_of(cluster_cost_ready.begin(), cluster_cost_ready.end(), + [](uint8_t ready) { return ready != 0; })); double total_cost = 0.0; for (double cost : thread_cost) total_cost += cost; if (total_cost <= 0.0) { @@ -161,8 +214,10 @@ bool TickSimulation::maybeRequestEpochFreeMigration_(uint64_t cycle) { &cluster_blocked_wait, &cluster_blocker_wait}; auto& prepared = scratch.prepared; prepared.prepare(input, assignment, waits, planning_generation); - const uint64_t history_cooldown = std::max(config_.rebalance_cooldown_cycles, interval * 2); - const uint64_t pingpong_cooldown = std::max(history_cooldown, interval * 4); + const uint64_t history_cooldown = std::max(config_.rebalance_cooldown_cycles, + detail::MigrationBenefit::multiply(interval, 2)); + const uint64_t pingpong_cooldown = + std::max(history_cooldown, detail::MigrationBenefit::multiply(interval, 4)); auto last_migration_cycle = [&](size_t c) -> uint64_t { return c < dynamic_cluster_last_migration_cycle_.size() ? dynamic_cluster_last_migration_cycle_[c] @@ -180,6 +235,19 @@ bool TickSimulation::maybeRequestEpochFreeMigration_(uint64_t cycle) { dynamic_cluster_last_target_thread_[c] == candidate_source && cycle < saturatingCycleAdd(last_cycle, pingpong_cooldown); }; + const auto profitable = [&](size_t actor, size_t from, size_t to, + const epoch_free_cost::MoveBreakdown& move) { + if (!benefit) return true; + if (!thread_cost_ready[from] || !thread_cost_ready[to]) return false; + const double horizon = + std::min(clock_parallel_->migrationHorizon(cycle), + double(detail::MigrationBenefit::multiply(benefit_interval, 8))); + return benefit->profitable( + move.active_gain, move.topology_delta, + cluster_cost[actor] / clock_parallel_->actor_rates[actor], move.old_max_active, horizon, + detail::MigrationBenefit::now() - planning_started, benefit->timer_ns, num_clusters, + benefit->rate(cycle, planning_started), config_.rebalance_min_gain); + }; size_t source = SIZE_MAX; size_t cluster = SIZE_MAX; size_t target = SIZE_MAX; @@ -209,7 +277,9 @@ bool TickSimulation::maybeRequestEpochFreeMigration_(uint64_t cycle) { if (breakdown.score >= best_breakdown.score - prepared.roundoff()) breakdown = prepared.scoreFull(c, candidate_target, config_.rebalance_min_gain, churn); - if (!breakdown.valid) continue; + if (!breakdown.valid || + !profitable(c, candidate_source, candidate_target, breakdown)) + continue; if (breakdown.score > best_breakdown.score || (breakdown.score == best_breakdown.score && c < cluster)) { best_breakdown = breakdown; @@ -221,7 +291,7 @@ bool TickSimulation::maybeRequestEpochFreeMigration_(uint64_t cycle) { } } } - if (source == SIZE_MAX) { + if (source == SIZE_MAX && !benefit) { size_t fallback_source = source_threads.front(); for (size_t t : source_threads) { if (thread_cost[t] > thread_cost[fallback_source]) fallback_source = t; @@ -345,6 +415,16 @@ bool TickSimulation::maybeRequestEpochFreeMigration_(uint64_t cycle) { recordDynamicSchedulerMarker_("Chronon epoch-free rebalance requested", cycle, last_rebalance_detail_); + if (benefit) { + benefit->before_ns_per_cycle = benefit->rate(cycle, planning_started); + benefit->window_cycle = benefit->pending_cycle = cycle; + benefit->window_ns = planning_started; + benefit->pending_generation = planning_generation; + benefit->pending = true; + published = true; + clock_parallel_->migration_requested_ns.store(detail::MigrationBenefit::now(), + std::memory_order_relaxed); + } migration_request_.cluster.store(cluster, std::memory_order_relaxed); migration_request_.source_thread.store(source, std::memory_order_relaxed); migration_request_.target_thread.store(target, std::memory_order_relaxed); diff --git a/src/sender/schedule/MigrationBenefit.hpp b/src/sender/schedule/MigrationBenefit.hpp new file mode 100644 index 0000000..08b9981 --- /dev/null +++ b/src/sender/schedule/MigrationBenefit.hpp @@ -0,0 +1,144 @@ +// SPDX-License-Identifier: MPL-2.0 +#pragma once + +#include +#include +#include +#include +#include +#include + +namespace chronon::sender::detail { + +// Admission/feedback state for the existing clock planner. One planner owns +// this state; workers continue executing while estimates are inspected. Costs +// and progress share the configured reference-clock basis, never local cycles. +struct MigrationBenefit { + static uint64_t now() { + return std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); + } + struct Confidence { + double cost = 0; + uint64_t samples = 0; + bool stable = false; + bool observe(double next, uint64_t count, bool ready) { + if (!ready || !std::isfinite(next)) { + *this = {}; + return false; + } + if (!samples || count < samples) { + cost = next; + samples = count; + stable = false; + } else if (count - samples >= 4) { + stable = std::abs(next - cost) <= 0.25 * std::max(next, cost); + cost = next; + samples = count; + } + return stable; + } + }; + struct Sample { + uint64_t active_ns = 0, active = 0, inactive_ns = 0, inactive = 0; + uint64_t cycles = 0, active_cycles = 0; + }; + struct Window { + Sample previous; + double cost = 0; + uint64_t samples = 0; + bool ready = false; + void observe(Sample now) { + if (now.active < previous.active || now.inactive < previous.inactive || + now.cycles < previous.cycles) + *this = {}; + const auto cycles = now.cycles - previous.cycles; + if (cycles < 4) return; + const auto active_cycles = std::min(cycles, now.active_cycles - previous.active_cycles); + const auto active = now.active - previous.active; + const auto inactive = now.inactive - previous.inactive; + if ((active_cycles && active < 4) || (active_cycles < cycles && inactive < 4)) return; + const double rate = double(active_cycles) / cycles; + cost = active_cycles ? rate * double(now.active_ns - previous.active_ns) / active : 0; + if (active_cycles < cycles) + cost += (1 - rate) * double(now.inactive_ns - previous.inactive_ns) / inactive; + samples += active_cycles == cycles ? active + : !active_cycles ? inactive + : std::min(active, inactive); + ready = true; + previous = now; + } + }; + std::vector confidence; + std::vector windows; + unsigned backoff = 1; + double planning_ns = 0, handoff_ns = 0, timer_ns = 0; + uint64_t window_cycle = 0, window_ns = 0; + double before_ns_per_cycle = 0; + uint64_t pending_generation = 0, pending_cycle = 0; + bool pending = false; + uint64_t planning_calls = 0, planning_total_ns = 0; + uint64_t feedback_good = 0, feedback_bad = 0; + + static uint64_t add(uint64_t a, uint64_t b) { return a + std::min(b, UINT64_MAX - a); } + static uint64_t multiply(uint64_t a, uint64_t b) { + return a && b > UINT64_MAX / a ? UINT64_MAX : a * b; + } + static double netSaving(double active_gain, double topology_delta) { + // Heuristic bonuses rank moves but are not nanoseconds saved. Do not + // claim that removing communication necessarily speeds the workload up; + // charge added communication fully and discount the active estimate. + return 0.5 * active_gain + std::min(0.0, topology_delta); + } + bool profitable(double active_gain, double topology_delta, double actor_tick_ns, + double max_active, double horizon, double current_planning_ns, double timer_ns, + size_t actors, double observed_ns_per_cycle, double min_gain) const { + const double saving = netSaving(active_gain, topology_delta); + // Timer/profiling allowance and a cold-cache allowance prevent a tiny + // nominal gain from buying frequent moves. These are conservative cost + // estimates, not an assertion that all elapsed handoff time is CPU work. + const double overhead = std::max(planning_ns, current_planning_ns) + handoff_ns + + 2 * timer_ns * actors + 8 * actor_tick_ns; + // Model cost omits coordinator/polling work. A tiny fractional model + // improvement cannot establish a useful wall-time gain when that work + // dominates. Apply the same safety margin to the requested gain floor. + const double uncertainty = + std::max(0.02 * max_active, 4 * min_gain * observed_ns_per_cycle); + return std::isfinite(saving) && observed_ns_per_cycle > 0 && saving > uncertainty && + horizon > 0 && saving * horizon > 4 * overhead; + } + void defer() { backoff = std::min(32u, backoff * 2); } + uint64_t interval(uint64_t base) const { return multiply(base, backoff); } + void startRun(uint64_t cycle, uint64_t now) { + window_cycle = cycle; + window_ns = now; + pending = false; // Never include time while the simulation was stopped. + } + double rate(uint64_t cycle, uint64_t now) const { + return cycle > window_cycle && now >= window_ns + ? static_cast(now - window_ns) / (cycle - window_cycle) + : 0; + } + // Returns false while one move is being assessed. There is no execution + // fence: only another ownership move waits for enough physical progress. + bool feedback(uint64_t cycle, uint64_t now, uint64_t generation, uint64_t interval) { + if (!pending) return true; + if (generation == pending_generation) return false; + if (cycle < add(pending_cycle, multiply(interval, 4))) return false; + const double after = rate(cycle, now); + if (after > 0 && before_ns_per_cycle > 0 && after < before_ns_per_cycle * 0.98) { + ++feedback_good; + backoff = 1; + } else { + ++feedback_bad; + defer(); + } + pending = false; + window_cycle = cycle; + window_ns = now; + return true; + } +}; + +} // namespace chronon::sender::detail