server /train: a per-turn slowdown receipt beside the cycle average

tests/CMakeLists.txt

@@ -150,6 +150,10 @@ endif ()
 
 llama_build(test-recurrent-state-rollback.cpp)
 
+# the server trainer's per-turn receipt, pure (tools/server/server-train-spans.h)
+llama_build_and_test(test-server-train-spans.cpp)
+target_include_directories(test-server-train-spans PRIVATE ${PROJECT_SOURCE_DIR}/tools/server)
+
 if (NOT WIN32 OR NOT BUILD_SHARED_LIBS)
     # these tests are disabled on Windows because they use internal functions not exported with LLAMA_API (when building with shared libraries)
     llama_build_and_test(test-unicode.cpp)

tests/test-server-train-spans.cpp

@@ -0,0 +1,41 @@
+// what this catches (Cormac on #39): the per-turn receipt pruned the training windows BEFORE
+// summing a turn's overlap, against a horizon at the turn's own end, so with no other turn in
+// flight every window that ended inside the turn was erased first and the slowed turn filed
+// as clean. Measured, then pruned.
+
+#include "server-train-spans.h"
+
+#include <cstdio>
+#include <cstdlib>
+
+#define CHECK(cond) do { if (!(cond)) { fprintf(stderr, "FAILED %s:%d: %s\n", __FILE__, __LINE__, #cond); exit(1); } } while (0)
+
+int main() {
+    // Cormac's case: a span [100, 200], one turn [50, 300], no other turn open (the server
+    // passes this turn's end as the oldest open start). Overlap 100 us, then the span goes.
+    {
+        std::vector<std::pair<int64_t, int64_t>> spans = {{100, 200}};
+        CHECK(server_train_turn_overlap_then_prune(spans, 50, 300, 300) == 100);
+        CHECK(spans.empty());
+    }
+    // a window still running counts to the turn's end and is never pruned
+    {
+        std::vector<std::pair<int64_t, int64_t>> spans = {{250, 0}};
+        CHECK(server_train_turn_overlap_then_prune(spans, 50, 300, 300) == 50);
+        CHECK(spans.size() == 1);
+    }
+    // another turn in flight since 150 keeps a window that ended at 200: it overlapped her
+    {
+        std::vector<std::pair<int64_t, int64_t>> spans = {{100, 200}};
+        CHECK(server_train_turn_overlap_then_prune(spans, 220, 300, 150) == 0);
+        CHECK(spans.size() == 1);
+    }
+    // an empty or inverted turn measures nothing but still prunes
+    {
+        std::vector<std::pair<int64_t, int64_t>> spans = {{100, 200}};
+        CHECK(server_train_turn_overlap_then_prune(spans, 300, 300, 300) == 0);
+        CHECK(spans.empty());
+    }
+    printf("test-server-train-spans: OK\n");
+    return 0;
+}

tools/server/server-context.cpp

