open-standards/wai/wai-rs/tests/dash_read_bounds.rs

//! Reading an MPD (`extensions/http-delivery.md` Appendix I) holds what the document
//! writes, not what its `@r` says, and reads an element's attributes in linear time.
//! Measured with a global allocator that counts the bytes live and their peak, and the
//! calling thread's CPU time.
//!
//! - A few kilobytes of 32 timelines, each one `S` of 2^20 segments, are read and
//!   compared with each other within a footprint bounded by the document's size: the
//!   runs are held as written and compared run by run, never expanded.
//! - Comparing them takes CPU time that does not grow with the segments a run lists.
//! - One element of 32000 attributes is read in time linear in its attributes: the
//!   names already read are kept in a set, not scanned again for each attribute.
//!
//! Its own test binary with a single test, so that no other test's allocations land in
//! the windows it measures.
#![cfg(feature = "http")]

use std::alloc::{GlobalAlloc, Layout, System};
use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
use std::time::Duration;

use wai::http_binding::dash_read::{check_dash_adjacent, read_mpd, read_xml};

/// The system allocator, counting the bytes live and their peak.
struct Count;
static CUR: AtomicUsize = AtomicUsize::new(0);
static PEAK: AtomicUsize = AtomicUsize::new(0);

fn grow(n: usize) {
    PEAK.fetch_max(CUR.fetch_add(n, Relaxed) + n, Relaxed);
}

unsafe impl GlobalAlloc for Count {
    unsafe fn alloc(&self, l: Layout) -> *mut u8 {
        grow(l.size());
        unsafe { System.alloc(l) }
    }
    unsafe fn alloc_zeroed(&self, l: Layout) -> *mut u8 {
        grow(l.size());
        unsafe { System.alloc_zeroed(l) }
    }
    unsafe fn realloc(&self, p: *mut u8, l: Layout, new: usize) -> *mut u8 {
        if new >= l.size() {
            grow(new - l.size());
        } else {
            CUR.fetch_sub(l.size() - new, Relaxed);
        }
        unsafe { System.realloc(p, l, new) }
    }
    unsafe fn dealloc(&self, p: *mut u8, l: Layout) {
        CUR.fetch_sub(l.size(), Relaxed);
        unsafe { System.dealloc(p, l) }
    }
}

#[global_allocator]
static ALLOC: Count = Count;

/// The calling thread's CPU time: on a loaded host wall-clock time grows with the load,
/// the thread's own CPU time only with its work.
#[cfg(unix)]
fn now() -> Duration {
    #[repr(C)]
    struct Timespec {
        tv_sec: i64,
        tv_nsec: i64,
    }
    unsafe extern "C" {
        fn clock_gettime(clock: i32, tp: *mut Timespec) -> i32;
    }
    #[cfg(target_vendor = "apple")]
    const CLOCK_THREAD_CPUTIME_ID: i32 = 16;
    #[cfg(not(target_vendor = "apple"))]
    const CLOCK_THREAD_CPUTIME_ID: i32 = 3;
    let mut ts = Timespec { tv_sec: 0, tv_nsec: 0 };
    assert_eq!(unsafe { clock_gettime(CLOCK_THREAD_CPUTIME_ID, &mut ts) }, 0, "clock_gettime");
    Duration::new(ts.tv_sec as u64, ts.tv_nsec as u32)
}

/// Elsewhere, wall-clock time from a fixed origin.
#[cfg(not(unix))]
fn now() -> Duration {
    static ORIGIN: std::sync::OnceLock<std::time::Instant> = std::sync::OnceLock::new();
    ORIGIN.get_or_init(std::time::Instant::now).elapsed()
}

/// The most bytes live at once beyond the start, while `f` runs.
fn peak<T>(f: impl FnOnce() -> T) -> (T, usize) {
    let start = CUR.load(Relaxed);
    PEAK.store(start, Relaxed);
    let out = f();
    (out, PEAK.load(Relaxed) - start)
}

