From 5932c95e8d8e6b2cd687b15900a81b0214bd6706 Mon Sep 17 00:00:00 2001 From: Nathan Easton Date: Thu, 13 Aug 2026 11:42:15 -0400 Subject: [PATCH 1/6] Add token/s on dashboard and reporting token/s in the metrics endpoint --- paddler_agent/src/continuous_batch_arbiter.rs | 1 + .../advance_generating_phase.rs | 8 +- .../src/continuous_batch_scheduler_context.rs | 2 + paddler_agent/src/lib.rs | 1 + paddler_agent/src/slot_aggregated_status.rs | 49 +++++++ paddler_agent/src/token_throughput_meter.rs | 124 ++++++++++++++++++ paddler_balancer/src/agent_controller.rs | 24 ++++ paddler_balancer/src/agent_controller_pool.rs | 29 ++++ .../src/buffered_request_manager.rs | 3 + .../src/controls_manages_senders_endpoint.rs | 1 + .../api/post_generate_embedding_batch.rs | 1 + .../http_route/api/ws_inference_socket/mod.rs | 1 + .../http_route/api/get_agents.rs | 1 + .../agent_socket_controller_context.rs | 1 + .../http_route/api/ws_agent_socket/mod.rs | 3 + .../http_route/get_metrics.rs | 5 + .../src/reconciliation_service.rs | 1 + paddler_balancer/src/request_from_agent.rs | 1 + paddler_balancer/src/statsd_service/mod.rs | 13 +- .../src/schemas/Agent.ts | 1 + .../tests/schemas/Agent.test.ts | 2 + paddler_gui/src/agent_running_data.rs | 1 + paddler_gui/src/running_balancer_snapshot.rs | 1 + paddler_gui/src/screen.rs | 1 + paddler_gui/src/ui/view_agent_card.rs | 13 +- .../src/agent_controller_snapshot.rs | 1 + .../register_agent_params.rs | 2 +- .../update_agent_status_params.rs | 2 +- .../src/slot_aggregated_status_snapshot.rs | 3 +- .../assert_slots_total_at_least.rs | 1 + .../src/agents_stream_watcher.rs | 1 + ...e_agent_controller_without_remote_agent.rs | 1 + paddler_tests/tests/harness_agents_watcher.rs | 1 + .../AgentListAgentStatus.module.css | 2 +- .../ts/components/AgentListAgentStatus.tsx | 6 + 35 files changed, 301 insertions(+), 7 deletions(-) create mode 100644 paddler_agent/src/token_throughput_meter.rs diff --git a/paddler_agent/src/continuous_batch_arbiter.rs b/paddler_agent/src/continuous_batch_arbiter.rs index a63c897b0..a7c17f4a1 100644 --- a/paddler_agent/src/continuous_batch_arbiter.rs +++ b/paddler_agent/src/continuous_batch_arbiter.rs @@ -306,6 +306,7 @@ impl ContinuousBatchArbiter { inference_parameters, model_path: model_path.clone(), multimodal_context, + slot_aggregated_status: slot_aggregated_status_manager.slot_aggregated_status.clone(), token_bos_str: model.token_to_piece( &SampledToken::Content(model.token_bos()), &mut special_token_decoder, diff --git a/paddler_agent/src/continuous_batch_scheduler/advance_generating_phase.rs b/paddler_agent/src/continuous_batch_scheduler/advance_generating_phase.rs index 75b25ab62..cb14f0374 100644 --- a/paddler_agent/src/continuous_batch_scheduler/advance_generating_phase.rs +++ b/paddler_agent/src/continuous_batch_scheduler/advance_generating_phase.rs @@ -49,7 +49,13 @@ impl AdvanceGeneratingPhase<'_> { }) .run(request, batch_index) { - SampleOutcome::Sampled(token) => token, + SampleOutcome::Sampled(token) => { + self.scheduler_context + .slot_aggregated_status + .record_generated_token(); + + token + } SampleOutcome::AllCandidatesEliminated => { error!( "{:?}: sequence {} sampling exhausted candidates", diff --git a/paddler_agent/src/continuous_batch_scheduler_context.rs b/paddler_agent/src/continuous_batch_scheduler_context.rs index d9473563c..0034850ff 100644 --- a/paddler_agent/src/continuous_batch_scheduler_context.rs +++ b/paddler_agent/src/continuous_batch_scheduler_context.rs @@ -6,6 +6,7 @@ use llama_cpp_bindings::mtmd::MtmdContext; use paddler_messaging::inference_parameters::InferenceParameters; use crate::chat_template_renderer::ChatTemplateRenderer; +use crate::slot_aggregated_status::SlotAggregatedStatus; pub struct ContinuousBatchSchedulerContext { pub agent_name: Option, @@ -15,6 +16,7 @@ pub struct ContinuousBatchSchedulerContext { pub model: Arc, pub model_path: PathBuf, pub multimodal_context: Option>, + pub slot_aggregated_status: Arc, pub token_bos_str: String, pub token_eos_str: String, pub token_nl_str: String, diff --git a/paddler_agent/src/lib.rs b/paddler_agent/src/lib.rs index f95c51a75..97af63a38 100644 --- a/paddler_agent/src/lib.rs +++ b/paddler_agent/src/lib.rs @@ -59,6 +59,7 @@ pub mod slot_aggregated_status; pub mod slot_aggregated_status_download_progress; pub mod slot_aggregated_status_manager; pub mod slot_guard; +pub mod token_throughput_meter; pub mod tool_call_buffer; pub mod tool_call_event; pub mod tool_call_pipeline; diff --git a/paddler_agent/src/slot_aggregated_status.rs b/paddler_agent/src/slot_aggregated_status.rs index a66ce0a17..b97480bb8 100644 --- a/paddler_agent/src/slot_aggregated_status.rs +++ b/paddler_agent/src/slot_aggregated_status.rs @@ -12,6 +12,7 @@ use tokio::sync::watch; use crate::agent_issue_fix::AgentIssueFix; use crate::dispenses_slots::DispensesSlots; +use crate::token_throughput_meter::TokenThroughputMeter; use paddler_messaging::atomic_value::AtomicValue; use paddler_messaging::produces_snapshot::ProducesSnapshot; use paddler_messaging::subscribes_to_updates::SubscribesToUpdates; @@ -27,6 +28,7 @@ pub struct SlotAggregatedStatus { slots_processing: AtomicValue, slots_total: AtomicValue, state_application_status_code: AtomicValue, + token_throughput_meter: TokenThroughputMeter, update_tx: watch::Sender<()>, uses_chat_template_override: AtomicValue, version: AtomicValue, @@ -50,6 +52,7 @@ impl SlotAggregatedStatus { ), slots_processing: AtomicValue::::new(0), slots_total: AtomicValue::::new(0), + token_throughput_meter: TokenThroughputMeter::new(), update_tx, uses_chat_template_override: AtomicValue::::new(false), version: AtomicValue::::new(0), @@ -174,6 +177,21 @@ impl SlotAggregatedStatus { pub fn slots_processing_count(&self) -> i32 { self.slots_processing.get() } + + /// Records that this agent has just generated one token, contributing to + /// the tokens-per-second rate reported in status snapshots. + /// + /// This does not bump `version` or notify subscribers: it happens once + /// per generated token, far too often to treat as a state change worth + /// pushing immediately. The periodic status update (sent once a second + /// regardless of `version`) is what picks the latest rate up. + pub fn record_generated_token(&self) { + self.token_throughput_meter.record_token(); + } + + pub fn tokens_per_second(&self) -> f64 { + self.token_throughput_meter.tokens_per_second() + } } impl DispensesSlots for SlotAggregatedStatus { @@ -211,6 +229,7 @@ impl ProducesSnapshot for SlotAggregatedStatus { slots_processing: self.slots_processing.get(), slots_total: self.slots_total.get(), state_application_status: self.state_application_status_code.get().try_into()?, + tokens_per_second: self.tokens_per_second(), uses_chat_template_override: self.uses_chat_template_override.get(), version: self.version.get(), }) @@ -341,6 +360,36 @@ mod tests { ); } + #[test] + fn make_snapshot_reports_zero_tokens_per_second_before_any_generation() { + let status = SlotAggregatedStatus::new(1); + + let snapshot = status.make_snapshot().unwrap(); + + assert_eq!(snapshot.tokens_per_second, 0.0); + } + + #[test] + fn record_generated_token_is_reflected_in_snapshot_after_a_window_closes() { + let status = SlotAggregatedStatus::new(1); + + for _ in 0..5 { + status.record_generated_token(); + } + + std::thread::sleep(std::time::Duration::from_millis(1050)); + + status.record_generated_token(); + + let snapshot = status.make_snapshot().unwrap(); + + assert!( + snapshot.tokens_per_second > 0.0, + "expected a positive tokens_per_second, got {}", + snapshot.tokens_per_second + ); + } + #[test] fn get_state_application_status_reflects_set_value() { let status = SlotAggregatedStatus::new(2); diff --git a/paddler_agent/src/token_throughput_meter.rs b/paddler_agent/src/token_throughput_meter.rs new file mode 100644 index 000000000..cc274565d --- /dev/null +++ b/paddler_agent/src/token_throughput_meter.rs @@ -0,0 +1,124 @@ +use std::time::Duration; +use std::time::Instant; + +use parking_lot::Mutex; + +/// How long a window stays open before its token count is turned into a +/// tokens/sec measurement. One second keeps the reported rate intuitive +/// (it is, literally, "tokens counted in the last second"). +const WINDOW_DURATION: Duration = Duration::from_secs(1); + +/// A measurement is considered stale, and reported as `0.0`, once this much +/// time has passed without a new token. Without this, a slot that produced a +/// quick burst and then went idle would keep reporting its last burst's rate +/// forever. +const STALE_AFTER: Duration = Duration::from_secs(2); + +struct ThroughputWindow { + tokens_per_second: f64, + window_started_at: Instant, + window_token_count: u64, +} + +/// Tracks generated tokens for a single agent and derives a continuously +/// updated, approximate tokens-per-second throughput value out of them. +/// +/// The meter is intentionally simple: it counts tokens in a rolling +/// one-second window and, once the window closes, turns that count into the +/// reported rate. There is nothing to configure. +pub struct TokenThroughputMeter { + window: Mutex, +} + +impl TokenThroughputMeter { + #[must_use] + pub fn new() -> Self { + Self { + window: Mutex::new(ThroughputWindow { + tokens_per_second: 0.0, + window_started_at: Instant::now(), + window_token_count: 0, + }), + } + } + + /// Records a single generated token, closing out and measuring the + /// current window if it has been open for at least [`WINDOW_DURATION`]. + pub fn record_token(&self) { + let mut window = self.window.lock(); + let elapsed = window.window_started_at.elapsed(); + + window.window_token_count += 1; + + if elapsed >= WINDOW_DURATION { + window.tokens_per_second = window.window_token_count as f64 / elapsed.as_secs_f64(); + window.window_token_count = 0; + window.window_started_at = Instant::now(); + } + } + + /// Returns the most recently measured tokens-per-second rate, or `0.0` + /// if generation has been idle for longer than [`STALE_AFTER`]. + #[must_use] + pub fn tokens_per_second(&self) -> f64 { + let window = self.window.lock(); + + if window.window_started_at.elapsed() > STALE_AFTER && window.window_token_count == 0 { + return 0.0; + } + + window.tokens_per_second + } +} + +impl Default for TokenThroughputMeter { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use std::thread::sleep; + + use super::*; + + #[test] + fn reports_zero_before_any_tokens_are_recorded() { + let meter = TokenThroughputMeter::new(); + + assert_eq!(meter.tokens_per_second(), 0.0); + } + + #[test] + fn measures_rate_once_a_window_closes() { + let meter = TokenThroughputMeter::new(); + + for _ in 0..5 { + meter.record_token(); + } + + sleep(WINDOW_DURATION + Duration::from_millis(50)); + + meter.record_token(); + + let rate = meter.tokens_per_second(); + + assert!(rate > 0.0, "expected a positive rate, got {rate}"); + } + + #[test] + fn decays_to_zero_after_being_idle() { + let meter = TokenThroughputMeter::new(); + + for _ in 0..5 { + meter.record_token(); + } + + sleep(WINDOW_DURATION + Duration::from_millis(50)); + meter.record_token(); + sleep(STALE_AFTER + Duration::from_millis(50)); + + assert_eq!(meter.tokens_per_second(), 0.0); + } +} diff --git a/paddler_balancer/src/agent_controller.rs b/paddler_balancer/src/agent_controller.rs index be5d17170..a110bac9e 100644 --- a/paddler_balancer/src/agent_controller.rs +++ b/paddler_balancer/src/agent_controller.rs @@ -59,6 +59,7 @@ pub struct AgentController { pub slots_processing: AtomicValue, pub slots_total: AtomicValue, pub state_application_status_code: AtomicValue, + pub tokens_per_second: RwLock, pub uses_chat_template_override: AtomicValue, } @@ -95,6 +96,10 @@ impl AgentController { self.model_path.read().clone() } + pub fn get_tokens_per_second(&self) -> f64 { + *self.tokens_per_second.read() + } + pub fn set_download_filename(&self, filename: Option) { let mut locked_filename = self.download_filename.write(); @@ -113,6 +118,12 @@ impl AgentController { *locked_path = model_path; } + pub fn set_tokens_per_second(&self, tokens_per_second: f64) { + let mut locked_tokens_per_second = self.tokens_per_second.write(); + + *locked_tokens_per_second = tokens_per_second; + } + pub async fn stop_responding_to(&self, request_id: String) -> Result<()> { self.send_rpc_message(AgentJsonRpcMessage::Notification( AgentJsonRpcNotification::StopRespondingTo(request_id), @@ -134,6 +145,7 @@ impl AgentController { model_path, slots_total, state_application_status, + tokens_per_second, uses_chat_template_override, version, .. @@ -184,6 +196,12 @@ impl AgentController { self.set_model_path(model_path); } + if tokens_per_second != self.get_tokens_per_second() { + changed = true; + + self.set_tokens_per_second(tokens_per_second); + } + if changed { AgentControllerUpdateResult::Updated } else { @@ -304,6 +322,7 @@ impl ProducesSnapshot for AgentController { slots_processing: self.slots_processing.get(), slots_total: self.slots_total.get(), state_application_status: self.state_application_status_code.get().try_into()?, + tokens_per_second: self.get_tokens_per_second(), uses_chat_template_override: self.uses_chat_template_override.get(), }) } @@ -368,6 +387,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), } } @@ -387,6 +407,7 @@ mod tests { slots_processing: 0, slots_total: 4, state_application_status: AgentStateApplicationStatus::Fresh, + tokens_per_second: 0.0, uses_chat_template_override: true, version: 1, }; @@ -418,6 +439,7 @@ mod tests { slots_processing: 0, slots_total: 0, state_application_status: AgentStateApplicationStatus::Fresh, + tokens_per_second: 0.0, uses_chat_template_override: false, version: 1, }; @@ -448,6 +470,7 @@ mod tests { slots_processing: 0, slots_total: 0, state_application_status: AgentStateApplicationStatus::Fresh, + tokens_per_second: 0.0, uses_chat_template_override: false, version: 1, }; @@ -481,6 +504,7 @@ mod tests { slots_processing: 0, slots_total: 0, state_application_status: AgentStateApplicationStatus::Fresh, + tokens_per_second: 0.0, uses_chat_template_override: false, version: 1, }; diff --git a/paddler_balancer/src/agent_controller_pool.rs b/paddler_balancer/src/agent_controller_pool.rs index 35922e6c1..4eb63b239 100644 --- a/paddler_balancer/src/agent_controller_pool.rs +++ b/paddler_balancer/src/agent_controller_pool.rs @@ -131,6 +131,16 @@ impl AgentControllerPool { slots_total, } } + + /// Sums the most recently reported tokens-per-second rate across every + /// registered agent, giving the balancer's overall generation throughput. + #[must_use] + pub fn total_tokens_per_second(&self) -> f64 { + self.agents + .iter() + .map(|entry| entry.value().get_tokens_per_second()) + .sum() + } } impl Default for AgentControllerPool { @@ -236,6 +246,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }) } @@ -297,6 +308,24 @@ mod tests { assert_eq!(total_slots.slots_total, 12); } + #[test] + fn total_tokens_per_second_sums_rate_across_agents() { + let pool = AgentControllerPool::default(); + + let first = agent_controller_with_slots(1, 4); + first.set_tokens_per_second(12.5); + + let second = agent_controller_with_slots(2, 8); + second.set_tokens_per_second(7.5); + + pool.register_agent_controller("first".to_owned(), first) + .unwrap(); + pool.register_agent_controller("second".to_owned(), second) + .unwrap(); + + assert_eq!(pool.total_tokens_per_second(), 20.0); + } + #[test] fn make_snapshot_includes_each_registered_agent() { let pool = AgentControllerPool::default(); diff --git a/paddler_balancer/src/buffered_request_manager.rs b/paddler_balancer/src/buffered_request_manager.rs index 85f117409..46304d744 100644 --- a/paddler_balancer/src/buffered_request_manager.rs +++ b/paddler_balancer/src/buffered_request_manager.rs @@ -143,6 +143,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }); @@ -219,6 +220,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }); @@ -268,6 +270,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }); diff --git a/paddler_balancer/src/controls_manages_senders_endpoint.rs b/paddler_balancer/src/controls_manages_senders_endpoint.rs index c4b6cefbc..86f47284e 100644 --- a/paddler_balancer/src/controls_manages_senders_endpoint.rs +++ b/paddler_balancer/src/controls_manages_senders_endpoint.rs @@ -110,6 +110,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }), ) diff --git a/paddler_balancer/src/inference_service/http_route/api/post_generate_embedding_batch.rs b/paddler_balancer/src/inference_service/http_route/api/post_generate_embedding_batch.rs index bd8d3472b..721ada85e 100644 --- a/paddler_balancer/src/inference_service/http_route/api/post_generate_embedding_batch.rs +++ b/paddler_balancer/src/inference_service/http_route/api/post_generate_embedding_batch.rs @@ -239,6 +239,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }) } diff --git a/paddler_balancer/src/inference_service/http_route/api/ws_inference_socket/mod.rs b/paddler_balancer/src/inference_service/http_route/api/ws_inference_socket/mod.rs index e3797efdc..aadd04882 100644 --- a/paddler_balancer/src/inference_service/http_route/api/ws_inference_socket/mod.rs +++ b/paddler_balancer/src/inference_service/http_route/api/ws_inference_socket/mod.rs @@ -342,6 +342,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }); diff --git a/paddler_balancer/src/management_service/http_route/api/get_agents.rs b/paddler_balancer/src/management_service/http_route/api/get_agents.rs index 2f2414652..752d1d343 100644 --- a/paddler_balancer/src/management_service/http_route/api/get_agents.rs +++ b/paddler_balancer/src/management_service/http_route/api/get_agents.rs @@ -83,6 +83,7 @@ mod tests { slots_processing: AtomicValue::::new(0), slots_total: AtomicValue::::new(0), state_application_status_code: AtomicValue::::new(status_code), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }) } diff --git a/paddler_balancer/src/management_service/http_route/api/ws_agent_socket/agent_socket_controller_context.rs b/paddler_balancer/src/management_service/http_route/api/ws_agent_socket/agent_socket_controller_context.rs index 9631b28b6..24045e5d3 100644 --- a/paddler_balancer/src/management_service/http_route/api/ws_agent_socket/agent_socket_controller_context.rs +++ b/paddler_balancer/src/management_service/http_route/api/ws_agent_socket/agent_socket_controller_context.rs @@ -92,6 +92,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }), ) diff --git a/paddler_balancer/src/management_service/http_route/api/ws_agent_socket/mod.rs b/paddler_balancer/src/management_service/http_route/api/ws_agent_socket/mod.rs index 2eb425c55..655b8bdf3 100644 --- a/paddler_balancer/src/management_service/http_route/api/ws_agent_socket/mod.rs +++ b/paddler_balancer/src/management_service/http_route/api/ws_agent_socket/mod.rs @@ -124,6 +124,7 @@ impl ControlsWebSocketEndpoint for AgentSocketController { slots_processing, slots_total, state_application_status, + tokens_per_second, uses_chat_template_override, version, }, @@ -159,6 +160,7 @@ impl ControlsWebSocketEndpoint for AgentSocketController { state_application_status_code: AtomicValue::::new( state_application_status as i32, ), + tokens_per_second: RwLock::new(tokens_per_second), uses_chat_template_override: AtomicValue::::new( uses_chat_template_override, ), @@ -409,6 +411,7 @@ mod tests { slots_processing: 0, slots_total: 1, state_application_status: AgentStateApplicationStatus::Fresh, + tokens_per_second: 0.0, uses_chat_template_override: false, version: 0, }, diff --git a/paddler_balancer/src/management_service/http_route/get_metrics.rs b/paddler_balancer/src/management_service/http_route/get_metrics.rs index 276c01990..2d4fe9ff7 100644 --- a/paddler_balancer/src/management_service/http_route/get_metrics.rs +++ b/paddler_balancer/src/management_service/http_route/get_metrics.rs @@ -24,6 +24,7 @@ async fn respond(app_data: Data) -> Result) -> Result::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }) } diff --git a/paddler_balancer/src/request_from_agent.rs b/paddler_balancer/src/request_from_agent.rs index 58f450132..985e848dc 100644 --- a/paddler_balancer/src/request_from_agent.rs +++ b/paddler_balancer/src/request_from_agent.rs @@ -453,6 +453,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }); diff --git a/paddler_balancer/src/statsd_service/mod.rs b/paddler_balancer/src/statsd_service/mod.rs index 7de8c06b5..5280c94c5 100644 --- a/paddler_balancer/src/statsd_service/mod.rs +++ b/paddler_balancer/src/statsd_service/mod.rs @@ -38,16 +38,24 @@ impl StatsdService { slots_total, } = self.agent_controller_pool.total_slots(); let requests_buffered = self.buffered_request_manager.buffered_request_counter.get(); + let tokens_per_second = self.agent_controller_pool.total_tokens_per_second(); let slots_processing = u64::try_from(slots_processing).context("slots_processing count is negative")?; let slots_total = u64::try_from(slots_total).context("slots_total count is negative")?; let requests_buffered = u64::try_from(requests_buffered).context("requests_buffered count is negative")?; + #[expect( + clippy::cast_possible_truncation, + clippy::cast_sign_loss, + reason = "StatsD gauges are unsigned integers; sub-token precision is not meaningful" + )] + let tokens_per_second = tokens_per_second.round() as u64; client.gauge("slots_processing", slots_processing)?; client.gauge("slots_total", slots_total)?; client.gauge("requests_buffered", requests_buffered)?; + client.gauge("tokens_per_second", tokens_per_second)?; client.flush()?; Ok(()) @@ -152,6 +160,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }), ) @@ -199,7 +208,7 @@ mod tests { let mut received_lines: Vec = Vec::new(); let mut datagram = [0_u8; 1024]; - for _ in 0..3 { + for _ in 0..4 { let byte_count = receiver.recv(&mut datagram).await.unwrap(); received_lines.push(String::from_utf8(datagram[..byte_count].to_vec()).unwrap()); @@ -208,6 +217,7 @@ mod tests { assert!(received_lines.contains(&"paddler.slots_processing:0|g".to_owned())); assert!(received_lines.contains(&"paddler.slots_total:0|g".to_owned())); assert!(received_lines.contains(&"paddler.requests_buffered:0|g".to_owned())); + assert!(received_lines.contains(&"paddler.tokens_per_second:0|g".to_owned())); } #[tokio::test] @@ -227,6 +237,7 @@ mod tests { "paddler.slots_processing:0|g".to_owned(), "paddler.slots_total:0|g".to_owned(), "paddler.requests_buffered:0|g".to_owned(), + "paddler.tokens_per_second:0|g".to_owned(), ]; assert!(expected_first_tick_lines.contains(&first_line)); diff --git a/paddler_client_javascript/src/schemas/Agent.ts b/paddler_client_javascript/src/schemas/Agent.ts index 1ad87eaad..5788f20a1 100644 --- a/paddler_client_javascript/src/schemas/Agent.ts +++ b/paddler_client_javascript/src/schemas/Agent.ts @@ -22,6 +22,7 @@ export const AgentSchema = z "Fresh", "Stuck", ]), + tokens_per_second: z.number(), uses_chat_template_override: z.boolean(), }) .strict(); diff --git a/paddler_client_javascript/tests/schemas/Agent.test.ts b/paddler_client_javascript/tests/schemas/Agent.test.ts index 3ad0ba4ef..647667910 100644 --- a/paddler_client_javascript/tests/schemas/Agent.test.ts +++ b/paddler_client_javascript/tests/schemas/Agent.test.ts @@ -17,6 +17,7 @@ test("parses a fully populated agent payload", function () { slots_processing: 1, slots_total: 4, state_application_status: "Applied", + tokens_per_second: 12.5, uses_chat_template_override: false, }); @@ -39,6 +40,7 @@ test("rejects an unknown state_application_status", function () { slots_processing: 0, slots_total: 1, state_application_status: "Unknown", + tokens_per_second: 0, uses_chat_template_override: false, }); }); diff --git a/paddler_gui/src/agent_running_data.rs b/paddler_gui/src/agent_running_data.rs index dae4782a2..ede9e0f25 100644 --- a/paddler_gui/src/agent_running_data.rs +++ b/paddler_gui/src/agent_running_data.rs @@ -23,6 +23,7 @@ impl AgentRunningData { slots_processing: status.slots_processing, slots_total: status.slots_total, state_application_status: status.state_application_status, + tokens_per_second: status.tokens_per_second, uses_chat_template_override: status.uses_chat_template_override, }; } diff --git a/paddler_gui/src/running_balancer_snapshot.rs b/paddler_gui/src/running_balancer_snapshot.rs index 91f6960e4..f591dca73 100644 --- a/paddler_gui/src/running_balancer_snapshot.rs +++ b/paddler_gui/src/running_balancer_snapshot.rs @@ -98,6 +98,7 @@ mod tests { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), }) } diff --git a/paddler_gui/src/screen.rs b/paddler_gui/src/screen.rs index 0d180df6b..ff5c4b3af 100644 --- a/paddler_gui/src/screen.rs +++ b/paddler_gui/src/screen.rs @@ -84,6 +84,7 @@ impl Screen { slots_processing: 0, slots_total: 0, state_application_status: AgentStateApplicationStatus::Fresh, + tokens_per_second: 0.0, uses_chat_template_override: false, }, } diff --git a/paddler_gui/src/ui/view_agent_card.rs b/paddler_gui/src/ui/view_agent_card.rs index e50cdef32..41cbb5663 100644 --- a/paddler_gui/src/ui/view_agent_card.rs +++ b/paddler_gui/src/ui/view_agent_card.rs @@ -96,9 +96,20 @@ pub fn view_agent_card( snapshot.slots_processing, snapshot.slots_total, snapshot.desired_slots_total, ); + let mut status_row_right = column![].spacing(SPACING_HALF); + + status_row_right = + status_row_right.push(text(format!("Slots: {slots_label}")).font(REGULAR)); + + if snapshot.tokens_per_second > 0.0 { + status_row_right = status_row_right.push( + text(format!("{:.1} tok/s", snapshot.tokens_per_second)).font(REGULAR), + ); + } + let status_row_content = row![ container(status_row_left).width(Fill), - text(format!("Slots: {slots_label}")).font(REGULAR), + status_row_right, ]; let card_content = column![name_row, status_row_content,].spacing(SPACING_BASE); diff --git a/paddler_messaging/src/agent_controller_snapshot.rs b/paddler_messaging/src/agent_controller_snapshot.rs index 559ffefa3..bcf7bd8c1 100644 --- a/paddler_messaging/src/agent_controller_snapshot.rs +++ b/paddler_messaging/src/agent_controller_snapshot.rs @@ -21,5 +21,6 @@ pub struct AgentControllerSnapshot { pub slots_processing: i32, pub slots_total: i32, pub state_application_status: AgentStateApplicationStatus, + pub tokens_per_second: f64, pub uses_chat_template_override: bool, } diff --git a/paddler_messaging/src/management_socket/balancer/notification_params/register_agent_params.rs b/paddler_messaging/src/management_socket/balancer/notification_params/register_agent_params.rs index e6153ebc3..20384abde 100644 --- a/paddler_messaging/src/management_socket/balancer/notification_params/register_agent_params.rs +++ b/paddler_messaging/src/management_socket/balancer/notification_params/register_agent_params.rs @@ -2,7 +2,7 @@ use crate::slot_aggregated_status_snapshot::SlotAggregatedStatusSnapshot; use serde::Deserialize; use serde::Serialize; -#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] #[serde(deny_unknown_fields)] pub struct RegisterAgentParams { pub name: Option, diff --git a/paddler_messaging/src/management_socket/balancer/notification_params/update_agent_status_params.rs b/paddler_messaging/src/management_socket/balancer/notification_params/update_agent_status_params.rs index f6e17c39f..3d3c91c6b 100644 --- a/paddler_messaging/src/management_socket/balancer/notification_params/update_agent_status_params.rs +++ b/paddler_messaging/src/management_socket/balancer/notification_params/update_agent_status_params.rs @@ -2,7 +2,7 @@ use crate::slot_aggregated_status_snapshot::SlotAggregatedStatusSnapshot; use serde::Deserialize; use serde::Serialize; -#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] #[serde(deny_unknown_fields)] pub struct UpdateAgentStatusParams { pub slot_aggregated_status_snapshot: SlotAggregatedStatusSnapshot, diff --git a/paddler_messaging/src/slot_aggregated_status_snapshot.rs b/paddler_messaging/src/slot_aggregated_status_snapshot.rs index 94dfdbc7f..6b685dbce 100644 --- a/paddler_messaging/src/slot_aggregated_status_snapshot.rs +++ b/paddler_messaging/src/slot_aggregated_status_snapshot.rs @@ -6,7 +6,7 @@ use serde::Serialize; use crate::agent_issue::AgentIssue; use crate::agent_state_application_status::AgentStateApplicationStatus; -#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] +#[derive(Clone, Debug, Default, Deserialize, PartialEq, Serialize)] #[serde(deny_unknown_fields)] pub struct SlotAggregatedStatusSnapshot { pub desired_slots_total: i32, @@ -19,6 +19,7 @@ pub struct SlotAggregatedStatusSnapshot { pub slots_processing: i32, pub slots_total: i32, pub state_application_status: AgentStateApplicationStatus, + pub tokens_per_second: f64, pub uses_chat_template_override: bool, pub version: i32, } diff --git a/paddler_test_cluster_harness/src/agents_status/assert_slots_total_at_least.rs b/paddler_test_cluster_harness/src/agents_status/assert_slots_total_at_least.rs index 5a37df6ad..2c3690b87 100644 --- a/paddler_test_cluster_harness/src/agents_status/assert_slots_total_at_least.rs +++ b/paddler_test_cluster_harness/src/agents_status/assert_slots_total_at_least.rs @@ -39,6 +39,7 @@ mod tests { slots_processing: 0, slots_total, state_application_status: AgentStateApplicationStatus::Applied, + tokens_per_second: 0.0, uses_chat_template_override: false, }], } diff --git a/paddler_test_cluster_harness/src/agents_stream_watcher.rs b/paddler_test_cluster_harness/src/agents_stream_watcher.rs index 1c733b93b..d0dfcd043 100644 --- a/paddler_test_cluster_harness/src/agents_stream_watcher.rs +++ b/paddler_test_cluster_harness/src/agents_stream_watcher.rs @@ -234,6 +234,7 @@ mod tests { slots_processing: 0, slots_total, state_application_status: AgentStateApplicationStatus::Fresh, + tokens_per_second: 0.0, uses_chat_template_override: false, } } diff --git a/paddler_tests/src/make_agent_controller_without_remote_agent.rs b/paddler_tests/src/make_agent_controller_without_remote_agent.rs index 908d590a2..6cc4f0559 100644 --- a/paddler_tests/src/make_agent_controller_without_remote_agent.rs +++ b/paddler_tests/src/make_agent_controller_without_remote_agent.rs @@ -43,6 +43,7 @@ pub fn make_agent_controller_without_remote_agent(id: &str) -> AgentController { state_application_status_code: AtomicValue::::new( AgentStateApplicationStatus::Fresh as i32, ), + tokens_per_second: RwLock::new(0.0), uses_chat_template_override: AtomicValue::::new(false), } } diff --git a/paddler_tests/tests/harness_agents_watcher.rs b/paddler_tests/tests/harness_agents_watcher.rs index c6527c608..3990282e1 100644 --- a/paddler_tests/tests/harness_agents_watcher.rs +++ b/paddler_tests/tests/harness_agents_watcher.rs @@ -26,6 +26,7 @@ fn make_snapshot(agent_id: &str, slots_total: i32) -> AgentControllerPoolSnapsho slots_processing: 0, slots_total, state_application_status: AgentStateApplicationStatus::Applied, + tokens_per_second: 0.0, uses_chat_template_override: false, }], } diff --git a/resources/ts/components/AgentListAgentStatus.module.css b/resources/ts/components/AgentListAgentStatus.module.css index 351886a0f..060823781 100644 --- a/resources/ts/components/AgentListAgentStatus.module.css +++ b/resources/ts/components/AgentListAgentStatus.module.css @@ -2,7 +2,7 @@ align-items: center; display: grid; gap: var(--spacing-half); - grid-template-columns: 1fr auto; + grid-template-columns: 1fr auto auto; justify-items: flex-start; overflow: hidden; text-overflow: ellipsis; diff --git a/resources/ts/components/AgentListAgentStatus.tsx b/resources/ts/components/AgentListAgentStatus.tsx index ceb51f143..6630b262c 100644 --- a/resources/ts/components/AgentListAgentStatus.tsx +++ b/resources/ts/components/AgentListAgentStatus.tsx @@ -10,6 +10,7 @@ export function AgentListAgentStatus({ slots_processing, slots_total, state_application_status, + tokens_per_second, }, }: { agent: Agent; @@ -33,6 +34,11 @@ export function AgentListAgentStatus({ {slots_processing}/{slots_total}/{desired_slots_total} + {tokens_per_second > 0 && ( + + {tokens_per_second.toFixed(1)} tok/s + + )} ); case "AttemptedAndNotAppliable": From 81961b532b2d3eeb5d245ddf97a3182385e697e4 Mon Sep 17 00:00:00 2001 From: Nathan Easton Date: Fri, 14 Aug 2026 09:27:57 -0400 Subject: [PATCH 2/6] Change comparison to satisfy clippy --- paddler_balancer/src/agent_controller.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/paddler_balancer/src/agent_controller.rs b/paddler_balancer/src/agent_controller.rs index a110bac9e..6c1be5005 100644 --- a/paddler_balancer/src/agent_controller.rs +++ b/paddler_balancer/src/agent_controller.rs @@ -196,7 +196,7 @@ impl AgentController { self.set_model_path(model_path); } - if tokens_per_second != self.get_tokens_per_second() { + if (tokens_per_second - self.get_tokens_per_second()).abs() > 0.01 { changed = true; self.set_tokens_per_second(tokens_per_second); From 057687ad781c0a67d889354e08e47307f623a90d Mon Sep 17 00:00:00 2001 From: Nathan Easton Date: Fri, 14 Aug 2026 10:07:50 -0400 Subject: [PATCH 3/6] Fix using u64 instead of f64 causing comparison issues and loss of precision --- paddler_agent/src/token_throughput_meter.rs | 12 ++++++------ paddler_balancer/src/agent_controller_pool.rs | 3 ++- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/paddler_agent/src/token_throughput_meter.rs b/paddler_agent/src/token_throughput_meter.rs index cc274565d..07da000d7 100644 --- a/paddler_agent/src/token_throughput_meter.rs +++ b/paddler_agent/src/token_throughput_meter.rs @@ -17,7 +17,7 @@ const STALE_AFTER: Duration = Duration::from_secs(2); struct ThroughputWindow { tokens_per_second: f64, window_started_at: Instant, - window_token_count: u64, + window_token_count: f64, } /// Tracks generated tokens for a single agent and derives a continuously @@ -37,7 +37,7 @@ impl TokenThroughputMeter { window: Mutex::new(ThroughputWindow { tokens_per_second: 0.0, window_started_at: Instant::now(), - window_token_count: 0, + window_token_count: 0.0, }), } } @@ -48,11 +48,11 @@ impl TokenThroughputMeter { let mut window = self.window.lock(); let elapsed = window.window_started_at.elapsed(); - window.window_token_count += 1; + window.window_token_count += 1.0; if elapsed >= WINDOW_DURATION { - window.tokens_per_second = window.window_token_count as f64 / elapsed.as_secs_f64(); - window.window_token_count = 0; + window.tokens_per_second = window.window_token_count / elapsed.as_secs_f64(); + window.window_token_count = 0.0; window.window_started_at = Instant::now(); } } @@ -63,7 +63,7 @@ impl TokenThroughputMeter { pub fn tokens_per_second(&self) -> f64 { let window = self.window.lock(); - if window.window_started_at.elapsed() > STALE_AFTER && window.window_token_count == 0 { + if window.window_started_at.elapsed() > STALE_AFTER && window.window_token_count == 0.0 { return 0.0; } diff --git a/paddler_balancer/src/agent_controller_pool.rs b/paddler_balancer/src/agent_controller_pool.rs index 4eb63b239..43f1c109a 100644 --- a/paddler_balancer/src/agent_controller_pool.rs +++ b/paddler_balancer/src/agent_controller_pool.rs @@ -323,7 +323,8 @@ mod tests { pool.register_agent_controller("second".to_owned(), second) .unwrap(); - assert_eq!(pool.total_tokens_per_second(), 20.0); + let actual = pool.total_tokens_per_second(); + assert!((actual - 20.0).abs() < 0.0001); } #[test] From ec13804d67108a7da7e1660ac6d35e1d61178478 Mon Sep 17 00:00:00 2001 From: Nathan Easton Date: Fri, 14 Aug 2026 10:20:50 -0400 Subject: [PATCH 4/6] more fixes for clippy. --- paddler_agent/src/slot_aggregated_status.rs | 3 ++- paddler_agent/src/token_throughput_meter.rs | 6 ++++-- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/paddler_agent/src/slot_aggregated_status.rs b/paddler_agent/src/slot_aggregated_status.rs index b97480bb8..02ef00347 100644 --- a/paddler_agent/src/slot_aggregated_status.rs +++ b/paddler_agent/src/slot_aggregated_status.rs @@ -366,7 +366,8 @@ mod tests { let snapshot = status.make_snapshot().unwrap(); - assert_eq!(snapshot.tokens_per_second, 0.0); + let actual = snapshot.tokens_per_second; + assert!((actual - 0.0).abs() < 0.0001); } #[test] diff --git a/paddler_agent/src/token_throughput_meter.rs b/paddler_agent/src/token_throughput_meter.rs index 07da000d7..c4d456360 100644 --- a/paddler_agent/src/token_throughput_meter.rs +++ b/paddler_agent/src/token_throughput_meter.rs @@ -87,7 +87,8 @@ mod tests { fn reports_zero_before_any_tokens_are_recorded() { let meter = TokenThroughputMeter::new(); - assert_eq!(meter.tokens_per_second(), 0.0); + let actual = meter.tokens_per_second(); + assert!((actual - 0.0).abs() < 0.0001); } #[test] @@ -119,6 +120,7 @@ mod tests { meter.record_token(); sleep(STALE_AFTER + Duration::from_millis(50)); - assert_eq!(meter.tokens_per_second(), 0.0); + let actual = meter.tokens_per_second(); + assert!((actual - 0.0).abs() < 0.0001); } } From 37c2b67e668f26a19dd7159c95292b2ba372de4e Mon Sep 17 00:00:00 2001 From: Nathan Easton Date: Fri, 14 Aug 2026 10:44:09 -0400 Subject: [PATCH 5/6] Added delay of 10ms to stop flaky test --- paddler_agent/src/management_socket_client_service.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/paddler_agent/src/management_socket_client_service.rs b/paddler_agent/src/management_socket_client_service.rs index a0035bd0f..c719382e4 100644 --- a/paddler_agent/src/management_socket_client_service.rs +++ b/paddler_agent/src/management_socket_client_service.rs @@ -648,6 +648,8 @@ mod tests { let refused_addr = probe.local_addr().unwrap(); drop(probe); + // Give OS time to clean up socket to ensure immediate refusal. + tokio::time::sleep(Duration::from_millis(10)).await; let service = service_with_socket_url(format!( "ws://{refused_addr}/api/v1/agent_socket/test-agent" From 70f8b163de4e031eecc531492b8885d4f660fcea Mon Sep 17 00:00:00 2001 From: Nathan Easton Date: Sat, 15 Aug 2026 21:40:06 -0400 Subject: [PATCH 6/6] Remove 10ms delay and actually fixed the issue with the flaky test. --- .../src/management_socket_client_service.rs | 31 ++++++++++++++----- 1 file changed, 23 insertions(+), 8 deletions(-) diff --git a/paddler_agent/src/management_socket_client_service.rs b/paddler_agent/src/management_socket_client_service.rs index c719382e4..2b19d4646 100644 --- a/paddler_agent/src/management_socket_client_service.rs +++ b/paddler_agent/src/management_socket_client_service.rs @@ -648,20 +648,35 @@ mod tests { let refused_addr = probe.local_addr().unwrap(); drop(probe); - // Give OS time to clean up socket to ensure immediate refusal. - tokio::time::sleep(Duration::from_millis(10)).await; let service = service_with_socket_url(format!( "ws://{refused_addr}/api/v1/agent_socket/test-agent" )); - let shutdown = CancellationToken::new(); - let keep_alive_result = - tokio::time::timeout(SHUTDOWN_BUDGET, service.keep_connection_alive(shutdown)) - .await - .expect("connecting to a refused port must fail fast instead of blocking"); + // The kernel may take ~1s before it answers a connect to a just-closed + // port with ECONNREFUSED, so a single attempt could time out. Retry + // within the budget instead of assuming the refusal is immediate. + let deadline = tokio::time::Instant::now() + SHUTDOWN_BUDGET; + + loop { + let shutdown = CancellationToken::new(); + + let keep_alive_result = tokio::select! { + keep_alive_result = service.keep_connection_alive(shutdown) => keep_alive_result, + () = tokio::time::sleep_until(deadline) => { + panic!("connecting to a refused port must fail within the shutdown budget"); + } + }; + + if let Err(err) = keep_alive_result { + assert!( + err.to_string().contains("Connection refused"), + "expected a connection refusal, got: {err}" + ); - assert!(keep_alive_result.is_err()); + return; + } + } } #[tokio::test]