@@ -2020,6 +2020,19 @@ struct server_context_impl {
     }
 
     void send_final_response(server_slot & slot) {
+        // every finished turn reaches the trainer: her per-turn baseline, or a slowdown sample
+        // t_prompt_last / t_gen_last are ggml_time_us TIMESTAMPS (server-common.h): generation
+        // began at the prompt's last batch and ended at the last token
+        if (trainer && slot.stats.t_gen_last > 0) {
+            // the earliest start of a turn still in flight: the trainer drops windows before it
+            int64_t oldest_open_us = slot.stats.t_gen_last;
+            for (const auto & other : slots) {
+                if (&other != &slot && other.is_processing() && other.stats.t_start > 0) {
+                    oldest_open_us = std::min(oldest_open_us, other.stats.t_start);
+                }
+            }
+            trainer->on_turn(slot.stats.t_prompt_last, slot.stats.t_gen_last, (int64_t) slot.stats.n_gen_steps(), oldest_open_us);
+        }
         auto res = std::make_unique<server_task_result_cmpl_final>();
 
         res->id      = slot.task->id;

tools/server/server-train-spans.h

@@ -0,0 +1,32 @@
+#pragma once
+
+// The per-turn receipt's one decision, pure (see server_trainer::on_turn): how much of a
+// finished turn a training window overlapped, and which windows can be forgotten after it.
+
+#include <algorithm>
+#include <cstdint>
+#include <utility>
+#include <vector>
+
+// spans: this run's training windows as [start, end) in ggml_time_us, end 0 while one runs.
+// Returns the overlap of [gen_start_us, gen_end_us) with every window, a running one counted
+// to the turn's end. THEN drops the windows that ended before oldest_open_us, the earliest
+// start of a turn still in flight (none of them can overlap a window that ended before it,
+// and turns not yet begun start after now). Measured BEFORE pruned: pruned first, a window
+// that ended inside this very turn was gone before it was counted, so a slowed turn filed as
+// clean (Cormac on #39).
+inline int64_t server_train_turn_overlap_then_prune(std::vector<std::pair<int64_t, int64_t>> & spans,
+                                                     int64_t gen_start_us, int64_t gen_end_us,
+                                                     int64_t oldest_open_us) {
+    int64_t overlap_us = 0;
+    if (gen_end_us > gen_start_us) {
+        for (const auto & [start, end] : spans) {
+            const int64_t e = end == 0 ? gen_end_us : end; // the window still running
+            overlap_us += std::max<int64_t>(0, std::min(e, gen_end_us) - std::max(start, gen_start_us));
+        }
+    }
+    spans.erase(std::remove_if(spans.begin(), spans.end(),
+                               [oldest_open_us](const std::pair<int64_t, int64_t> & w) { return w.second != 0 && w.second < oldest_open_us; }),
+                spans.end());
+    return overlap_us;
+}

tools/server/server-train.cpp

@@ -1,4 +1,5 @@
 #include "server-train.h"
+#include "server-train-spans.h"
 
 #include "common.h"
 #include "log.h"
@@ -465,6 +466,10 @@ json server_trainer::start(const json & body_in) {
     share_ppm.store(body.value("share_ppm", (int64_t) 250000));
     max_slowdown_ppm.store(body.value("max_slowdown_ppm", (int64_t) 0));
     busy_window_ms = busy_window_tokens = busy_yield_ms = busy_yield_tokens = 0;
+    window_spans.clear();
+    turn_slowdown_pct.fill(0);
+    turns_overlapped = 0;
+    turn_slowdown_ppm_max = 0;
     rate_no_window = rate_estimate{};
     rate_in_window = rate_estimate{};
     last_no_window_sample = std::chrono::steady_clock::time_point{};
@@ -493,6 +498,32 @@ int64_t server_trainer::yield_for_slowdown(int64_t d_ms, int64_t max_slowdown_pp
     return y > 0 ? (int64_t) std::ceil(y) : 0;
 }
 
+void server_trainer::on_turn(int64_t gen_start_us, int64_t gen_end_us, int64_t steps, int64_t oldest_open_us) {
+    std::lock_guard<std::mutex> lock(mu);
+    // measured, then pruned, on every finished turn (a short one too: it still ends windows'
+    // relevance); see server-train-spans.h
+    const int64_t overlap_us = server_train_turn_overlap_then_prune(window_spans, gen_start_us, gen_end_us, oldest_open_us);
+    if (steps < TURN_MIN_STEPS || gen_end_us <= gen_start_us) {
+        return;
+    }
+    const double ms = (double) (gen_end_us - gen_start_us) / 1000.0;
+    if (overlap_us == 0) {
+        turn_rate_clean.add(ms, (double) steps);
+        turns_clean += 1;
+        return;
+    }
+    const double clean = turn_rate_clean.per_ms();
+    if (clean > 0) { // no clean turn yet: nothing to measure her against, never a guessed sample
+        // the WHOLE turn's rate: a turn a window overlapped for 20% of its span reads diluted,
+        // which is what she felt over that turn, not the in-window slowdown
+        const double  slowdown = std::clamp(1.0 - ((double) steps / ms) / clean, 0.0, 1.0);
+        const int64_t ppm      = (int64_t) std::llround(slowdown * 1e6);
+        turn_slowdown_pct[(size_t) std::llround(slowdown * 100)] += 1;
+        turns_overlapped += 1;
+        turn_slowdown_ppm_max = std::max(turn_slowdown_ppm_max, ppm);
+    }
+}
+
 bool server_trainer::serving_busy() const {
     if (!yield_to_turns.load() || !serving.busy_slots) {
         return false;
@@ -520,6 +551,9 @@ bool server_trainer::before_window(bool, void * user_data) {
     if (last_ms > 0) { // 0 = no window yet (the first call), never a sample
         self.window_ms_samples.push_back(last_ms);
         self.window_ms_max = std::max(self.window_ms_max, last_ms);
+        if (!self.window_spans.empty() && self.window_spans.back().second == 0) {
+            self.window_spans.back().second = ggml_time_us();
+        }
         if (self.window_busy_start && working_t0) { // her rate WITH a window running
             self.busy_window_ms += last_ms;
             self.busy_window_tokens += tokens_t0 - self.window_tokens_start;
@@ -597,6 +631,7 @@ bool server_trainer::before_window(bool, void * user_data) {
             self.windows_while_busy += 1;
         }
         self.window_started = std::chrono::steady_clock::now();
+        self.window_spans.emplace_back(ggml_time_us(), 0);
         self.window_tokens_start = tokens_t1;
         self.window_busy_start = self.serving.busy_slots && self.serving.busy_slots() > 0;
     }
@@ -679,6 +714,32 @@ json server_trainer::status() const {
     if (rate_in_window.ms > 0) {
         s["decode_tps_in_window_recent"] = rate_in_window.per_ms() * 1000.0;
     }
+    // the per-turn receipt: each turn that overlapped a window, against her clean turns. A
+    // sample is the whole turn's slowdown, so a turn a window only partly overlapped reads
+    // diluted: a p95 here is what her turns felt, never the slowdown inside a window. p50/p95
+    // to the 1% bucket (a fixed histogram), max exact.
+    s["turns_clean"] = turns_clean;
+    s["turns_overlapped"] = turns_overlapped;
+    if (turn_rate_clean.ms > 0) {
+        s["turn_tps_clean"] = turn_rate_clean.per_ms() * 1000.0;
+    }
+    if (turns_overlapped > 0) {
+        const auto pct = [&](double p) {
+            const int64_t rank = std::min(turns_overlapped - 1, (int64_t) (p * (double) turns_overlapped));
+            int64_t seen = 0;
+            for (size_t b = 0; b < turn_slowdown_pct.size(); ++b) {
+                seen += turn_slowdown_pct[b];
+                if (seen > rank) {
+                    // a bucket rounds to the nearest 1%, which can sit above the exact max
+                    return std::min((int64_t) b * 10000, turn_slowdown_ppm_max);
+                }
+            }
+            return (int64_t) 1000000;
+        };
+        s["turn_slowdown_ppm_p50"] = pct(0.50);
+        s["turn_slowdown_ppm_p95"] = pct(0.95);
+        s["turn_slowdown_ppm_max"] = turn_slowdown_ppm_max;
+    }
     s["busy_window_ms"] = busy_window_ms;
     s["busy_yield_ms"] = busy_yield_ms;
     s["windows"] = windows.load();
@@ -707,6 +768,7 @@ void server_trainer::run(json req, examples_data ex) {
         pause_requested = false;
         paused = false;
         waiting_for_serving = false;
+        close_window_span();
         running.store(false);
     };
 
@@ -952,5 +1014,14 @@ void server_trainer::run(json req, examples_data ex) {
         state["state"] = "done";
         state["adapter"] = out;
     }
+    close_window_span();
     running.store(false);
 }
+
+void server_trainer::close_window_span() {
+    // a run that ends (done, failed, cancelled) ends its last window: an open span would read
+    // every later turn as overlapped by training that no longer runs
+    if (!window_spans.empty() && window_spans.back().second == 0) {
+        window_spans.back().second = ggml_time_us();
+    }
+}

tools/server/server-train.h

@@ -45,6 +45,7 @@
 #include "json.h"
 #include "ggml-opt.h"
 
+#include <array>
 #include <atomic>
 #include <cmath>
 #include <condition_variable>
@@ -102,6 +103,14 @@ class server_trainer {
     // Starts a run on a worker thread; refuses (ok=false) while one is running or on bad input.
     common_json start(const common_json & body_in);
     common_json status() const;
+
+    // A finished turn, from serving: its generation began at gen_start_us and ended at
+    // gen_end_us (ggml_time_us) over `steps` decode steps. A turn that overlapped no training
+    // window is her per-turn baseline; one that overlapped a window is a per-turn slowdown
+    // sample against it (the bound is on the cycle's average, Joel's bar is per turn: Cormac).
+    // oldest_open_us: the earliest start of any turn still in flight (gen_end_us when none); no
+    // window that ended before it can overlap a turn still to finish.
+    void on_turn(int64_t gen_start_us, int64_t gen_end_us, int64_t steps, int64_t oldest_open_us);
     // Stops the running job at its next training window (no adapter is written); ok=false when
     // nothing is running.
     common_json cancel();
@@ -111,6 +120,7 @@ class server_trainer {
 private:
     void run(common_json req, examples_data ex);
     static bool before_window(bool train, void * user_data);
+    void close_window_span(); // under mu: a run's end ends its last window
     static void on_batch(bool train, ggml_opt_context_t, ggml_opt_dataset_t, ggml_opt_result_t,
                          int64_t ibatch, int64_t ibatch_max, int64_t);
 
@@ -157,6 +167,18 @@ class server_trainer {
     // price of the bound staying true.
     static constexpr int64_t PROBE_EVERY_MS = 30000;
     static constexpr int64_t PROBE_MS = 500;
+    // THE PER-TURN RECEIPT. Windows of this run as [start, end) in ggml_time_us (end 0 while
+    // one runs), every turn's overlap measured against them; a turn's rate is per slot, so its
+    // baseline is other turns' rate, never the lane's total rate. Guarded by mu.
+    static constexpr int64_t TURN_MIN_STEPS = 16; // fewer decode steps time the scheduler, not her
+    // Bounded: spans that ended before the oldest turn still in flight began can overlap no
+    // turn still to finish, and are dropped; samples live in a fixed histogram of 1% buckets.
+    std::vector<std::pair<int64_t, int64_t>> window_spans;
+    rate_estimate turn_rate_clean;               // turns that overlapped no window, across runs
+    std::array<int64_t, 101> turn_slowdown_pct{}; // turns that overlapped a window, this run, by % slower
+    int64_t turns_overlapped{0};
+    int64_t turn_slowdown_ppm_max{0};
+    int64_t turns_clean{0};
     rate_estimate rate_no_window;   // guarded by mu
     rate_estimate rate_in_window;   // guarded by mu
     std::chrono::steady_clock::time_point last_no_window_sample{};