//! how fast a network reader may replay from its cursor.
//!
//! a replay reads and encodes rows as fast as its reader takes them, so one
//! client asking for cursor 0 could keep a core busy for as long as it likes.
//! capping it at a multiple of what keeping up with the live stream costs
//! still lets it catch up, while what it costs stays in proportion to what
//! hydrant writes anyway.
//!
//! the cost is counted in bytes sent, because that's what operators think in
//! and frames run from about 500 bytes (`/stream`) to a few kb (relay commits
//! carrying their blocks). rows a filter drops cost a fixed amount instead.
use std::num::NonZeroU64;
use std::time::{Duration, Instant};
use crate::config::{Config, StreamKind};
/// what reading a row and dropping it costs, in sent bytes. measured on jetstream at about
/// 250 ns a filtered row against 8 us for a sent 660 byte one, so about 25 bytes, rounded up
/// because a real scan also pays for disk reads a test db mostly doesn't
const FILTERED_ROW_BYTES: u64 = 32;
/// the pace a network reader's cursor replay is held to.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ReplayCap {
/// times what keeping up with the live stream costs this reader
speed: u32,
/// bytes a second a replay may always use, however quiet the box is
floor: NonZeroU64,
}
impl ReplayCap {
pub(crate) fn new(speed: u32, floor: NonZeroU64) -> Self {
Self { speed, floor }
}
/// none when `stream` isn't in `HYDRANT_THROTTLED_STREAMS`.
pub(crate) fn from_config(config: &Config, stream: StreamKind) -> Option<Self> {
// validate refuses a zero floor, this only keeps a config built by hand from dividing by it
let floor = NonZeroU64::new(config.stream_replay_floor_bytes).unwrap_or(NonZeroU64::MIN);
config
.throttled_streams
.contains(&stream)
.then(|| Self::new(config.stream_replay_speed, floor))
}
}
/// what one replay read went through.
pub(crate) struct Read {
/// rows the read scanned, sent or not
pub(crate) scanned: usize,
pub(crate) sent: usize,
pub(crate) sent_bytes: usize,
}
impl Read {
fn cost(&self) -> u64 {
let filtered = self.scanned.saturating_sub(self.sent) as u64;
self.sent_bytes as u64 + filtered * FILTERED_ROW_BYTES
}
}
/// how often the live rate is sampled. shorter samples would see single commits as bursts
const SAMPLE_EVERY: Duration = Duration::from_secs(1);
/// about how long the live rate takes to follow a change
const RATE_HORIZON: Duration = Duration::from_secs(10);
/// paces one replay. it takes the time as an argument so tests can run it on a made up clock.
pub(crate) struct Pacer {
cap: ReplayCap,
/// bytes the replay may use right away. a read that costs more leaves it below zero, and
/// the wait pays that back
balance: f64,
refilled: Instant,
/// what this replay cost so far and over how many rows, so the live rate, which is
/// counted in rows, can be priced the way this reader's filters price a row
cost: u64,
scanned: u64,
/// stream positions a second, averaged
live_rate: f64,
sampled: Instant,
written_at_sample: u64,
}
impl Pacer {
/// `written` is how many positions the stream has taken so far.
pub(crate) fn new(cap: ReplayCap, written: u64, now: Instant) -> Self {
Self {
cap,
// so a replay that fits in a second of the floor never waits
balance: cap.floor.get() as f64,
refilled: now,
cost: 0,
scanned: 0,
live_rate: 0.0,
sampled: now,
written_at_sample: written,
}
}
/// bytes a second the replay may use right now
fn budget(&self) -> f64 {
let row = self.cost as f64 / self.scanned.max(1) as f64;
(self.live_rate * row * f64::from(self.cap.speed)).max(self.cap.floor.get() as f64)
}
/// charges a read, and says how long to wait before the next.
pub(crate) fn charge(&mut self, read: Read, written: u64, now: Instant) -> Duration {
self.sample(written, now);
let cost = read.cost();
self.cost += cost;
self.scanned += read.scanned as u64;
let budget = self.budget();
let idle = now.saturating_duration_since(self.refilled).as_secs_f64();
// at most a second of budget banked, so a reader that idled can't come back with a burst
self.balance = (self.balance + idle * budget).min(budget) - cost as f64;
self.refilled = now;
if self.balance >= 0.0 {
Duration::ZERO
} else {
Duration::from_secs_f64(-self.balance / budget)
}
}
fn sample(&mut self, written: u64, now: Instant) {
let elapsed = now.saturating_duration_since(self.sampled);
if elapsed < SAMPLE_EVERY {
return;
}
let secs = elapsed.as_secs_f64();
let rate = written.saturating_sub(self.written_at_sample) as f64 / secs;
let weight = 1.0 - (-secs / RATE_HORIZON.as_secs_f64()).exp();
self.live_rate += (rate - self.live_rate) * weight;
self.sampled = now;
self.written_at_sample = written;
}
}
#[cfg(test)]
mod tests {
use super::*;
fn cap(speed: u32, floor: u64) -> ReplayCap {
ReplayCap::new(speed, NonZeroU64::new(floor).unwrap())
}
fn secs(s: f64) -> Duration {
Duration::from_secs_f64(s)
}
/// `rows` rows that each send a `frame` byte frame
fn sent(rows: usize, frame: usize) -> Read {
Read {
scanned: rows,
sent: rows,
sent_bytes: rows * frame,
}
}
/// bytes a reader sends in `over` reading 250 row chunks of `frame` byte frames whenever
/// the pacer lets it, while the stream takes `live` positions a second
fn replayed(cap: ReplayCap, live: f64, frame: usize, over: Duration) -> usize {
let start = Instant::now();
let written = |at: Instant| (at.duration_since(start).as_secs_f64() * live) as u64;
let mut pacer = Pacer::new(cap, 0, start);
let mut now = start;
let mut bytes = 0;
while now < start + over {
bytes += 250 * frame;
now += pacer.charge(sent(250, frame), written(now), now);
}
bytes
}
#[test]
fn a_replay_on_a_quiet_box_runs_at_the_floor() {
// one second banked, then the floor
let bytes = replayed(cap(10, 1_000_000), 0.0, 500, secs(60.0));
assert!((61_000_000..=61_125_000).contains(&bytes), "{bytes}");
}
#[test]
fn a_replay_runs_at_its_speed_times_what_the_live_stream_costs_it() {
// 5000 rows a second of 2 kb frames is 10 MB a second live, so 100 MB a second at 10x
let over = |s| replayed(cap(10, 1_000_000), 5000.0, 2000, secs(s));
let total = over(120.0);
// the rate climbs over the first few horizons, so the second minute is all 100 MB a second
let late = total - over(60.0);
assert!((5_950_000_000..=6_050_000_000).contains(&late), "{late}");
assert!(total <= 100_000_000 * 120 + 1_000_000 + 500_000, "{total}");
}
#[test]
fn a_live_rate_under_the_floor_leaves_the_floor() {
// 100 rows a second of 500 bytes at 10x is 500 kb a second, under the 10 MB floor
let bytes = replayed(cap(10, 10_000_000), 100.0, 500, secs(60.0));
assert!((610_000_000..=610_125_000).contains(&bytes), "{bytes}");
}
#[test]
fn filtered_rows_cost_a_fixed_amount_and_sent_ones_their_bytes() {
let now = Instant::now();
let mut pacer = Pacer::new(cap(10, 32_000), 0, now);
let filtered = Read {
scanned: 1000,
sent: 0,
sent_bytes: 0,
};
// a second of floor banked, then a whole second owed for the next 1000 dropped rows
assert_eq!(pacer.charge(filtered, 0, now), Duration::ZERO);
let filtered = Read {
scanned: 1000,
sent: 0,
sent_bytes: 0,
};
assert_eq!(pacer.charge(filtered, 0, now), secs(1.0));
let mixed = Read {
scanned: 1000,
sent: 500,
sent_bytes: 16_000,
};
assert_eq!(pacer.charge(mixed, 0, now + secs(1.0)), secs(1.0));
}
#[test]
fn a_read_bigger_than_the_balance_waits_it_back() {
let now = Instant::now();
let mut pacer = Pacer::new(cap(10, 1000), 0, now);
assert_eq!(pacer.charge(sent(10, 100), 0, now), Duration::ZERO);
assert_eq!(pacer.charge(sent(5, 100), 0, now), secs(0.5));
// paid back at exactly that point, then charged again
assert_eq!(pacer.charge(sent(5, 50), 0, now + secs(0.5)), secs(0.25));
}
#[test]
fn idling_banks_at_most_a_second() {
let now = Instant::now();
let mut pacer = Pacer::new(cap(10, 1000), 0, now);
pacer.charge(sent(10, 100), 0, now);
let later = now + secs(3600.0);
assert_eq!(pacer.charge(sent(10, 100), 0, later), Duration::ZERO);
assert_eq!(pacer.charge(sent(10, 100), 0, later), secs(1.0));
}
#[test]
fn the_rate_follows_the_head_within_a_few_horizons() {
let start = Instant::now();
let mut pacer = Pacer::new(cap(1, 1), 0, start);
for s in 1..=60 {
pacer.sample(s * 1000, start + secs(s as f64));
}
assert!(
(pacer.live_rate - 1000.0).abs() < 5.0,
"{}",
pacer.live_rate
);
// the head stops, so the rate falls back toward nothing
for s in 61..=120 {
pacer.sample(60_000, start + secs(s as f64));
}
assert!(pacer.live_rate < 5.0, "{}", pacer.live_rate);
}
#[test]
fn only_the_listed_streams_are_capped() {
let config = Config {
throttled_streams: vec![StreamKind::Relay],
..Config::default()
};
assert_eq!(ReplayCap::from_config(&config, StreamKind::Jetstream), None);
assert_eq!(ReplayCap::from_config(&config, StreamKind::Indexer), None);
assert_eq!(
ReplayCap::from_config(&config, StreamKind::Relay),
Some(cap(10, 10_000_000))
);
}
}