/// The least CPU time of `f` and of `g` over seven runs each, taken in turn. On a virtual
/// machine a thread's CPU clock also counts the time its host held the virtual CPU, so a
/// busy host slows every run: taking the two in turn lets a busy spell slow both alike, and
/// the least of each is the run it disturbed least.
fn cpu_pair(f: impl Fn(), g: impl Fn()) -> (Duration, Duration) {
    let time = |h: &dyn Fn()| {
        let t = now();
        h();
        now() - t
    };
    (0..7).map(|_| (time(&f), time(&g))).fold((Duration::MAX, Duration::MAX), |(a, b), (x, y)| (a.min(x), b.min(y)))
}

/// `n` Representations, each a timeline of one `S` of 2^20 segments.
fn bomb(n: usize) -> String {
    bomb_of(n, 1 << 20)
}

/// `n` Representations, each a timeline of one `S` of `segments` segments.
fn bomb_of(n: usize, segments: u64) -> String {
    let r = segments - 1;
    let reps: String = (0..n)
        .map(|i| format!("<Representation id=\"r{i}\" bandwidth=\"1\"><SegmentList timescale=\"1\"><SegmentTimeline><S t=\"0\" d=\"1\" r=\"{r}\"/></SegmentTimeline></SegmentList></Representation>"))
        .collect();
    format!("<MPD><Period id=\"p\"><AdaptationSet>{reps}</AdaptationSet></Period></MPD>")
}

/// One element of `n` attributes.
fn wide(n: usize) -> String {
    let attrs: String = (0..n).map(|i| format!(" a{i}=\"\"")).collect();
    format!("<MPD{attrs}/>")
}

#[test]
fn an_mpd_is_held_as_written_and_read_in_linear_time() {
    // 32 timelines of 2^20 segments in about 5 KB: an expanded reader holds 2^25
    // segments (hundreds of megabytes); this one, a run each.
    for n in [1usize, 8, 32] {
        let text = bomb(n);
        let (periods, held) = peak(|| read_mpd(&text).unwrap());
        let tls = &periods[0].sets[0].timelines;
        assert_eq!(tls.len(), n);
        assert!(tls.iter().all(|t| t.runs.len() == 1 && t.segment_count() == 1 << 20));
        let bound = 64 * text.len() + (1 << 20);
        assert!(held <= bound, "{n} timelines in {} bytes held {held} bytes (bound {bound})", text.len());
        // Compared with themselves, each of 32 WAI timelines against 32 classical ones.
        let (adjacent, held) = peak(|| check_dash_adjacent(&text, &text));
        adjacent.unwrap();
        assert!(held <= 2 * bound, "comparing {n} timelines held {held} bytes");
    }
    // The comparison is run by run: 32 timelines of 2^20 segments each, compared with
    // themselves, take about the CPU time of 32 timelines of 2^10; one that steps a
    // segment at a time takes about a thousand times as long.
    let (few, many) = (bomb_of(32, 1 << 10), bomb_of(32, 1 << 20));
    let (t_few, t_many) = cpu_pair(|| check_dash_adjacent(&few, &few).unwrap(), || check_dash_adjacent(&many, &many).unwrap());
    assert!(t_many <= t_few * 4 + Duration::from_millis(20), "2^20 segments a timeline took {t_many:?}, 2^10 took {t_few:?}");
    // One element of 32000 attributes against one of 1000: linear, so about 32 times the
    // CPU time, and up to about three times that again where the larger element's sets
    // leave a cache a busy host shares; a reader that scans the names read so far for each
    // one takes about a thousand times. The sizes are 32 apart so that no cache effect
    // reaches the bound between the two.
    let (short, long) = (wide(1_000), wide(32_000));
    read_xml(&long).unwrap();
    let (t_short, t_long) = cpu_pair(|| drop(read_xml(&short)), || drop(read_xml(&long)));
    assert!(t_long <= t_short * 256 + Duration::from_millis(20), "32000 attributes took {t_long:?}, 1000 took {t_short:?}");
}