#include #ifndef SERIAL_PROMPT_SNAKE_TURN_SWAP #define SERIAL_PROMPT_SNAKE_TURN_SWAP 1 #endif #ifndef CASE12_RESIDUAL_AGGRESSIVE_PROMPT_ORDER #define CASE12_RESIDUAL_AGGRESSIVE_PROMPT_ORDER 1 #endif #ifndef CASE12_RESIDUAL_AGGRESSIVE_LATER_ORDER #define CASE12_RESIDUAL_AGGRESSIVE_LATER_ORDER CASE12_RESIDUAL_AGGRESSIVE_PROMPT_ORDER #endif #ifndef CASE12_RESIDUAL_CANONICAL_FIRST_DECODE #define CASE12_RESIDUAL_CANONICAL_FIRST_DECODE 1 #endif using namespace std; // SJF is valuable only when ready prompts differ enough to justify disturbing // FIFO/cohort order. All gates below use public objective parameters and the // currently visible ready set. constexpr long double kSjfSpreadSloFraction = 0.15L; constexpr long double kSjfMinWaitingWeight = 0.02L; constexpr int kSjfSpreadStageMask = 1; // Gate P PRE only. constexpr long double kSjfBacklogPerCloud = 8.0L; constexpr long double kSjfLatencyDominantWeight = 0.90L; constexpr size_t kFlatDecodeBurstPerCloud = 4; #ifndef PIPELINE_POST_SYNC_MIN_THROUGHPUT_WEIGHT #define PIPELINE_POST_SYNC_MIN_THROUGHPUT_WEIGHT 0.10L #endif #ifndef STAGGERED_PREFILL_POST_PROTECTION_MODE #define STAGGERED_PREFILL_POST_PROTECTION_MODE 3 #endif #ifndef CROSS_PREFILL_POST_MIN_WAITING_WEIGHT #define CROSS_PREFILL_POST_MIN_WAITING_WEIGHT 0.10L #endif #ifndef SERIALIZED_PREEMPT_MODE #define SERIALIZED_PREEMPT_MODE 3 #endif #ifndef LONG_DECODE_POSTERIOR #define LONG_DECODE_POSTERIOR 0.895L #endif #ifndef PURE_TP_CLOUD_COHORT_POSTERIOR #define PURE_TP_CLOUD_COHORT_POSTERIOR 0.895L #endif #ifndef PURE_TP_CLOUD_COHORT_MIN_POSTS #define PURE_TP_CLOUD_COHORT_MIN_POSTS 8 #endif #ifndef PREFILL_UPLOAD_ADMISSION_GATE_MODE #define PREFILL_UPLOAD_ADMISSION_GATE_MODE 1 #endif #ifndef PREFILL_UPLOAD_QUEUE_LIMIT #define PREFILL_UPLOAD_QUEUE_LIMIT 0 #endif #ifndef PREFILL_UPLOAD_LONG_ROUNDS #define PREFILL_UPLOAD_LONG_ROUNDS 3 #endif #ifndef PREFILL_UPLOAD_LONG_OFFSET #define PREFILL_UPLOAD_LONG_OFFSET 1 #endif #ifndef PREFILL_UPLOAD_BURST_DIVISOR #define PREFILL_UPLOAD_BURST_DIVISOR 2 #endif #ifndef PREFILL_UPLOAD_MIN_BURST_ROUNDS #define PREFILL_UPLOAD_MIN_BURST_ROUNDS 8 #endif #ifndef PREFILL_UPLOAD_NETWORK_SLO_RATIO #define PREFILL_UPLOAD_NETWORK_SLO_RATIO 0.25L #endif #ifndef PREFILL_UPLOAD_MAX_THROUGHPUT_WEIGHT #define PREFILL_UPLOAD_MAX_THROUGHPUT_WEIGHT 0.95L #endif #ifndef PREFILL_UPLOAD_DOMINANCE_RATIO #define PREFILL_UPLOAD_DOMINANCE_RATIO 2.0L #endif #ifndef PREFILL_UPLOAD_WAITING_EVENT_MASK #define PREFILL_UPLOAD_WAITING_EVENT_MASK 1 #endif #ifndef PUBLIC_DECODE_BUILD_TARGET #define PUBLIC_DECODE_BUILD_TARGET 0 #endif #ifndef PUBLIC_DECODE_RESERVOIR_PERCENT #define PUBLIC_DECODE_RESERVOIR_PERCENT 100 #endif #ifndef PUBLIC_DECODE_RESERVOIR_BONUS #define PUBLIC_DECODE_RESERVOIR_BONUS 2 #endif #ifndef PUBLIC_DECODE_DRAIN_TARGET #define PUBLIC_DECODE_DRAIN_TARGET 0 #endif #ifndef PUBLIC_DECODE_NEWEST_TIE #define PUBLIC_DECODE_NEWEST_TIE 0 #endif #ifndef PUBLIC_DECODE_RESERVOIR_HYSTERESIS #define PUBLIC_DECODE_RESERVOIR_HYSTERESIS 0 #endif #ifndef PUBLIC_DECODE_MIN_EQUAL_FINISHES #define PUBLIC_DECODE_MIN_EQUAL_FINISHES 2 #endif // A pre-reservoir proof uses only completions, continuations, and work that is // already visible. Two live chains that have each survived twice the public // knee support projecting an issued sole-cloud prefill into the next decode // frontier; hidden remaining lengths and future arrivals are never predicted. #ifndef PUBLIC_DECODE_EARLY_SURVIVORS #define PUBLIC_DECODE_EARLY_SURVIVORS 2 #endif #ifndef PUBLIC_DECODE_EARLY_TOKEN_PERCENT #define PUBLIC_DECODE_EARLY_TOKEN_PERCENT 200 #endif #ifndef PUBLIC_DECODE_EARLY_READY_PERCENT #define PUBLIC_DECODE_EARLY_READY_PERCENT 100 #endif #ifndef PUBLIC_DECODE_EARLY_PROJECT_PREFILL #define PUBLIC_DECODE_EARLY_PROJECT_PREFILL 1 #endif #ifndef PUBLIC_DECODE_EARLY_PROJECT_MIN_FINISHES #define PUBLIC_DECODE_EARLY_PROJECT_MIN_FINISHES 1 #endif #ifndef PUBLIC_DECODE_EARLY_PROJECT_MIN_PHASE #define PUBLIC_DECODE_EARLY_PROJECT_MIN_PHASE 2 #endif #ifndef PUBLIC_DECODE_EARLY_PROJECT_PROMPT_RANGE #define PUBLIC_DECODE_EARLY_PROJECT_PROMPT_RANGE 1 #endif #ifndef LATENCY_DEEP_FRESH_DPRE_TARGET #define LATENCY_DEEP_FRESH_DPRE_TARGET 12 #endif #ifndef LATENCY_DEEP_PPRE_CLASS_CHUNK #define LATENCY_DEEP_PPRE_CLASS_CHUNK 3 #endif // In a public two-cloud flow-shop regime with long, input-independent prompt // computation, reserve one remote for an early decode wave after this many // prompt computations have run while decode work was already ready. #ifndef PUBLIC_RANK_PIPELINE_RUN #define PUBLIC_RANK_PIPELINE_RUN 7 #endif // Soft public input-rank prior, learned consistently across independent // workload families. It is used only when public batching capacity supports // remote-sized cohorts and the currently visible class frequencies are close // enough that a fixed capacity split is meaningful. Three quarters of class // assignments use the prior pools; a golden-angle-like stride interleaves the // ordinary live-load policy to retain adaptation. constexpr int kRankPriorPercent = 75; constexpr int kRankPriorStride = 38; constexpr int kRankPriorEvidencePerRemote = 2; constexpr int kRankPriorMaxBatchMultiplier = 2; constexpr int kRankPriorMaxClassRatioPercent = 125; namespace { using RequestId = long long; enum class Phase { ReadyPrefillPre, RunningPrefillPre, WaitingPrefillUpload, ReadyPrefillProcess, RunningPrefillProcess, WaitingPrefillDownload, ReadyPrefillPost, RunningPrefillPost, ReadyDecodePre, RunningDecodePre, WaitingDecodeUpload, ReadyDecodeProcess, RunningDecodeProcess, WaitingDecodeDownload, ReadyDecodePost, RunningDecodePost, Finished, }; struct Request { RequestId id = -1; long long inputLength = 0; long double arrivalTime = 0; long long arrivalOrder = 0; long double prefillProcessTime = 0; // Exact completion time of the already-enqueued prompt DOWN transfer. // This is derived only from issued work and the public FIFO link model. long double prefillDownloadFinishTime = 0; long double decodePreReadyTime = 0; long double decodeProcessStartTime = 0; long long tokensProduced = 0; // Set only after this request is selected in the public prompt/decode // overlap rule. It carries no hidden survival information. bool publicRankPipeline = false; int remote = -1; int nextLayer = 0; int runningLayerEnd = 0; Phase phase = Phase::ReadyPrefillPre; }; struct TimingRow { long double size = 0; array duration{}; }; vector splitWords(const string& line) { istringstream in(line); vector words; string word; while (in >> word) words.push_back(word); return words; } bool parseInteger(const string& text, long long& value) { try { size_t used = 0; value = stoll(text, &used); return used == text.size(); } catch (...) { return false; } } int cloudIndex(const string& server) { if (server.size() < 2 || server[0] != 'C') return -1; long long index = -1; if (!parseInteger(server.substr(1), index)) return -1; if (index < 0 || index > numeric_limits::max()) return -1; return static_cast(index); } class Scheduler { public: bool readConfiguration() { if (!(cin >> cloudCount_ >> scheduleCost_ >> networkLatency_ >> bandwidth_ >> bytesPerToken_ >> layerCount_)) { return false; } if (!(cin >> sloFirstToken_ >> sloTokenGap_ >> throughputUpper_ >> throughputBase_ >> distanceBase_ >> throughputWeight_ >> waitingWeight_)) { return false; } int rowCount = 0; if (!(cin >> rowCount) || cloudCount_ < 0 || layerCount_ <= 0 || rowCount < 0) { return false; } for (int i = 0; i < rowCount; ++i) { TimingRow row; if (!(cin >> row.size)) return false; for (long double& value : row.duration) { if (!(cin >> value)) return false; } for (int column = 0; column < 6; ++column) { if (row.duration[column] >= 0) { curves_[column].emplace_back(row.size, row.duration[column]); } } } for (auto& curve : curves_) { sort(curve.begin(), curve.end()); } cloudBusy_.assign(cloudCount_, false); cloudDecodeStreak_.assign(cloudCount_, 0); cloudPromptRunsWithReadyDecode_.assign(cloudCount_, 0); cloudPrefillFinish_.assign(cloudCount_, 0); decodeProcessSingletonTime_ = interpolatedDuration(1, 4); return true; } void run() { string firstToken; while (cin >> firstToken) { if (firstToken == "END") return; long double now = 0; try { size_t used = 0; now = stold(firstToken, &used); if (used != firstToken.size()) return; } catch (...) { return; } int eventCount = 0; if (!(cin >> eventCount) || eventCount < 0) return; string line; getline(cin, line); currentTime_ = now; vector> frame; frame.reserve(eventCount); for (int i = 0; i < eventCount; ++i) { if (!getline(cin, line)) return; // Never act on a truncated frame. vector words = splitWords(line); if (words.empty()) return; frame.push_back(move(words)); } // Applying the whole frame before making a decision is essential: a FIN in // the same frame as D POST completion must suppress the next decode token. tokenCompletionsThisFrame_.clear(); for (const auto& event : frame) applyEvent(event, now); finalizeTokenObservations(); vector assignments = makeAssignments(); cout << assignments.size() << '\n'; for (const string& assignment : assignments) cout << assignment << '\n'; cout.flush(); } } private: int cloudCount_ = 0; long double scheduleCost_ = 0; long double networkLatency_ = 0; long double bandwidth_ = 0; long long bytesPerToken_ = 0; int layerCount_ = 0; long double sloFirstToken_ = 0; long double sloTokenGap_ = 0; long double throughputUpper_ = 0; long double throughputBase_ = 0; long double distanceBase_ = 0; long double throughputWeight_ = 0; long double waitingWeight_ = 0; array>, 6> curves_; long double decodeProcessSingletonTime_ = 0; unordered_map requests_; vector arrivalOrder_; long long nextArrivalOrder_ = 0; bool edgeBusy_ = false; int edgeDecodeStreak_ = 0; vector cloudBusy_; vector cloudDecodeStreak_; vector cloudPromptRunsWithReadyDecode_; vector cloudPrefillFinish_; int nextRemoteTieBreak_ = 0; long long firstTokenPosts_ = 0; long long firstTokenContinuations_ = 0; long long tokenPosts_ = 0; long long tokenContinuations_ = 0; long long completedDecodeRequests_ = 0; long long minimumCompletedTokens_ = numeric_limits::max(); long long maximumCompletedTokens_ = 0; vector> tokenCompletionsThisFrame_; int pilotTargetLatched_ = 0; mutable bool sjfAdmissionSuppressed_ = false; bool freshBurstLevelingEngaged_ = false; bool freshDpostPipelineEngaged_ = false; bool residualAggressiveLaterEngaged_ = false; mutable bool prefillUploadPipelineEngaged_ = false; // A public prompt-drain latch used by the conservative two-cloud // flow-shop rule below. The cohort boundary is the number of arrivals // already visible when the rule first becomes applicable; later arrivals // never become a reason to speculate about the current drain. mutable bool safePromptTailDrainEngaged_ = false; mutable long long safePromptTailCohortLimit_ = 0; // Once an actually observed long decode stream reaches a strong public // whole-round timing knee, keep decode cohorts near that knee. This latch // contains no request-count or arrival-time oracle: it is opened only by // already ready work and the public DPRE/DPROC/DPOST curves. bool publicDecodeKneeEngaged_ = false; bool publicDecodeReservoirFilled_ = false; long double currentTime_ = 0; long double promptUpAvailableTime_ = 0; long double promptDownAvailableTime_ = 0; long double promptTransferDuration(long long inputLength) const { if (bandwidth_ <= 0) return numeric_limits::infinity(); return networkLatency_ + 8.0L * max(0LL, inputLength) * bytesPerToken_ / (bandwidth_ * 1.0e6L); } Request* findRequest(RequestId id) { auto it = requests_.find(id); return it == requests_.end() ? nullptr : &it->second; } static bool isDecodePhase(Phase phase) { return phase >= Phase::ReadyDecodePre && phase <= Phase::RunningDecodePost; } static bool isPrefillComputePhase(Phase phase) { return phase == Phase::RunningPrefillPre || phase == Phase::WaitingPrefillUpload || phase == Phase::ReadyPrefillProcess || phase == Phase::RunningPrefillProcess || phase == Phase::WaitingPrefillDownload || phase == Phase::ReadyPrefillPost || phase == Phase::RunningPrefillPost; } void applyEvent(const vector& event, long double now) { if (event.empty()) return; if (event[0] == "ARR") { applyArrival(event, now); } else if (event[0] == "TDN") { applyTaskDone(event); } else if (event[0] == "XDN") { applyTransferDone(event); } else if (event[0] == "FIN") { applyFinish(event); } } void applyArrival(const vector& event, long double now) { if (event.size() < 3) return; long long id = -1, inputLength = 0; if (!parseInteger(event[1], id) || !parseInteger(event[2], inputLength)) return; // Request identifiers are unique in legal inputs. Keeping this guard makes a // malformed duplicate harmless instead of corrupting a running request. if (requests_.count(id)) return; Request request; request.id = id; request.inputLength = max(0LL, inputLength); request.arrivalTime = now; request.arrivalOrder = nextArrivalOrder_++; request.prefillProcessTime = interpolatedDuration(request.inputLength, 1); requests_.emplace(id, request); arrivalOrder_.push_back(id); } void releaseServer(const string& server) { if (server == "E") { edgeBusy_ = false; return; } int cloud = cloudIndex(server); if (0 <= cloud && cloud < cloudCount_) cloudBusy_[cloud] = false; } void applyTaskDone(const vector& event) { // TDN if (event.size() < 5) return; releaseServer(event[1]); const string& family = event[2]; const string& stage = event[3]; if (family == "P") { if (stage == "PRE" || stage == "POST") { if (event.size() < 7) return; long long id = -1; if (!parseInteger(event[5], id)) return; Request* request = findRequest(id); if (!request || request->phase == Phase::Finished) return; request->phase = (stage == "PRE") ? Phase::WaitingPrefillUpload : Phase::ReadyDecodePre; if (stage == "PRE") { promptUpAvailableTime_ = max(promptUpAvailableTime_, currentTime_) + promptTransferDuration(request->inputLength); } else { request->decodePreReadyTime = currentTime_; } } else if (stage == "PROC") { if (event.size() < 9) return; long long layerEnd = 0, id = -1; if (!parseInteger(event[5], layerEnd) || !parseInteger(event[7], id)) return; Request* request = findRequest(id); if (!request || request->phase == Phase::Finished) return; request->nextLayer = clamp(static_cast(layerEnd), 0, layerCount_); request->phase = request->nextLayer == layerCount_ ? Phase::WaitingPrefillDownload : Phase::ReadyPrefillProcess; if (request->nextLayer == layerCount_) { promptDownAvailableTime_ = max(promptDownAvailableTime_, currentTime_) + promptTransferDuration(request->inputLength); request->prefillDownloadFinishTime = promptDownAvailableTime_; } const int cloud = cloudIndex(event[1]); if (0 <= cloud && cloud < cloudCount_) { cloudPrefillFinish_[cloud] = 0; } } return; } if (family != "D" || event.size() < 7) return; long long batchSize = 0; if (!parseInteger(event[5], batchSize) || batchSize < 0) return; const size_t firstId = 6; const size_t available = event.size() > firstId ? event.size() - firstId - 1 : 0; const size_t count = min(static_cast(batchSize), available); for (size_t i = 0; i < count; ++i) { long long id = -1; if (!parseInteger(event[firstId + i], id)) continue; Request* request = findRequest(id); if (!request || request->phase == Phase::Finished) continue; if (stage == "PRE") request->phase = Phase::WaitingDecodeUpload; else if (stage == "PROC") request->phase = Phase::WaitingDecodeDownload; else if (stage == "POST") { const bool first = request->tokensProduced == 0; ++request->tokensProduced; request->phase = Phase::ReadyDecodePre; tokenCompletionsThisFrame_.emplace_back(id, first); } } } void applyTransferDone(const vector& event) { // XDN if (event.size() < 7) return; const string& direction = event[1]; const string& kind = event[4]; long long batchSize = 0; if (!parseInteger(event[5], batchSize) || batchSize < 0) return; const size_t available = event.size() - 6; const size_t count = min(static_cast(batchSize), available); for (size_t i = 0; i < count; ++i) { long long id = -1; if (!parseInteger(event[6 + i], id)) continue; Request* request = findRequest(id); if (!request || request->phase == Phase::Finished) continue; if (kind == "PRE" && direction == "UP") { request->phase = Phase::ReadyPrefillProcess; } else if (kind == "PRE" && direction == "DOWN") { request->phase = Phase::ReadyPrefillPost; } else if (kind == "DEC" && direction == "UP") { request->phase = Phase::ReadyDecodeProcess; } else if (kind == "DEC" && direction == "DOWN") { request->phase = Phase::ReadyDecodePost; } } } void applyFinish(const vector& event) { if (event.size() < 2) return; long long id = -1; if (!parseInteger(event[1], id)) return; Request* request = findRequest(id); if (request && request->phase != Phase::Finished) { ++completedDecodeRequests_; minimumCompletedTokens_ = min(minimumCompletedTokens_, request->tokensProduced); maximumCompletedTokens_ = max(maximumCompletedTokens_, request->tokensProduced); request->phase = Phase::Finished; } } void finalizeTokenObservations() { for (const auto& [id, first] : tokenCompletionsThisFrame_) { Request* request = findRequest(id); const bool continued = request && request->phase != Phase::Finished; ++tokenPosts_; if (continued) ++tokenContinuations_; if (first) { ++firstTokenPosts_; if (continued) ++firstTokenContinuations_; } } if (freshDpostPipelineEngaged_) { const bool sustainFragmentedDecodeWave = shouldSustainFragmentedDecodeWave(); const bool visiblePrefillRequest = any_of( arrivalOrder_.begin(), arrivalOrder_.end(), [&](RequestId id) { const Request& request = requests_.at(id); return request.phase < Phase::ReadyDecodePre; }); const bool visibleRequest = any_of( arrivalOrder_.begin(), arrivalOrder_.end(), [&](RequestId id) { const Request& request = requests_.at(id); if (sustainFragmentedDecodeWave) { return request.phase != Phase::Finished && isDecodePhase(request.phase); } return request.phase != Phase::Finished && request.tokensProduced == 0; }); const bool visibleFreshCloudOrDown = any_of( arrivalOrder_.begin(), arrivalOrder_.end(), [&](RequestId id) { const Request& request = requests_.at(id); return request.tokensProduced == 0 && (request.phase == Phase::RunningDecodeProcess || request.phase == Phase::WaitingDecodeDownload); }); const bool latencyTailCohort = !sustainFragmentedDecodeWave && waitingWeight_ > throughputWeight_ && !visibleFreshCloudOrDown; if (!visibleRequest || latencyTailCohort || (sustainFragmentedDecodeWave && visiblePrefillRequest)) { freshDpostPipelineEngaged_ = false; } } } bool shouldSustainFragmentedDecodeWave() const { if (waitingWeight_ <= throughputWeight_ || cloudCount_ < 4) return false; return adjacent_find( curves_[4].begin(), curves_[4].end(), [](const auto& left, const auto& right) { return right.second + 1.0e-12L < left.second; }) == curves_[4].end(); } long double interpolatedDuration(long double size, int column) const { const auto& points = curves_[column]; if (points.empty()) return 0; if (size <= points.front().first) return points.front().second; if (size >= points.back().first) return points.back().second; auto upper = upper_bound(points.begin(), points.end(), size, [](long double value, const auto& point) { return value < point.first; }); auto lower = prev(upper); if (upper->first == lower->first) return lower->second; long double ratio = (size - lower->first) / (upper->first - lower->first); return lower->second + ratio * (upper->second - lower->second); } long double publicDecodeRoundDuration(int count) const { if (count <= 0 || bandwidth_ <= 0) { return numeric_limits::infinity(); } const long double transfers = 2.0L * (networkLatency_ + 8.0L * count * bytesPerToken_ / (bandwidth_ * 1.0e6L)); return 3.0L * scheduleCost_ + transfers + interpolatedDuration(count, 3) + interpolatedDuration(count, 4) + interpolatedDuration(count, 5); } int strongPublicDecodeKnee() const { if (curves_[3].empty() || curves_[4].empty() || curves_[5].empty()) { return 0; } const int maximum = max(1, static_cast( floor(curves_[4].back().first + 1.0e-12L))); int best = 1; long double bestPerRequest = publicDecodeRoundDuration(1); for (int count = 2; count <= maximum; ++count) { const long double perRequest = publicDecodeRoundDuration(count) / count; if (perRequest + 1.0e-12L < bestPerRequest) { bestPerRequest = perRequest; best = count; } } const long double maximumPerRequest = publicDecodeRoundDuration(maximum) / maximum; // A material, interior knee is qualitatively different from ordinary // batching noise. Keep this endpoint dormant on monotone/flat curves. if (best < 8 || 5 * best > 4 * maximum || 20.0L * bestPerRequest >= 19.0L * maximumPerRequest) { return 0; } return best; } int publicDecodeReservoirBuildSize(int knee) const { #if PUBLIC_DECODE_BUILD_TARGET > 0 return PUBLIC_DECODE_BUILD_TARGET; #else // The predecessor of the global knee on the lower convex envelope is // the cheapest public way to build backlog: alternating this point with // the global knee realizes every mean cohort size between them without // touching the noisy points above the envelope. vector> hull; for (int count = 1; count <= knee; ++count) { const pair point = {count, publicDecodeRoundDuration(count)}; while (hull.size() >= 2) { const auto& a = hull[hull.size() - 2]; const auto& b = hull[hull.size() - 1]; const long double left = (b.second - a.second) * (point.first - b.first); const long double right = (point.second - b.second) * (b.first - a.first); if (left + 1.0e-12L < right) break; hull.pop_back(); } hull.push_back(point); } return hull.size() >= 2 ? hull[hull.size() - 2].first : knee; #endif } bool earlyProjectedPrefillEligible(const Request& request) const { if (static_cast(request.phase) >= PUBLIC_DECODE_EARLY_PROJECT_MIN_PHASE) { return true; } return PUBLIC_DECODE_EARLY_PROJECT_PROMPT_RANGE && !curves_[0].empty() && request.inputLength <= curves_[0].back().first + 1.0e-12L; } bool publicDecodeKneeEligible(int readyCount) { const int knee = strongPublicDecodeKnee(); if (knee == 0 || cloudCount_ != 1 || waitingWeight_ > 0 || throughputWeight_ <= 0 || !staggeredArrivalsObserved() || !longDecodeObserved() || sloTokenGap_ <= 0 || bandwidth_ <= 0) { return false; } // Ordinarily two equal completed chains prove that the stream is long // enough to amortize reservoir construction. There is one earlier, // still fully public bridge: after the first long FIN, two other live // chains have themselves already survived twice the public knee. A // sole-cloud prefill that has completed PPRE (or whose input remains // inside the measured public PPRE range) is guaranteed future decode // work and may complete a projected knee frontier. Any observed // completion-length mismatch closes the endpoint immediately. const bool completedEvidence = completedDecodeRequests_ >= PUBLIC_DECODE_MIN_EQUAL_FINISHES && minimumCompletedTokens_ == maximumCompletedTokens_ && minimumCompletedTokens_ >= 2LL * knee; int observedLongSurvivors = 0; if (PUBLIC_DECODE_EARLY_SURVIVORS > 0 && (completedDecodeRequests_ == 0 || (minimumCompletedTokens_ == maximumCompletedTokens_ && minimumCompletedTokens_ >= 2LL * knee))) { const long long observedTokenFloor = (1LL * PUBLIC_DECODE_EARLY_TOKEN_PERCENT * knee + 99) / 100; for (RequestId id : arrivalOrder_) { const Request& request = requests_.at(id); if (request.phase != Phase::Finished && request.tokensProduced >= observedTokenFloor) { ++observedLongSurvivors; } } } const bool earlyEvidence = PUBLIC_DECODE_EARLY_SURVIVORS > 0 && observedLongSurvivors >= PUBLIC_DECODE_EARLY_SURVIVORS; if (!completedEvidence && !earlyEvidence) { return false; } const long double oneTokenTransfer = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); if (oneTokenTransfer > 0.10L * sloTokenGap_) return false; int projectedReadyCount = readyCount; if (PUBLIC_DECODE_EARLY_PROJECT_PREFILL && completedDecodeRequests_ >= PUBLIC_DECODE_EARLY_PROJECT_MIN_FINISHES) { for (RequestId id : arrivalOrder_) { const Request& request = requests_.at(id); if (request.phase != Phase::Finished && request.phase < Phase::ReadyDecodePre && earlyProjectedPrefillEligible(request)) { ++projectedReadyCount; } } } const int earlyReadyTarget = max(1, (PUBLIC_DECODE_EARLY_READY_PERCENT * knee + 99) / 100); const int engageTarget = PUBLIC_DECODE_EARLY_PROJECT_PREFILL ? earlyReadyTarget : (completedEvidence ? knee : earlyReadyTarget); if (!publicDecodeKneeEngaged_ && projectedReadyCount >= engageTarget) { publicDecodeKneeEngaged_ = true; } return publicDecodeKneeEngaged_; } bool decodeRoundStillReturning(int cloud) const { return any_of(arrivalOrder_.begin(), arrivalOrder_.end(), [&](RequestId id) { const Request& request = requests_.at(id); if (request.remote != cloud || request.phase == Phase::Finished) { return false; } return request.phase == Phase::ReadyDecodePre || request.phase == Phase::RunningDecodePre || request.phase == Phase::WaitingDecodeUpload || request.phase == Phase::RunningDecodeProcess || request.phase == Phase::WaitingDecodeDownload || request.phase == Phase::ReadyDecodePost || request.phase == Phase::RunningDecodePost; }); } long double throughputPressure() const { if (throughputWeight_ <= 0) return 0; if (waitingWeight_ <= 0) return 1; if (distanceBase_ <= 0) return 0; const long double tpRange = max(1e-12L, throughputUpper_ - throughputBase_); const long double tpSlope = throughputWeight_ / tpRange; const long double waitSlope = waitingWeight_ / distanceBase_; return tpSlope / max(1e-18L, tpSlope + waitSlope); } int evidenceTarget() const { const int localEvidence = 1 + static_cast(llround(3 * throughputPressure())); return max(cloudCount_, localEvidence); } bool longDecodeObserved() const { const int evidence = max(1, evidenceTarget()); const long double firstPosterior = (firstTokenContinuations_ + 1.0L) / (firstTokenPosts_ + 2.0L); const long double allPosterior = (tokenContinuations_ + 1.0L) / (tokenPosts_ + 2.0L); const bool firstSignal = firstTokenPosts_ >= evidence && firstPosterior >= LONG_DECODE_POSTERIOR; const bool sustainedSignal = tokenPosts_ >= 2LL * evidence && allPosterior >= LONG_DECODE_POSTERIOR; return firstSignal || sustainedSignal; } bool shortDecodeObserved() const { const int evidence = 1; if (firstTokenPosts_ < evidence) return false; const long double firstPosterior = (firstTokenContinuations_ + 1.0L) / (firstTokenPosts_ + 2.0L); return firstPosterior <= 0.50L; } bool latencyRelevantDeepBurstObserved() const { if (cloudCount_ <= 0 || 3.0L * waitingWeight_ < 2.0L * throughputWeight_) { return false; } const size_t target = static_cast(8 * max(1, cloudCount_)); if (arrivalOrder_.size() < target) return false; const long double newest = requests_.at(arrivalOrder_.back()).arrivalTime; size_t sameFrame = 0; for (auto it = arrivalOrder_.rbegin(); it != arrivalOrder_.rend(); ++it) { if (fabsl(requests_.at(*it).arrivalTime - newest) > 1.0e-12L) break; ++sameFrame; } return sameFrame >= target; } bool pilotIsUseful() const { if (longDecodeObserved() || shortDecodeObserved() || latencyRelevantDeepBurstObserved()) { return false; } if (throughputWeight_ >= 0.90L || sloTokenGap_ <= 0 || bandwidth_ <= 0) { return false; } const long double oneTokenTransfer = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); return oneTokenTransfer >= 0.10L * sloTokenGap_; } int remotePopulation(int remote) const { int count = 0; for (RequestId id : arrivalOrder_) { auto it = requests_.find(id); if (it == requests_.end()) continue; const Request& request = it->second; if (request.remote == remote && request.phase != Phase::Finished) ++count; } return count; } int remoteInputClassAssignments(int remote, long long inputLength) const { int count = 0; for (RequestId id : arrivalOrder_) { const Request& request = requests_.at(id); if (request.remote == remote && request.inputLength == inputLength) { ++count; } } return count; } int inputClassAffinityChunk() const { if (curves_[4].empty()) return 1; return max(1, static_cast(llround( 0.30L * curves_[4].back().first))); } int pilotTargetPerRemote() { if (pilotTargetLatched_ > 0) return pilotTargetLatched_; const int remotes = max(1, cloudCount_); const int evidenceTargetPerRemote = max(1, (evidenceTarget() + remotes - 1) / remotes); const int observedCapacity = max(1, (static_cast(arrivalOrder_.size()) + remotes - 1) / remotes); pilotTargetLatched_ = min(evidenceTargetPerRemote, observedCapacity); return pilotTargetLatched_; } bool pilotNeedsAdmission() { const int target = pilotTargetPerRemote(); for (int cloud = 0; cloud < cloudCount_; ++cloud) { if (remotePopulation(cloud) < target) return true; } return false; } long double estimatedRemoteLoad(int remote) const { long double load = cloudBusy_[remote] ? scheduleCost_ : 0; for (RequestId id : arrivalOrder_) { auto it = requests_.find(id); if (it == requests_.end()) continue; const Request& request = it->second; if (request.remote != remote || request.phase == Phase::Finished) continue; if (isPrefillComputePhase(request.phase)) { int remainingLayers = max(0, layerCount_ - request.nextLayer); load += request.prefillProcessTime * remainingLayers / max(1, layerCount_); load += scheduleCost_; } else if (isDecodePhase(request.phase)) { // Output length is intentionally unknown. A short rolling horizon // balances persistent decode streams without pretending to predict it. load += 3 * decodeProcessSingletonTime_ + scheduleCost_; } } return load; } bool edgeLimitedFlatDecodeBurst() const { if (cloudCount_ <= 1 || bandwidth_ <= 0 || sloTokenGap_ <= 0 || throughputWeight_ < waitingWeight_ || staggeredArrivalsObserved() || curves_[4].empty() || arrivalOrder_.size() < kFlatDecodeBurstPerCloud * static_cast(cloudCount_) || curves_[4].back().first + 1.0e-12L < static_cast(arrivalOrder_.size())) { return false; } long double minimumDecode = numeric_limits::infinity(); long double maximumDecode = 0; for (const auto& point : curves_[4]) { minimumDecode = min(minimumDecode, point.second); maximumDecode = max(maximumDecode, point.second); } if (maximumDecode - minimumDecode > 1.0e-12L * max(1.0L, maximumDecode)) { return false; } long double edgePromptWork = 0; long double singleCloudPromptWork = 0; for (RequestId id : arrivalOrder_) { const Request& request = requests_.at(id); const long double edgeAdmissionWork = scheduleCost_ + interpolatedDuration(request.inputLength, 0); const long double requestEdgeWork = 2.0L * scheduleCost_ + interpolatedDuration(request.inputLength, 0) + interpolatedDuration(request.inputLength, 2); const long double requestCloudWork = scheduleCost_ + request.prefillProcessTime; const long double promptTransfer = networkLatency_ + 8.0L * request.inputLength * bytesPerToken_ / (bandwidth_ * 1.0e6L); if (edgeAdmissionWork + 1.0e-12L < requestCloudWork || requestEdgeWork + 1.0e-12L < requestCloudWork || 2.0L * promptTransfer > 0.10L * requestEdgeWork) { return false; } edgePromptWork += requestEdgeWork; singleCloudPromptWork += requestCloudWork; } const long double oneTokenTransfer = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); return edgePromptWork >= singleCloudPromptWork && oneTokenTransfer <= 0.10L * sloTokenGap_; } vector publicRankPipelineClasses() const { if (cloudCount_ != 2 || layerCount_ != 1 || waitingWeight_ <= 0 || throughputWeight_ < 2.0L * waitingWeight_ || distanceBase_ <= 0 || sloTokenGap_ <= 0 || bandwidth_ <= 0 || !staggeredArrivalsObserved() || curves_[1].empty() || curves_[4].empty() || curves_[4].back().first > 8.0L * cloudCount_) { return {}; } vector classes; unordered_map visibleClassCounts; classes.reserve(arrivalOrder_.size()); for (RequestId id : arrivalOrder_) { const long long inputLength = requests_.at(id).inputLength; classes.push_back(inputLength); ++visibleClassCounts[inputLength]; } sort(classes.begin(), classes.end()); classes.erase(unique(classes.begin(), classes.end()), classes.end()); if (classes.size() != 5) return {}; // The public-rank partition assumes that the five visible input // classes are being generated symmetrically. Stop using it as soon // as the observed prefix contradicts that assumption; this uses only // ARR events already received by the scheduler. int minimumClassCount = numeric_limits::max(); int maximumClassCount = 0; for (long long input : classes) { const int count = visibleClassCounts.at(input); minimumClassCount = min(minimumClassCount, count); maximumClassCount = max(maximumClassCount, count); } if (maximumClassCount - minimumClassCount > 1) return {}; // The partition is useful only when public prompt compute cannot itself // distinguish the five classes. Otherwise ordinary live-load // placement retains the information carried by the prompt curve. const long double reference = interpolatedDuration(classes.front(), 1); for (long long input : classes) { if (fabsl(interpolatedDuration(input, 1) - reference) > 1.0e-12L * max(1.0L, fabsl(reference))) { return {}; } } return classes; } int publicRankPartitionRemote(long long inputLength) const { const vector classes = publicRankPipelineClasses(); if (classes.empty()) return -1; const auto found = lower_bound(classes.begin(), classes.end(), inputLength); if (found == classes.end() || *found != inputLength) return -1; const int rank = static_cast(found - classes.begin()); // Two adjacent middle ranks form the decode lane, while the two // complementary ranks form the prompt lane. Rank zero is the public // balancing class: it always goes to the currently smaller lane, with // a prompt-lane tie break. Before all five ranks are visible this // function returns -1 and preserves the carrier's pilot assignments. if (rank == 2 || rank == 3) return 0; if (rank == 1 || rank == 4) return 1; const int onZero = remotePopulation(0); const int onOne = remotePopulation(1); return onZero < onOne ? 0 : 1; } int chooseRemote(long long inputLength) { if (cloudCount_ <= 0) return -1; if (edgeLimitedFlatDecodeBurst()) return 0; const int publicPartition = publicRankPartitionRemote(inputLength); if (publicPartition >= 0) return publicPartition; if (throughputWeight_ >= 4.0L * waitingWeight_ && !curves_[1].empty() && !curves_[4].empty() && curves_[4].back().first + 1.0e-12L >= cloudCount_ + 1 && curves_[4].back().first <= kRankPriorMaxBatchMultiplier * cloudCount_) { set visibleClassSet; unordered_map visibleClassCount; for (RequestId id : arrivalOrder_) { const long long length = requests_.at(id).inputLength; visibleClassSet.insert(length); ++visibleClassCount[length]; } const int classCount = static_cast(visibleClassSet.size()); int minimumClassCount = numeric_limits::max(); int maximumClassCount = 0; for (long long length : visibleClassSet) { minimumClassCount = min(minimumClassCount, visibleClassCount[length]); maximumClassCount = max(maximumClassCount, visibleClassCount[length]); } const bool balancedVisibleClasses = 100LL * maximumClassCount <= 1LL * kRankPriorMaxClassRatioPercent * minimumClassCount; const bool enoughClassEvidence = all_of( visibleClassSet.begin(), visibleClassSet.end(), [&](long long length) { return length <= curves_[1].front().first + 1.0e-12L && visibleClassCount[length] >= kRankPriorEvidencePerRemote * cloudCount_; }); if (classCount >= 4 && cloudCount_ >= classCount && balancedVisibleClasses && enoughClassEvidence) { vector classes(visibleClassSet.begin(), visibleClassSet.end()); const int rank = static_cast(lower_bound( classes.begin(), classes.end(), inputLength) - classes.begin()); vector poolSizes(classCount, 1); vector remoteOrder; for (int value = 2; value < classCount; ++value) { remoteOrder.push_back(value); } const int priorRankCount = static_cast(remoteOrder.size()); for (int extra = 0; extra < cloudCount_ - classCount; ++extra) { ++poolSizes[remoteOrder[extra % priorRankCount]]; } // Ranks 0 and 1 have one remote each. Spare capacity cycles // over ranks 2..m-1; appending the low ranks is deterministic // symmetry breaking for the contiguous remote pools. remoteOrder.push_back(0); remoteOrder.push_back(1); int totalAssigned = 0; for (int cloud = 0; cloud < cloudCount_; ++cloud) { totalAssigned += remoteInputClassAssignments(cloud, inputLength); } const bool usePrior = (1LL * totalAssigned * kRankPriorStride) % 100 < kRankPriorPercent; if (usePrior) { int firstRemote = 0; for (int value : remoteOrder) { if (value == rank) break; firstRemote += poolSizes[value]; } int assignedInPool = 0; for (int offset = 0; offset < poolSizes[rank]; ++offset) { assignedInPool += remoteInputClassAssignments( firstRemote + offset, inputLength); } return firstRemote + assignedInPool % poolSizes[rank]; } } } // When throughput strongly dominates, keep a short, curve-scaled run // of equal public input classes on one remote. A moderate fraction of // the largest measured DPROC batch preserves several complete cohorts // while letting normal live-load balancing open another remote often // enough. The public curve guard prevents oversized chunks when the // measured batching range is large relative to K. if (throughputWeight_ >= 4.0L * waitingWeight_ && !curves_[4].empty()) { const int chunk = inputClassAffinityChunk(); if (chunk <= 2 * cloudCount_) { for (int offset = 0; offset < cloudCount_; ++offset) { const int cloud = (nextRemoteTieBreak_ + offset) % cloudCount_; const int assigned = remoteInputClassAssignments(cloud, inputLength); if (assigned > 0 && assigned % chunk != 0) { nextRemoteTieBreak_ = (cloud + 1) % cloudCount_; return cloud; } } } } vector load(cloudCount_); const bool restrictToPilotDeficit = pilotIsUseful(); const int pilotTarget = pilotTargetPerRemote(); long double best = numeric_limits::infinity(); for (int cloud = 0; cloud < cloudCount_; ++cloud) { load[cloud] = estimatedRemoteLoad(cloud); if (restrictToPilotDeficit && remotePopulation(cloud) >= pilotTarget) continue; best = min(best, load[cloud]); } // Rotate equal choices; all comparisons use public, live state only. const long double epsilon = 1e-12L; for (int offset = 0; offset < cloudCount_; ++offset) { int cloud = (nextRemoteTieBreak_ + offset) % cloudCount_; if (restrictToPilotDeficit && remotePopulation(cloud) >= pilotTarget) continue; if (load[cloud] <= best + epsilon) { nextRemoteTieBreak_ = (cloud + 1) % cloudCount_; return cloud; } } return 0; } vector collect(Phase phase, int remote = -2) const { vector result; for (RequestId id : arrivalOrder_) { auto it = requests_.find(id); if (it == requests_.end()) continue; const Request& request = it->second; if (request.phase != phase) continue; if (remote != -2 && request.remote != remote) continue; result.push_back(id); } return result; } bool staggeredArrivalsObserved() const { if (arrivalOrder_.size() < 2) return false; const long double first = requests_.at(arrivalOrder_.front()).arrivalTime; const long double last = requests_.at(arrivalOrder_.back()).arrivalTime; return last > first + 1.0e-12L; } bool sparseSingletonStreamObserved() const { if (arrivalOrder_.size() < 3 || sloTokenGap_ <= 0) return false; const size_t n = arrivalOrder_.size(); const long double a = requests_.at(arrivalOrder_[n - 3]).arrivalTime; const long double b = requests_.at(arrivalOrder_[n - 2]).arrivalTime; const long double c = requests_.at(arrivalOrder_[n - 1]).arrivalTime; const long double minimumGap = 2.0L * sloTokenGap_; return b > a + 1.0e-12L && c > b + 1.0e-12L && b - a >= minimumGap && c - b >= minimumGap; } bool safePromptTailCohortPending() const { if (!safePromptTailDrainEngaged_) return false; for (RequestId id : arrivalOrder_) { const Request& request = requests_.at(id); if (request.arrivalOrder >= safePromptTailCohortLimit_) continue; if (request.phase >= Phase::ReadyPrefillPre && request.phase <= Phase::RunningPrefillPost) { return true; } } return false; } bool safeTwoCloudPromptTailOrder(const vector& ids) const { if (ids.size() <= 1 || cloudCount_ != 2 || throughputWeight_ <= 0 || waitingWeight_ < 4.0L * throughputWeight_ || bandwidth_ <= 0 || curves_[1].size() != 1 || !latencyRelevantDeepBurstObserved() || !staggeredArrivalsObserved() || tokenPosts_ != 0 || layerCount_ < 32 || arrivalOrder_.size() < static_cast(4 * max(1, layerCount_)) || sloTokenGap_ <= 0 || promptTransferDuration(1) > 0.10L * sloTokenGap_) { return false; } if (safePromptTailDrainEngaged_) { return all_of(ids.begin(), ids.end(), [&](RequestId id) { return requests_.at(id).arrivalOrder < safePromptTailCohortLimit_; }); } // Every prompt in the visible cohort has completed P PRE and its FIFO // upload. Hence every remaining cloud job is already exposed, and a // fresh decode cannot be needed to make any of this work arrive. if (!collect(Phase::ReadyPrefillPre).empty() || !collect(Phase::RunningPrefillPre).empty() || !collect(Phase::WaitingPrefillUpload).empty()) { return false; } for (RequestId id : arrivalOrder_) { const Request& request = requests_.at(id); if (isDecodePhase(request.phase) && request.phase != Phase::ReadyDecodePre) { return false; } } map visibleInputFrequencies; for (RequestId id : arrivalOrder_) { ++visibleInputFrequencies[requests_.at(id).inputLength]; } if (visibleInputFrequencies.size() < 2) return false; int minimumFrequency = numeric_limits::max(); int maximumFrequency = 0; for (const auto& [input, frequency] : visibleInputFrequencies) { (void)input; minimumFrequency = min(minimumFrequency, frequency); maximumFrequency = max(maximumFrequency, frequency); } if (minimumFrequency != maximumFrequency) return false; // The public cloud service is input-independent and strictly dominates // each job's known DOWN + P POST tail. Long-tail-first is Johnson's // late partition on this now-closed, already-visible prompt drain. const long double compute = curves_[1].front().second; for (RequestId id : ids) { const Request& request = requests_.at(id); const long double tail = promptTransferDuration(request.inputLength) + scheduleCost_ + interpolatedDuration(request.inputLength, 2); if (compute <= tail + 1.0e-12L) return false; } safePromptTailDrainEngaged_ = true; safePromptTailCohortLimit_ = nextArrivalOrder_; return true; } RequestId shortestPrefill(const vector& ids, int stage) const { // With at most one live request per cloud there is no prompt-compute // congestion to drain; preserving arrival order keeps the decode cohort // aligned. Once oversubscribed, shortest remaining prompt work reduces // mean P-POST completion time. const size_t initialFifoLimit = static_cast(max(0, cloudCount_)) * static_cast(layerCount_ == 1 ? 3 : 1); if (ids.size() <= 1 || arrivalOrder_.size() <= initialFifoLimit) { return ids.front(); } const bool extremeBacklog = stage == 0 && cloudCount_ > 0 && static_cast(ids.size()) > kSjfBacklogPerCloud * static_cast(cloudCount_); const bool latencyDominant = waitingWeight_ >= kSjfLatencyDominantWeight; const bool applySpreadGate = !extremeBacklog && !latencyDominant && (kSjfSpreadStageMask & (1 << max(0, min(stage, 2)))) != 0; if (applySpreadGate && waitingWeight_ < kSjfMinWaitingWeight) { if (stage == 0) sjfAdmissionSuppressed_ = true; return ids.front(); } RequestId bestId = ids.front(); long double bestCost = numeric_limits::infinity(); long double worstCost = 0; const bool safePromptTailOrder = stage == 1 && safeTwoCloudPromptTailOrder(ids); for (RequestId id : ids) { const Request& request = requests_.at(id); long double cost = 0; if (stage == 0) { const long double transfer = networkLatency_ + 8.0L * request.inputLength * bytesPerToken_ / (bandwidth_ * 1.0e6L); cost = interpolatedDuration(request.inputLength, 0) + request.prefillProcessTime + interpolatedDuration(request.inputLength, 2) + 2.0L * transfer; } else if (stage == 1) { cost = request.prefillProcessTime * max(0, layerCount_ - request.nextLayer) / max(1, layerCount_); if (safePromptTailOrder) { cost = -(promptTransferDuration(request.inputLength) + scheduleCost_ + interpolatedDuration(request.inputLength, 2)); } // With a single compute layer and a throughput-dominant // objective, include the prompt's guaranteed DOWN/P-POST tail // when choosing among ready cloud work. This is ordinary SJF // on the complete public prompt tail rather than compute alone. if (layerCount_ == 1 && throughputWeight_ > waitingWeight_ && bandwidth_ > 0) { const long double transfer = networkLatency_ + 8.0L * request.inputLength * bytesPerToken_ / (bandwidth_ * 1.0e6L); cost += transfer + interpolatedDuration(request.inputLength, 2); } } else { cost = interpolatedDuration(request.inputLength, 2); } if (cost < bestCost || (cost == bestCost && request.arrivalOrder < requests_.at(bestId).arrivalOrder)) { bestCost = cost; bestId = id; } worstCost = max(worstCost, cost); } const long double spread = worstCost - bestCost; const long double reference = max(1.0e-12L, sloFirstToken_); if (applySpreadGate && spread < kSjfSpreadSloFraction * reference) { if (stage == 0) sjfAdmissionSuppressed_ = true; return ids.front(); } return bestId; } bool latencyDeepFreshAlignmentEnabled() const { if (LATENCY_DEEP_FRESH_DPRE_TARGET <= 1 || bandwidth_ <= 0 || sloTokenGap_ <= 0 || staggeredArrivalsObserved() || !latencyRelevantDeepBurstObserved() || throughputWeight_ <= 0) { return false; } const long double oneTokenTransfer = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); return oneTokenTransfer <= 0.10L * sloTokenGap_; } bool freshBurstAlignmentEnabled() const { if (!latencyDeepFreshAlignmentEnabled() && (throughputWeight_ < 0.80L || waitingWeight_ > 0.15L)) { return false; } const long long minimumEvidence = max(16, 4LL * max(1, cloudCount_)); if (tokenPosts_ == 0) return true; if (tokenPosts_ < minimumEvidence) return false; const long double survivalPosterior = (tokenContinuations_ + 1.0L) / (tokenPosts_ + 2.0L); return survivalPosterior >= 0.90L; } void holdSparseFreshBursts(vector& ready) { if (ready.empty() || !freshBurstAlignmentEnabled()) return; const bool hasFreshReady = any_of(ready.begin(), ready.end(), [&](RequestId id) { return requests_.at(id).tokensProduced == 0; }); if (!hasFreshReady) return; struct BurstState { int arrivals = 0; int freshReady = 0; bool prefillPending = false; }; map bursts; for (RequestId id : arrivalOrder_) { const Request& request = requests_.at(id); BurstState& burst = bursts[request.arrivalTime]; ++burst.arrivals; if (request.phase == Phase::ReadyDecodePre && request.tokensProduced == 0) { ++burst.freshReady; } if (isPrefillComputePhase(request.phase)) burst.prefillPending = true; } set heldArrivalTimes; const int minimumBurst = max(8, 2 * max(1, cloudCount_)); const int resourceTarget = max(2, 2 * max(1, cloudCount_)); for (const auto& [arrivalTime, burst] : bursts) { const int fractionTarget = max(2, (burst.arrivals + 5) / 6); const int releaseTarget = latencyDeepFreshAlignmentEnabled() ? min(burst.arrivals, LATENCY_DEEP_FRESH_DPRE_TARGET) : min(fractionTarget, resourceTarget); if (burst.arrivals >= minimumBurst && burst.prefillPending && burst.freshReady < releaseTarget) { heldArrivalTimes.insert(arrivalTime); } } const size_t before = ready.size(); erase_if(ready, [&](RequestId id) { const Request& request = requests_.at(id); return request.tokensProduced == 0 && heldArrivalTimes.count(request.arrivalTime) != 0; }); if (ready.size() != before) freshBurstLevelingEngaged_ = true; } bool shouldHoldFreshDecodePre(const vector& ready) const { // If the public FIFO link state proves that prompt uploads will occupy // UP beyond the completion of an already-issued prompt DOWN + P POST, // delaying the first D PRE cannot create an UP bubble. Admit that // guaranteed prompt into the fresh cohort and save one later transfer // setup. This branch never waits for an ARR or for unissued work. const bool singleObservedDeepWave = !staggeredArrivalsObserved() && deepArrivalBurstObserved() && throughputWeight_ > waitingWeight_ && waitingWeight_ > 0 && tokenPosts_ == 0; if (!ready.empty() && singleObservedDeepWave && collect(Phase::ReadyPrefillPre).empty() && !collect(Phase::WaitingPrefillUpload).empty()) { const bool allDecodeWorkIsFreshReady = all_of(ready.begin(), ready.end(), [&](RequestId id) { const Request& request = requests_.at(id); return request.tokensProduced == 0; }) && all_of(arrivalOrder_.begin(), arrivalOrder_.end(), [&](RequestId id) { const Phase phase = requests_.at(id).phase; return !isDecodePhase(phase) || phase == Phase::ReadyDecodePre; }); long double nextPromptPostDone = numeric_limits::infinity(); if (allDecodeWorkIsFreshReady) { for (RequestId id : collect(Phase::WaitingPrefillDownload)) { const Request& request = requests_.at(id); nextPromptPostDone = min(nextPromptPostDone, request.prefillDownloadFinishTime + scheduleCost_ + interpolatedDuration(request.inputLength, 2)); } } const long double joinedDecodePreDone = nextPromptPostDone + scheduleCost_ + interpolatedDuration( static_cast(ready.size() + 1), 3); if (joinedDecodePreDone + 1.0e-12L < promptUpAvailableTime_) { return true; } } // In an observed mixed waiting-dominant stream, the already-issued // prompt pipeline guarantees a nearby event. Briefly forming a wider // first-token cohort then avoids a fragmented Edge transfer without // waiting for an unknown future arrival. const bool mixedWaitingStream = staggeredArrivalsObserved() && throughputWeight_ >= 0.10L && throughputWeight_ < waitingWeight_; const bool waitingOnlyStream = staggeredArrivalsObserved() && throughputWeight_ <= 0 && waitingWeight_ > 0; const bool objectiveAllowsHold = throughputWeight_ >= 0.80L || mixedWaitingStream || waitingOnlyStream; if (ready.empty() || freshBurstLevelingEngaged_ || !objectiveAllowsHold || sloTokenGap_ <= 0 || bandwidth_ <= 0) { return false; } const bool allFresh = all_of(ready.begin(), ready.end(), [&](RequestId id) { return requests_.at(id).tokensProduced == 0; }); if (!allFresh) return false; // Once every prompt assigned to the represented remotes has drained, // decode can overlap the tail of other remotes without delaying local // prompt compute. The ready cohort and the absence of unassigned prompts // are both visible; no prediction of a future arrival or output length is // involved. const long double oneTokenNetwork = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); const bool throughputDominantDrainedRemote = waitingWeight_ > 0 && throughputWeight_ >= 4.0L * waitingWeight_; if ((mixedWaitingStream || throughputDominantDrainedRemote) && oneTokenNetwork <= 0.10L * sloTokenGap_ && collect(Phase::ReadyPrefillPre).empty() && all_of(ready.begin(), ready.end(), [&](RequestId id) { return !remoteHasPrefillWork(requests_.at(id).remote); })) { return false; } // A first token has no active TPOT gap clock, but holding it too long can // phase-shift later rounds into avoidable endpoint contention. Bound the // waiting-only cohort delay by two public gap-SLO intervals, measured // exclusively // from visible P POST completion events. if (waitingOnlyStream) { long double oldestReadyAge = 0; for (RequestId id : ready) { oldestReadyAge = max(oldestReadyAge, currentTime_ - requests_.at(id).decodePreReadyTime); } if (oldestReadyAge >= 2.0L * sloTokenGap_) return false; } // For a waiting-only objective, hold fresh decode admission only on an // actively running prompt computation: its completion is a concrete, // near-term cohort event. A queued upload, merely ready computation, or // prompt download can delay the first token without contributing any // throughput score. Objectives that value throughput retain the broader // four-phase cohort-alignment rule. const unsigned prefillPhaseMask = waitingOnlyStream ? 4U : 15U; const bool prefillStillAdvancing = ((prefillPhaseMask & 1U) != 0 && !collect(Phase::WaitingPrefillUpload).empty()) || ((prefillPhaseMask & 2U) != 0 && !collect(Phase::ReadyPrefillProcess).empty()) || ((prefillPhaseMask & 4U) != 0 && !collect(Phase::RunningPrefillProcess).empty()) || ((prefillPhaseMask & 8U) != 0 && !collect(Phase::WaitingPrefillDownload).empty()); const long double oneTokenTransfer = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); return prefillStillAdvancing && oneTokenTransfer <= 0.10L * sloTokenGap_; } bool hasDecodeWorkForRemote(int remote) const { for (RequestId id : arrivalOrder_) { auto it = requests_.find(id); if (it == requests_.end()) continue; const Request& request = it->second; if (request.remote == remote && isDecodePhase(request.phase)) return true; } return false; } bool remoteHasPrefillWork(int remote) const { for (RequestId id : arrivalOrder_) { const Request& request = requests_.at(id); if (request.remote == remote && isPrefillComputePhase(request.phase)) { return true; } } return false; } void restrictResidualCanonicalFirstDecode(vector& ready) const { #if CASE12_RESIDUAL_CANONICAL_FIRST_DECODE if (CASE12_RESIDUAL_AGGRESSIVE_LATER_ORDER == 0 || !residualAggressiveLaterEngaged_ || ready.empty() || cloudCount_ != 1 || throughputWeight_ < 0.95L || waitingWeight_ > 0.05L || distanceBase_ <= 0 || bandwidth_ <= 0) { return; } map> arrivalCohorts; for (RequestId id : arrivalOrder_) { arrivalCohorts[requests_.at(id).arrivalTime].push_back(id); } if (arrivalCohorts.empty()) return; const vector& first = arrivalCohorts.begin()->second; map> firstByInput; for (RequestId id : first) firstByInput[requests_.at(id).inputLength].push_back(id); if (firstByInput.size() != 4) return; int minimum = numeric_limits::max(); int maximum = 0; int minimumClasses = 0; for (const auto& [inputLength, ids] : firstByInput) { (void)inputLength; minimum = min(minimum, static_cast(ids.size())); maximum = max(maximum, static_cast(ids.size())); } for (const auto& [inputLength, ids] : firstByInput) { (void)inputLength; if (static_cast(ids.size()) == minimum) ++minimumClasses; } if (minimum != 1 || maximum != 2 || minimumClasses != 1) return; // If the shortest class is the deficit, the ordinary snake is already // canonical for the first-response/first-decode boundary. Rebatching // that boundary can move a late decode transfer without buying TDR. if (firstByInput.begin()->second.size() == 1) return; vector inputs; for (const auto& [inputLength, ids] : firstByInput) { (void)ids; inputs.push_back(inputLength); } auto cohort = next(arrivalCohorts.begin()); if (cohort == arrivalCohorts.end()) return; map> secondByInput; for (RequestId id : cohort->second) { secondByInput[requests_.at(id).inputLength].push_back(id); } if (secondByInput.size() != inputs.size()) return; for (long long inputLength : inputs) { if (secondByInput[inputLength].size() != 2) return; } // The prompt order is 0,1,1,2,2,3,3,0. Admit its first decode in // the original snake's already-issued pair windows, 0,1 / 2,3 / // 3,2 / 1,0. This retains the prompt TDR gain without exposing // downstream grouping to unknown output lengths. vector> groups = { {secondByInput[inputs[0]][0], secondByInput[inputs[1]][0]}, {secondByInput[inputs[2]][0], secondByInput[inputs[3]][0]}, {secondByInput[inputs[3]][1], secondByInput[inputs[2]][1]}, {secondByInput[inputs[1]][1], secondByInput[inputs[0]][1]}, }; auto firstDecodeIssued = [&](RequestId id) { const Request& request = requests_.at(id); return request.tokensProduced > 0 || request.phase > Phase::ReadyDecodePre; }; set managed; for (const vector& group : groups) { managed.insert(group.begin(), group.end()); } for (const vector& group : groups) { if (all_of(group.begin(), group.end(), firstDecodeIssued)) continue; const bool complete = all_of(group.begin(), group.end(), [&](RequestId id) { return requests_.at(id).phase == Phase::ReadyDecodePre; }); erase_if(ready, [&](RequestId id) { const Request& request = requests_.at(id); if (request.tokensProduced != 0 || managed.count(id) == 0) { return false; } return !complete || find(group.begin(), group.end(), id) == group.end(); }); return; } #else (void)ready; #endif } bool deepArrivalBurstObserved() const { const size_t target = static_cast( max(1, PREFILL_UPLOAD_MIN_BURST_ROUNDS) * max(1, cloudCount_)); map counts; for (RequestId id : arrivalOrder_) { if (++counts[requests_.at(id).arrivalTime] >= target) return true; } return false; } int prefillUploadQueueLimit() const { if (PREFILL_UPLOAD_QUEUE_LIMIT > 0) { return max(1, PREFILL_UPLOAD_QUEUE_LIMIT); } // A sparse/staggered stream needs a single outstanding transfer so that // newly visible jobs remain reorderable. A deep burst already exposes // its prompt choices, so keep a small pipeline. When the public // objective values throughput at least twice as much as waiting, retain // three outstanding transfers per cloud; otherwise half a cloud-round // is enough to prevent avoidable channel bubbles. if (!deepArrivalBurstObserved()) { const size_t matureBacklog = static_cast(2 * max(1, cloudCount_)); const long double oneTokenTransfer = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); const bool materiallySerialized = sloTokenGap_ > 0 && oneTokenTransfer >= PREFILL_UPLOAD_NETWORK_SLO_RATIO * sloTokenGap_; if (materiallySerialized && staggeredArrivalsObserved() && collect(Phase::ReadyPrefillPre).size() >= matureBacklog) { prefillUploadPipelineEngaged_ = true; } return prefillUploadPipelineEngaged_ ? 2 : 1; } if (throughputWeight_ >= PREFILL_UPLOAD_DOMINANCE_RATIO * waitingWeight_) { return max(1, PREFILL_UPLOAD_LONG_ROUNDS) * max(1, cloudCount_) + max(0, PREFILL_UPLOAD_LONG_OFFSET); } return max(2, cloudCount_ / max(1, PREFILL_UPLOAD_BURST_DIVISOR)); } bool shouldHoldPrefillPreForUpload() const { const bool uploadEventGuaranteed = (((PREFILL_UPLOAD_WAITING_EVENT_MASK & 1) != 0) && !collect(Phase::WaitingPrefillUpload).empty()) || (((PREFILL_UPLOAD_WAITING_EVENT_MASK & 2) != 0) && !collect(Phase::WaitingDecodeUpload).empty()); if (PREFILL_UPLOAD_ADMISSION_GATE_MODE == 0 || waitingWeight_ <= 0 || throughputWeight_ >= PREFILL_UPLOAD_MAX_THROUGHPUT_WEIGHT || sloTokenGap_ <= 0 || bandwidth_ <= 0 || !uploadEventGuaranteed) { return false; } const size_t queuedUploads = (((PREFILL_UPLOAD_WAITING_EVENT_MASK & 1) != 0) ? collect(Phase::WaitingPrefillUpload).size() : 0U) + (((PREFILL_UPLOAD_WAITING_EVENT_MASK & 2) != 0) ? collect(Phase::WaitingDecodeUpload).size() : 0U); if (queuedUploads < static_cast(prefillUploadQueueLimit())) { return false; } // Reordering a FIFO upload queue is useful only after online evidence // that more choice can exist: either arrivals have become staggered, or // a genuinely deep same-frame burst is already visible. This avoids // idling a small one-shot workload merely because its first upload is // expensive. if (!staggeredArrivalsObserved() && !deepArrivalBurstObserved()) { return false; } const long double oneTokenTransfer = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); const long double networkRatio = oneTokenTransfer / sloTokenGap_; const bool materiallySerialized = networkRatio >= PREFILL_UPLOAD_NETWORK_SLO_RATIO; const bool negligiblyCheap = networkRatio <= 0.001L; if (!materiallySerialized && !negligiblyCheap) { return false; } if (PREFILL_UPLOAD_ADMISSION_GATE_MODE == 2 && longDecodeObserved()) { return false; } if (PREFILL_UPLOAD_ADMISSION_GATE_MODE == 3 && throughputWeight_ > waitingWeight_) { return false; } if (PREFILL_UPLOAD_ADMISSION_GATE_MODE == 4 && !shortDecodeObserved()) { return false; } return true; } bool serializedNetworkNeedsPrefillFairness() const { if (waitingWeight_ <= 0 || throughputWeight_ < 0.95L || sloTokenGap_ <= 0 || bandwidth_ <= 0) { return false; } const long double oneTokenTransfer = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); return oneTokenTransfer >= 0.25L * sloTokenGap_; } bool publicRankPipelineRegime() const { if (PUBLIC_RANK_PIPELINE_RUN < 0 || arrivalOrder_.size() < static_cast(8 * cloudCount_) || publicRankPipelineClasses().empty()) { return false; } long double shortestPrompt = numeric_limits::infinity(); for (const auto& point : curves_[1]) { shortestPrompt = min(shortestPrompt, point.second); } const long double measuredBatch = max(1.0L, curves_[4].back().first); const long double oneTokenTransfer = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); const long double decodeRound = 3.0L * scheduleCost_ + interpolatedDuration(measuredBatch, 3) + interpolatedDuration(measuredBatch, 4) + interpolatedDuration(measuredBatch, 5) + 2.0L * oneTokenTransfer; return oneTokenTransfer <= 0.10L * sloTokenGap_ && shortestPrompt >= 3.0L * decodeRound; } bool promptComputeMainlineActive() const { return !collect(Phase::WaitingPrefillUpload).empty() || !collect(Phase::ReadyPrefillProcess).empty() || !collect(Phase::RunningPrefillProcess).empty(); } int prefillChunkEnd(const Request& request) const { if ((SERIALIZED_PREEMPT_MODE & 1) == 0 || !serializedNetworkNeedsPrefillFairness()) { // The normal admission-first branch does not preempt prompt work at a // layer boundary, so splitting only adds another fixed schedule cost. return layerCount_; } const int remaining = layerCount_ - request.nextLayer; if (remaining <= 1 || !hasDecodeWorkForRemote(request.remote)) { return layerCount_; } const long double perLayer = max(0.0L, request.prefillProcessTime) / max(1, layerCount_); // A coarse two-gap quantum is a broad empirical plateau: it exposes // decode DOWN opportunities without paying setup cost every few layers. const long double target = max(scheduleCost_ * 2, sloTokenGap_ * 2.0L); const int layers = perLayer > 0 ? max(1, static_cast(floor(target / perLayer))) : remaining; return request.nextLayer + min(remaining, layers); } string makeDecodeBatch(const string& server, const string& stage, int remote, const vector& ids) const { ostringstream out; out << server << " D " << stage << ' ' << remote << ' ' << ids.size(); for (RequestId id : ids) out << ' ' << id; return out.str(); } void scheduleCloud(int cloud, vector& assignments) { if (cloudBusy_[cloud]) return; vector decode = collect(Phase::ReadyDecodeProcess, cloud); vector prefill = collect(Phase::ReadyPrefillProcess, cloud); if (decode.empty() && prefill.empty()) return; const bool publicKneeCohort = publicDecodeKneeEligible(static_cast(decode.size())); const bool guaranteedPrefillForCloud = PUBLIC_DECODE_EARLY_PROJECT_PREFILL && completedDecodeRequests_ >= PUBLIC_DECODE_EARLY_PROJECT_MIN_FINISHES && any_of(arrivalOrder_.begin(), arrivalOrder_.end(), [&](RequestId id) { const Request& request = requests_.at(id); return request.phase != Phase::Finished && request.phase < Phase::ReadyDecodePre && earlyProjectedPrefillEligible(request) && (request.remote == cloud || cloudCount_ == 1); }); if (publicKneeCohort && prefill.empty() && !decode.empty() && (decodeRoundStillReturning(cloud) || (decode.size() < static_cast(strongPublicDecodeKnee()) && guaranteedPrefillForCloud))) { // A deliberately retained ready remainder is useful only if it can // meet the already issued result coming back through DOWN/DPOST/ // DPRE. Waiting here is bounded by visible work; it never waits for // an unknown arrival or an unobserved continuation. return; } // With a pure-throughput objective and negligible communication, all // decode work already visible upstream of this cloud is a guaranteed // near-term cohort. Draining it to D PROC removes avoidable cloud setups. const long double oneTokenNetwork = bandwidth_ > 0 ? networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L) : numeric_limits::infinity(); const bool upstreamDecodeForRemote = any_of( arrivalOrder_.begin(), arrivalOrder_.end(), [&](RequestId id) { const Request& request = requests_.at(id); return request.remote == cloud && (request.phase == Phase::ReadyDecodePre || request.phase == Phase::WaitingDecodeUpload); }); if (waitingWeight_ <= 0 && throughputWeight_ > 0 && sloTokenGap_ > 0 && oneTokenNetwork <= 0.10L * sloTokenGap_ && tokenPosts_ >= PURE_TP_CLOUD_COHORT_MIN_POSTS && (tokenContinuations_ + 1.0L) / (tokenPosts_ + 2.0L) >= PURE_TP_CLOUD_COHORT_POSTERIOR && prefill.empty() && !decode.empty() && upstreamDecodeForRemote) { return; } // Decode gets low-gap service, but a bounded streak guarantees prefill // progress even under a permanent generation stream. bool runPrefill = !prefill.empty(); if ((SERIALIZED_PREEMPT_MODE & 2) != 0 && serializedNetworkNeedsPrefillFairness() && !decode.empty()) { constexpr int kDecodeStreakLimit = 8; runPrefill = !prefill.empty() && cloudDecodeStreak_[cloud] >= kDecodeStreakLimit; } const bool runPublicRankPipeline = cloud == 0 && publicRankPipelineRegime() && !prefill.empty() && !decode.empty() && cloudPromptRunsWithReadyDecode_[cloud] >= PUBLIC_RANK_PIPELINE_RUN; if (runPublicRankPipeline) runPrefill = false; if (!runPrefill) { if (publicKneeCohort) { const int knee = strongPublicDecodeKnee(); const int measuredMaximum = max(1, static_cast( floor(curves_[4].back().first + 1.0e-12L))); const int reservoirThreshold = max(1, measuredMaximum * PUBLIC_DECODE_RESERVOIR_PERCENT / 100 + PUBLIC_DECODE_RESERVOIR_BONUS); if (decode.size() >= static_cast(reservoirThreshold)) { publicDecodeReservoirFilled_ = true; } else if (decode.size() + PUBLIC_DECODE_RESERVOIR_HYSTERESIS <= static_cast(reservoirThreshold)) { publicDecodeReservoirFilled_ = false; } stable_sort(decode.begin(), decode.end(), [&](RequestId left, RequestId right) { const Request& a = requests_.at(left); const Request& b = requests_.at(right); if (a.tokensProduced != b.tokensProduced) { return a.tokensProduced < b.tokensProduced; } #if PUBLIC_DECODE_NEWEST_TIE return a.arrivalOrder > b.arrivalOrder; #else return a.arrivalOrder < b.arrivalOrder; #endif }); // First build one public-curve-sized survivor reservoir. That // gives the online max-min order room to align token progress; // afterwards the strongest whole-round knee drains work without // reverting to the inefficient descending tail. const int drainTarget = PUBLIC_DECODE_DRAIN_TARGET > 0 ? PUBLIC_DECODE_DRAIN_TARGET : knee; const int target = publicDecodeReservoirFilled_ ? drainTarget : publicDecodeReservoirBuildSize(knee); if (target > 0 && decode.size() > static_cast(target)) { decode.resize(static_cast(target)); } } assignments.push_back(makeDecodeBatch("C" + to_string(cloud), "PROC", cloud, decode)); for (RequestId id : decode) { requests_.at(id).decodeProcessStartTime = currentTime_; requests_.at(id).phase = Phase::RunningDecodeProcess; if (runPublicRankPipeline) { requests_.at(id).publicRankPipeline = true; } } cloudDecodeStreak_[cloud]++; cloudPromptRunsWithReadyDecode_[cloud] = 0; } else { Request& request = requests_.at(shortestPrefill(prefill, 1)); const int layerBegin = request.nextLayer; const int layerEnd = prefillChunkEnd(request); ostringstream out; out << 'C' << cloud << " P PROC " << layerBegin << ' ' << layerEnd << ' ' << cloud << ' ' << request.id; assignments.push_back(out.str()); request.runningLayerEnd = layerEnd; request.phase = Phase::RunningPrefillProcess; const long double layerFraction = static_cast(layerEnd - layerBegin) / max(1, layerCount_); cloudPrefillFinish_[cloud] = currentTime_ + scheduleCost_ + request.prefillProcessTime * layerFraction; cloudDecodeStreak_[cloud] = 0; if (!decode.empty() && publicRankPipelineRegime()) { ++cloudPromptRunsWithReadyDecode_[cloud]; } else { cloudPromptRunsWithReadyDecode_[cloud] = 0; } } cloudBusy_[cloud] = true; } void scheduleEdge(vector& assignments) { if (edgeBusy_) return; vector decodePost = collect(Phase::ReadyDecodePost); vector prefillPost = collect(Phase::ReadyPrefillPost); vector decodePre = collect(Phase::ReadyDecodePre); vector prefillPre = collect(Phase::ReadyPrefillPre); holdSparseFreshBursts(decodePre); const bool mixedWaitingStream = staggeredArrivalsObserved() && throughputWeight_ >= 0.10L && throughputWeight_ < waitingWeight_; const long double freshOneTokenNetwork = bandwidth_ > 0 ? networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L) : numeric_limits::infinity(); const bool throughputDominantDrainedRemote = waitingWeight_ > 0 && throughputWeight_ >= 4.0L * waitingWeight_; if ((mixedWaitingStream || throughputDominantDrainedRemote) && sloTokenGap_ > 0 && freshOneTokenNetwork <= 0.10L * sloTokenGap_ && prefillPre.empty() && !decodePre.empty()) { erase_if(decodePre, [&](RequestId id) { const Request& request = requests_.at(id); return request.tokensProduced == 0 && remoteHasPrefillWork(request.remote); }); } restrictResidualCanonicalFirstDecode(decodePre); if (safePromptTailCohortPending()) { // Decode begins only after the closed public prompt cohort has // drained, so changing its internal order cannot expose hidden // output survival to the scheduling decision. erase_if(decodePre, [&](RequestId id) { const Request& request = requests_.at(id); return request.tokensProduced == 0 && request.arrivalOrder < safePromptTailCohortLimit_; }); } // When throughput has zero value, batch efficiency before a first token // is unpriced. Staging fresh admissions one at a time offsets their // later decode waves and prevents synchronized endpoint contention once // TPOT clocks become active. As soon as any represented request has a // token, normal compatible batching resumes. const bool pureWaitingFresh = staggeredArrivalsObserved() && throughputWeight_ <= 0 && waitingWeight_ > 0 && !decodePre.empty() && all_of(decodePre.begin(), decodePre.end(), [&](RequestId id) { return requests_.at(id).tokensProduced == 0; }); if (pureWaitingFresh && decodePre.size() > 1) decodePre.resize(1); if (decodePost.empty() && prefillPost.empty() && decodePre.empty() && prefillPre.empty()) return; // In an observed sparse stream, a request that has already produced a // token owns an active TPOT clock. Protect only that known-aged work; // fresh requests remain governed by the normal admission policy. const bool protectAged = waitingWeight_ > throughputWeight_ && sparseSingletonStreamObserved(); vector agedPost; vector agedPre; copy_if(decodePost.begin(), decodePost.end(), back_inserter(agedPost), [this](RequestId id) { return requests_.at(id).tokensProduced > 0; }); copy_if(decodePre.begin(), decodePre.end(), back_inserter(agedPre), [this](RequestId id) { return requests_.at(id).tokensProduced > 0; }); if (protectAged && !agedPost.empty()) { assignments.push_back(makeDecodeBatch("E", "POST", -1, agedPost)); for (RequestId id : agedPost) requests_.at(id).phase = Phase::RunningDecodePost; ++edgeDecodeStreak_; edgeBusy_ = true; return; } if (protectAged && !agedPre.empty()) { assignments.push_back(makeDecodeBatch("E", "PRE", -1, agedPre)); for (RequestId id : agedPre) requests_.at(id).phase = Phase::RunningDecodePre; ++edgeDecodeStreak_; edgeBusy_ = true; return; } // Close an already-issued public overlap wave while prompt compute is // still active. POST uses only the marker attached at D PROC issue; // another PRE is possible only after the visible continuation event. vector pipelinePost; vector pipelinePre; copy_if(decodePost.begin(), decodePost.end(), back_inserter(pipelinePost), [this](RequestId id) { return requests_.at(id).publicRankPipeline; }); copy_if(decodePre.begin(), decodePre.end(), back_inserter(pipelinePre), [this](RequestId id) { return requests_.at(id).publicRankPipeline; }); if (publicRankPipelineRegime() && promptComputeMainlineActive() && !pipelinePost.empty()) { assignments.push_back(makeDecodeBatch("E", "POST", -1, pipelinePost)); for (RequestId id : pipelinePost) { requests_.at(id).phase = Phase::RunningDecodePost; } ++edgeDecodeStreak_; edgeBusy_ = true; return; } if (publicRankPipelineRegime() && promptComputeMainlineActive() && !pipelinePre.empty()) { assignments.push_back(makeDecodeBatch("E", "PRE", -1, pipelinePre)); for (RequestId id : pipelinePre) { requests_.at(id).phase = Phase::RunningDecodePre; } ++edgeDecodeStreak_; edgeBusy_ = true; return; } // Establish one small balanced pilot before making the endpoint admission // policy depend on observed decode survival. Once sustained decode is // actually observed, the original admission-first endpoint resumes. const bool admitPrefill = !pilotIsUseful() || pilotNeedsAdmission(); const long double oneTokenNetwork = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); const bool mixedWaitingPostSync = !decodePost.empty() && oneTokenNetwork <= 0.10L * sloTokenGap_ && throughputWeight_ >= PIPELINE_POST_SYNC_MIN_THROUGHPUT_WEIGHT && throughputWeight_ < waitingWeight_; const long long measuredDecodeBatch = curves_[4].empty() ? 1 : max(1LL, llround(curves_[4].back().first)); const bool finalPromptPostOnly = prefillPost.size() == 1 && prefillPre.empty() && collect(Phase::RunningPrefillPre).empty() && collect(Phase::WaitingPrefillUpload).empty() && collect(Phase::ReadyPrefillProcess).empty() && collect(Phase::RunningPrefillProcess).empty() && collect(Phase::WaitingPrefillDownload).empty(); const bool establishedDecodeCohort = !decodePre.empty() && all_of(decodePre.begin(), decodePre.end(), [&](RequestId id) { return requests_.at(id).tokensProduced > 0; }); // If the sole remaining prompt task competes with a large established // decode cohort, advance that cohort by one already-ready DPRE task. // The exchange waits only for work visible in this frame: it delays one // PPOST by one public endpoint task while releasing more than two full // measured DPROC batches whose TPOT clocks are already active. const bool largeDpreBeforeFinalPromptPost = finalPromptPostOnly && establishedDecodeCohort && waitingWeight_ > 0 && throughputWeight_ >= 5.0L * waitingWeight_ && sloTokenGap_ > 0 && oneTokenNetwork <= 0.10L * sloTokenGap_ && static_cast(decodePre.size()) >= 2LL * measuredDecodeBatch + 1; if (largeDpreBeforeFinalPromptPost) { if (shouldHoldFreshDecodePre(decodePre)) return; assignments.push_back(makeDecodeBatch("E", "PRE", -1, decodePre)); for (RequestId id : decodePre) { requests_.at(id).phase = Phase::RunningDecodePre; } ++edgeDecodeStreak_; } else if (!prefillPost.empty()) { Request& request = requests_.at(shortestPrefill(prefillPost, 2)); assignments.push_back("E P POST " + to_string(request.remote) + " " + to_string(request.id)); request.phase = Phase::RunningPrefillPost; edgeDecodeStreak_ = 0; } else if (waitingWeight_ <= 0 && throughputWeight_ > 0 && staggeredArrivalsObserved() && !decodePost.empty()) { // With no waiting-score term, close known decode results before // admitting another prompt. Delay this POST only when a D PROC is // actually running and can contribute a guaranteed compatible result; // other waiting/ready phases are too weak a reason to idle Edge. const long double oneTokenTransfer = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); const bool cheapNetwork = oneTokenTransfer <= 0.10L * sloTokenGap_; const bool runningDecodeWillAdvance = !collect(Phase::RunningDecodeProcess).empty(); const bool prefillStillAdvancing = !collect(Phase::WaitingPrefillUpload).empty() || !collect(Phase::ReadyPrefillProcess).empty() || !collect(Phase::RunningPrefillProcess).empty() || !collect(Phase::WaitingPrefillDownload).empty(); if (cheapNetwork && runningDecodeWillAdvance && !prefillStillAdvancing) return; assignments.push_back(makeDecodeBatch("E", "POST", -1, decodePost)); for (RequestId id : decodePost) requests_.at(id).phase = Phase::RunningDecodePost; ++edgeDecodeStreak_; } else if (waitingWeight_ <= 0 && throughputWeight_ > 0 && staggeredArrivalsObserved() && !decodePre.empty()) { // A running cloud decode or an already-issued DOWN transfer is a // guaranteed endpoint wake-up. In a communication-cheap, // sustained generation stream, briefly waiting for that visible // downstream work can merge the next D PRE cohort without relying // on an unknown arrival. Deliberately exclude UP/Ready-DPROC: the // cloud-side cohort gate may be waiting for this very D PRE wave. const bool downstreamDecodeWillAdvance = !collect(Phase::RunningDecodeProcess).empty() || !collect(Phase::WaitingDecodeDownload).empty(); if (tokenPosts_ >= PURE_TP_CLOUD_COHORT_MIN_POSTS && (tokenContinuations_ + 1.0L) / (tokenPosts_ + 2.0L) >= PURE_TP_CLOUD_COHORT_POSTERIOR && sloTokenGap_ > 0 && freshOneTokenNetwork <= 0.10L * sloTokenGap_ && decodePre.size() == 1 && downstreamDecodeWillAdvance) { return; } if (shouldHoldFreshDecodePre(decodePre)) return; assignments.push_back(makeDecodeBatch("E", "PRE", -1, decodePre)); for (RequestId id : decodePre) requests_.at(id).phase = Phase::RunningDecodePre; ++edgeDecodeStreak_; } else if (!prefillPre.empty() && admitPrefill) { // In a waiting-relevant deep wave, reserve Edge for a prompt POST // whose DOWN transfer or cloud completion is already guaranteed. // This waits only for issued work, never for an unknown arrival. const bool waitingDeepWave = waitingWeight_ > throughputWeight_ && latencyRelevantDeepBurstObserved(); const bool allCloudsBusy = !cloudBusy_.empty() && all_of(cloudBusy_.begin(), cloudBusy_.end(), [](bool busy) { return busy; }); const bool waitingProtectedStaggeredWave = waitingWeight_ > throughputWeight_ && staggeredArrivalsObserved() && allCloudsBusy; const bool protectIssuedPromptDown = waitingDeepWave || ((STAGGERED_PREFILL_POST_PROTECTION_MODE & 1) != 0 && waitingProtectedStaggeredWave); if (protectIssuedPromptDown && !collect(Phase::WaitingPrefillDownload).empty()) { return; } long double shortestPrefillPreTask = numeric_limits::infinity(); for (RequestId id : prefillPre) { shortestPrefillPreTask = min( shortestPrefillPreTask, scheduleCost_ + interpolatedDuration( requests_.at(id).inputLength, 0)); } bool guaranteedPromptFinishWouldBeCrossed = false; bool queuedPromptFinishWouldBeCrossed = false; for (int cloud = 0; cloud < cloudCount_; ++cloud) { const long double finish = cloudPrefillFinish_[cloud]; const bool crossed = finish > currentTime_ + 1.0e-12L && finish < currentTime_ + shortestPrefillPreTask - 1.0e-12L; guaranteedPromptFinishWouldBeCrossed |= crossed; queuedPromptFinishWouldBeCrossed |= crossed && !collect(Phase::ReadyPrefillProcess, cloud).empty(); } const bool protectCrossedPromptFinish = waitingDeepWave && guaranteedPromptFinishWouldBeCrossed; const bool protectQueuedStaggeredFinish = (STAGGERED_PREFILL_POST_PROTECTION_MODE & 2) != 0 && waitingProtectedStaggeredWave && queuedPromptFinishWouldBeCrossed; if (protectCrossedPromptFinish || protectQueuedStaggeredFinish) { return; } // Do not commit another prompt to a known-busy FIFO upload channel // when communication itself consumes a material token-gap budget. // The outstanding upload guarantees the next wake-up; deferring P PRE // preserves the option to select the shortest prompt among all work // visible then, without waiting for an unknown future arrival. if (shouldHoldPrefillPreForUpload()) return; set observedInputClasses; for (RequestId id : arrivalOrder_) { observedInputClasses.insert(requests_.at(id).inputLength); } const long long largestMeasuredDecodeBatch = curves_[4].empty() ? 1 : max(1LL, llround(curves_[4].back().first)); const long double affinityLifoReadyLimit = 2.0L * static_cast(observedInputClasses.size()) * static_cast(largestMeasuredDecodeBatch); const bool affinityPromptOrder = !curves_[4].empty() && throughputWeight_ >= 4.0L * waitingWeight_ && inputClassAffinityChunk() <= 2 * cloudCount_; const bool boundedAffinityLifo = static_cast(prefillPre.size()) <= affinityLifoReadyLimit; // Preserve FIFO while the visible prompt wave is very broad. Once // it fits within two measured decode cohorts per observed input // class, take the newest ready prompt to keep newly opened affinity // cohorts supplied. No future class or arrival is assumed. RequestId selectedPrefill = affinityPromptOrder ? (boundedAffinityLifo ? prefillPre.back() : prefillPre.front()) : shortestPrefill(prefillPre, 0); if (LATENCY_DEEP_PPRE_CLASS_CHUNK > 0 && latencyRelevantDeepBurstObserved() && prefillPre.size() > 1) { const long double cohortArrival = requests_.at(prefillPre.front()).arrivalTime; unordered_map cohortByInput; unordered_map admittedByInput; for (RequestId id : arrivalOrder_) { const Request& seen = requests_.at(id); if (fabsl(seen.arrivalTime - cohortArrival) <= 1.0e-12L) { ++cohortByInput[seen.inputLength]; if (seen.phase != Phase::ReadyPrefillPre) { ++admittedByInput[seen.inputLength]; } } } const int classCohortLimit = 2 * max(1, cloudCount_); const bool boundedClassCohort = all_of( cohortByInput.begin(), cohortByInput.end(), [&](const auto& entry) { return entry.second <= classCohortLimit; }); if (boundedClassCohort) { auto promptCost = [&](const Request& candidate) { const long double transfer = bandwidth_ > 0 ? networkLatency_ + 8.0L * candidate.inputLength * bytesPerToken_ / (bandwidth_ * 1.0e6L) : numeric_limits::infinity(); return interpolatedDuration(candidate.inputLength, 0) + candidate.prefillProcessTime + interpolatedDuration(candidate.inputLength, 2) + 2.0L * transfer; }; selectedPrefill = *min_element( prefillPre.begin(), prefillPre.end(), [&](RequestId left, RequestId right) { const Request& a = requests_.at(left); const Request& b = requests_.at(right); if (a.arrivalTime != b.arrivalTime) { return a.arrivalTime < b.arrivalTime; } const int roundA = admittedByInput[a.inputLength] / LATENCY_DEEP_PPRE_CLASS_CHUNK; const int roundB = admittedByInput[b.inputLength] / LATENCY_DEEP_PPRE_CLASS_CHUNK; if (roundA != roundB) return roundA < roundB; const long double costA = promptCost(a); const long double costB = promptCost(b); if (costA != costB) return costA < costB; return a.arrivalOrder < b.arrivalOrder; }); } } // With one cloud and an overwhelmingly throughput-oriented // objective, every visible prompt shares one serialized pipeline. // Rotate evenly across public input classes within the current // arrival cohort. Traverse class lengths in opposite directions // on alternating rounds, avoiding a full-range jump at each boundary. if (cloudCount_ == 1 && throughputWeight_ >= 0.95L && waitingWeight_ <= 0.05L && distanceBase_ > 0 && bandwidth_ > 0 && prefillPre.size() > 1) { const long double cohortArrival = requests_.at(prefillPre.front()).arrivalTime; const bool sameArrivalCohort = all_of( prefillPre.begin(), prefillPre.end(), [&](RequestId id) { return fabsl(requests_.at(id).arrivalTime - cohortArrival) <= 1.0e-12L; }); if (sameArrivalCohort) { unordered_map admittedByInput; unordered_map cohortFrequency; unordered_map priorFrequency; bool allPriorRequestsStillInPrompt = true; for (RequestId id : arrivalOrder_) { const Request& seen = requests_.at(id); if (fabsl(seen.arrivalTime - cohortArrival) <= 1.0e-12L) { ++cohortFrequency[seen.inputLength]; if (seen.phase != Phase::ReadyPrefillPre) { ++admittedByInput[seen.inputLength]; } } else if (seen.arrivalTime < cohortArrival) { ++priorFrequency[seen.inputLength]; allPriorRequestsStillInPrompt &= !isDecodePhase(seen.phase) && seen.phase != Phase::Finished; } } const auto hasSingleClassDeficit = [](const unordered_map& frequencies) { if (frequencies.size() < 4) return false; int minimum = numeric_limits::max(); int maximum = 0; int minimumClasses = 0; for (const auto& [inputLength, count] : frequencies) { (void)inputLength; minimum = min(minimum, count); maximum = max(maximum, count); } for (const auto& [inputLength, count] : frequencies) { (void)inputLength; if (count == minimum) ++minimumClasses; } return minimum == 1 && maximum == 2 && minimumClasses == 1; }; // An even class count keeps the two passes on the same // parity at the turn. Odd-width turns can change which // prompt becomes the next decode-ready singleton. bool currentHasTwoPerClass = cohortFrequency.size() >= 4 && cohortFrequency.size() % 2 == 0; bool repeatsKnownClasses = cohortFrequency.size() == priorFrequency.size(); for (const auto& [inputLength, count] : cohortFrequency) { currentHasTwoPerClass &= count == 2; repeatsKnownClasses &= priorFrequency.count(inputLength) != 0; } vector sortedInputClasses; sortedInputClasses.reserve(cohortFrequency.size()); long double largestTransfer = 0; for (const auto& [inputLength, count] : cohortFrequency) { (void)count; sortedInputClasses.push_back(inputLength); largestTransfer = max(largestTransfer, promptTransferDuration(inputLength)); } sort(sortedInputClasses.begin(), sortedInputClasses.end()); int admittedTotal = 0; for (const auto& [inputLength, count] : admittedByInput) { (void)inputLength; admittedTotal += count; } // When a public two-pass class snake reaches its turn, // sending the longest transfer twice in succession raises // mean prompt completion time without shortening the // serialized prompt tail. On a repeated, already-visible // class pattern, issue the next-shortest class once before // resuming the ordinary descending pass. const bool avoidRepeatedLongestAtTurn = SERIAL_PROMPT_SNAKE_TURN_SWAP != 0 && !priorFrequency.empty() && tokenPosts_ == 0 && allPriorRequestsStillInPrompt && hasSingleClassDeficit(priorFrequency) && currentHasTwoPerClass && repeatsKnownClasses && sloFirstToken_ > 0 && 2.0L * largestTransfer >= 0.50L * sloFirstToken_ && admittedTotal == static_cast(sortedInputClasses.size()); const long long turnInput = avoidRepeatedLongestAtTurn ? sortedInputClasses[sortedInputClasses.size() - 2] : 0; selectedPrefill = *min_element( prefillPre.begin(), prefillPre.end(), [&](RequestId left, RequestId right) { const Request& a = requests_.at(left); const Request& b = requests_.at(right); if (avoidRepeatedLongestAtTurn) { const bool turnA = a.inputLength == turnInput; const bool turnB = b.inputLength == turnInput; if (turnA != turnB) return turnA; } const int admittedA = admittedByInput[a.inputLength]; const int admittedB = admittedByInput[b.inputLength]; if (admittedA != admittedB) { return admittedA < admittedB; } if (a.inputLength != b.inputLength) { const bool descending = admittedA % 2 != 0; return descending ? a.inputLength > b.inputLength : a.inputLength < b.inputLength; } return a.arrivalOrder < b.arrivalOrder; }); #if CASE12_RESIDUAL_AGGRESSIVE_PROMPT_ORDER const bool materialSerializedPrompt = sloFirstToken_ > 0 && 2.0L * largestTransfer >= 0.50L * sloFirstToken_; long long deficientPriorInput = 0; for (const auto& [inputLength, count] : priorFrequency) { if (count == 1) deficientPriorInput = inputLength; } const bool aggressiveLaterOrder = !priorFrequency.empty() && cohortFrequency.size() == 4 && currentHasTwoPerClass && repeatsKnownClasses && hasSingleClassDeficit(priorFrequency) && deficientPriorInput != sortedInputClasses.front() && cohortArrival - requests_.at(arrivalOrder_.front()).arrivalTime >= sloTokenGap_ && tokenPosts_ == 0 && allPriorRequestsStillInPrompt && materialSerializedPrompt; if (aggressiveLaterOrder && CASE12_RESIDUAL_AGGRESSIVE_LATER_ORDER != 0 && admittedTotal >= 0 && admittedTotal < 8) { residualAggressiveLaterEngaged_ = true; static constexpr array kRanks = {0, 1, 1, 2, 2, 3, 3, 0}; const long long desiredInput = sortedInputClasses[kRanks[admittedTotal]]; auto chosen = min_element( prefillPre.begin(), prefillPre.end(), [&](RequestId left, RequestId right) { const Request& a = requests_.at(left); const Request& b = requests_.at(right); const bool desiredA = a.inputLength == desiredInput; const bool desiredB = b.inputLength == desiredInput; if (desiredA != desiredB) return desiredA; return a.arrivalOrder < b.arrivalOrder; }); if (chosen != prefillPre.end() && requests_.at(*chosen).inputLength == desiredInput) { selectedPrefill = *chosen; } } #endif } } if (distanceBase_ <= 0 && throughputWeight_ > 0 && prefillPre.size() > 1 && bandwidth_ > 0 && sloFirstToken_ > 0) { const long double cohortArrival = requests_.at(prefillPre.front()).arrivalTime; const bool sameArrivalCohort = all_of( prefillPre.begin(), prefillPre.end(), [&](RequestId id) { return fabsl(requests_.at(id).arrivalTime - cohortArrival) <= 1.0e-12L; }); long double knownSerialPromptBound = 0; long double largestTransfer = 0; long double smallestTransfer = numeric_limits::infinity(); for (RequestId id : prefillPre) { const Request& candidate = requests_.at(id); const long double transfer = networkLatency_ + 8.0L * candidate.inputLength * bytesPerToken_ / (bandwidth_ * 1.0e6L); largestTransfer = max(largestTransfer, transfer); smallestTransfer = min(smallestTransfer, transfer); knownSerialPromptBound += interpolatedDuration(candidate.inputLength, 0) + candidate.prefillProcessTime + interpolatedDuration(candidate.inputLength, 2) + 2.0L * transfer + 3.0L * scheduleCost_; } const long double visibleAge = max(0.0L, currentTime_ - cohortArrival); const bool promptSlackProved = visibleAge + knownSerialPromptBound <= sloFirstToken_; const bool serializedTransferDominates = 2.0L * largestTransfer >= 0.50L * sloFirstToken_ && largestTransfer > smallestTransfer + 1.0e-12L; if (sameArrivalCohort && promptSlackProved && serializedTransferDominates) { selectedPrefill = *max_element( prefillPre.begin(), prefillPre.end(), [&](RequestId left, RequestId right) { const Request& a = requests_.at(left); const Request& b = requests_.at(right); if (a.inputLength != b.inputLength) { return a.inputLength < b.inputLength; } return a.arrivalOrder > b.arrivalOrder; }); } } Request& request = requests_.at(selectedPrefill); request.remote = chooseRemote(request.inputLength); if (request.remote < 0) return; assignments.push_back("E P PRE " + to_string(request.remote) + " " + to_string(request.id)); request.phase = Phase::RunningPrefillPre; edgeDecodeStreak_ = 0; } else if (!mixedWaitingPostSync && !decodePre.empty() && (!sjfAdmissionSuppressed_ || throughputWeight_ <= waitingWeight_ || (throughputWeight_ >= 0.80L && longDecodeObserved())) && throughputWeight_ < 0.95L && staggeredArrivalsObserved()) { // With staggered arrivals, advance an already-known next decode wave // before closing the smaller POST wave. This exposes more compatible // endpoint work without waiting for a future arrival. if (shouldHoldFreshDecodePre(decodePre)) return; assignments.push_back(makeDecodeBatch("E", "PRE", -1, decodePre)); for (RequestId id : decodePre) requests_.at(id).phase = Phase::RunningDecodePre; ++edgeDecodeStreak_; } else if (!decodePost.empty()) { // A decode result transfer is already guaranteed to finish. Leaving E // idle until the current DOWN cohort drains lets one cross-remote POST // replace several tiny POST tasks without betting on a future arrival. const long double oneTokenTransfer = networkLatency_ + 8.0L * bytesPerToken_ / (bandwidth_ * 1.0e6L); const bool cheapNetwork = oneTokenTransfer <= 0.10L * sloTokenGap_; const int affinityChunk = inputClassAffinityChunk(); // A nontrivial input-affinity cohort already preserves cloud-side // batching. Closing each visible result immediately keeps the // shared Edge from stretching the next-token cadence while retaining // the original coalescing policy outside that public regime. const bool eagerAffinityDecodePost = !curves_[4].empty() && throughputWeight_ >= 4.0L * waitingWeight_ && affinityChunk > 1 && affinityChunk <= 2 * cloudCount_; const bool decodeWillAdvance = !eagerAffinityDecodePost && ( !collect(Phase::WaitingDecodeUpload).empty() || !collect(Phase::ReadyDecodeProcess).empty() || !collect(Phase::RunningDecodeProcess).empty() || !collect(Phase::WaitingDecodeDownload).empty()); const bool prefillStillAdvancing = !collect(Phase::WaitingPrefillUpload).empty() || !collect(Phase::ReadyPrefillProcess).empty() || !collect(Phase::RunningPrefillProcess).empty() || !collect(Phase::WaitingPrefillDownload).empty(); const bool mayWaitAcrossPrefill = waitingWeight_ > CROSS_PREFILL_POST_MIN_WAITING_WEIGHT; bool dpostCohortNeedsMore = decodeWillAdvance; const bool freshDecodePost = !decodePost.empty() && all_of(decodePost.begin(), decodePost.end(), [&](RequestId id) { return requests_.at(id).tokensProduced == 0; }); if (freshDecodePost) { set readyRemotes; for (RequestId id : decodePost) { readyRemotes.insert(requests_.at(id).remote); } bool representedRemoteStillFragmented = false; set activeFreshRemotes; long double earliestStart = numeric_limits::infinity(); long double latestStart = 0; for (RequestId id : arrivalOrder_) { const Request& request = requests_.at(id); if (request.tokensProduced != 0) continue; const bool scheduledDecode = request.phase == Phase::RunningDecodeProcess || request.phase == Phase::WaitingDecodeDownload || request.phase == Phase::ReadyDecodePost; if (scheduledDecode) { activeFreshRemotes.insert(request.remote); earliestStart = min(earliestStart, request.decodeProcessStartTime); latestStart = max(latestStart, request.decodeProcessStartTime); } if (readyRemotes.count(request.remote) != 0 && request.phase != Phase::ReadyDecodePost && isDecodePhase(request.phase)) { representedRemoteStillFragmented = true; } } const long double endpointRound = interpolatedDuration(decodePost.size(), 3) + interpolatedDuration(decodePost.size(), 5) + 2.0L * scheduleCost_; const long double extraEndpointWork = max(0, static_cast(activeFreshRemotes.size()) - 1) * endpointRound; const long double cloudRound = interpolatedDuration(decodePost.size(), 4) + scheduleCost_; const long double allEndpointWork = activeFreshRemotes.size() * endpointRound; if (!representedRemoteStillFragmented && activeFreshRemotes.size() == static_cast(cloudCount_) && activeFreshRemotes.size() > 1 && cloudRound > allEndpointWork && latestStart - earliestStart > extraEndpointWork) { freshDpostPipelineEngaged_ = true; } } // Once a clearly fragmented fresh wave proves that the cloud round // dominates all endpoint work, keep that wave pipelined through its // visible continuations instead of repeatedly rebuilding cohorts. if (freshDpostPipelineEngaged_ && (freshDecodePost || shouldSustainFragmentedDecodeWave())) { dpostCohortNeedsMore = false; } if (throughputWeight_ >= 0.10L && cheapNetwork && dpostCohortNeedsMore && (!prefillStillAdvancing || mayWaitAcrossPrefill)) return; assignments.push_back(makeDecodeBatch("E", "POST", -1, decodePost)); for (RequestId id : decodePost) requests_.at(id).phase = Phase::RunningDecodePost; ++edgeDecodeStreak_; } else if (!decodePre.empty()) { if (shouldHoldFreshDecodePre(decodePre)) return; assignments.push_back(makeDecodeBatch("E", "PRE", -1, decodePre)); for (RequestId id : decodePre) requests_.at(id).phase = Phase::RunningDecodePre; ++edgeDecodeStreak_; } else { // The pilot is already in flight. Waiting here is based only on known // work and lets its first-token outcome become observable; it never waits // for an unknown future arrival. return; } edgeBusy_ = true; } vector makeAssignments() { vector assignments; assignments.reserve(static_cast(cloudCount_) + 1); // Every task selected below was ready before this response. Scheduling the // cloud lines first cannot make an Edge task ready (or vice versa), so the // assignments have no forbidden same-timestamp dependency. for (int cloud = 0; cloud < cloudCount_; ++cloud) scheduleCloud(cloud, assignments); scheduleEdge(assignments); return assignments; } }; } // namespace int main() { ios::sync_with_stdio(false); cin.tie(nullptr); Scheduler scheduler; if (!scheduler.readConfiguration()) return 0; scheduler.run(); return 0; }