diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index c3305283012..5fc7d54312d 100644 --- a/tests/CMakeLists.txt +++ b/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) diff --git a/tests/test-server-train-spans.cpp b/tests/test-server-train-spans.cpp new file mode 100644 index 00000000000..a3e2cdbd312 --- /dev/null +++ b/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 +#include + +#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> 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> 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> 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> 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; +} diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index a35e8d2a2b4..cb99e47171c 100644 --- a/tools/server/server-context.cpp +++ b/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(); res->id = slot.task->id; diff --git a/tools/server/server-train-spans.h b/tools/server/server-train-spans.h new file mode 100644 index 00000000000..989a9653a95 --- /dev/null +++ b/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 +#include +#include +#include + +// 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> & 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(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 & w) { return w.second != 0 && w.second < oldest_open_us; }), + spans.end()); + return overlap_us; +} diff --git a/tools/server/server-train.cpp b/tools/server/server-train.cpp index 99d8d2cbf1d..2b531cfe30a 100644 --- a/tools/server/server-train.cpp +++ b/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 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(); + } +} diff --git a/tools/server/server-train.h b/tools/server/server-train.h index d7e380b9a24..f05062c6a42 100644 --- a/tools/server/server-train.h +++ b/tools/server/server-train.h @@ -45,6 +45,7 @@ #include "json.h" #include "ggml-opt.h" +#include #include #include #include @@ -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> window_spans; + rate_estimate turn_rate_clean; // turns that overlapped no window, across runs + std::array 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{};