163 lines
7.6 KiB
Python
163 lines
7.6 KiB
Python
# Copyright (c) Alibaba, Inc. and its affiliates.
|
|
"""Unit tests for WorkloadTimeline + WorkloadThroughput (P4)."""
|
|
import unittest
|
|
|
|
from evalscope.perf.utils.benchmark_util import BenchmarkData
|
|
from evalscope.perf.utils.workload_timeline import WorkloadThroughput, WorkloadTimeline
|
|
|
|
|
|
def _ok(*, start: float, completed: float, prompt: int, completion: int, cached: int = 0,
|
|
is_warmup: bool = False, success: bool = True) -> BenchmarkData:
|
|
d = BenchmarkData(success=success)
|
|
d.is_warmup = is_warmup
|
|
d.start_time = start
|
|
d.completed_time = completed
|
|
d.prompt_tokens = prompt
|
|
d.completion_tokens = completion
|
|
d.cached_tokens = cached
|
|
return d
|
|
|
|
|
|
class TestWorkloadTimelineFeed(unittest.TestCase):
|
|
|
|
def test_skips_warmup_failed_and_zero_timestamps(self):
|
|
tl = WorkloadTimeline()
|
|
tl.feed(_ok(start=1.0, completed=2.0, prompt=10, completion=5, is_warmup=True))
|
|
tl.feed(_ok(start=1.0, completed=2.0, prompt=10, completion=5, success=False))
|
|
tl.feed(_ok(start=0.0, completed=0.0, prompt=10, completion=5)) # zero timestamps
|
|
self.assertEqual(tl.n_points, 0)
|
|
self.assertEqual(tl.wall_time, 0.0)
|
|
|
|
def test_cumulative_counts_and_wall_time(self):
|
|
tl = WorkloadTimeline()
|
|
tl.feed(_ok(start=100.0, completed=101.0, prompt=20, completion=10, cached=0))
|
|
tl.feed(_ok(start=100.5, completed=102.0, prompt=30, completion=15, cached=20))
|
|
tl.feed(_ok(start=101.0, completed=104.0, prompt=40, completion=20, cached=35))
|
|
# Cumulative: completion = 10 + 15 + 20 = 45
|
|
# new_prompt = 20 + (30-20) + (40-35) = 35
|
|
# cached_prompt = 0 + 20 + 35 = 55
|
|
# total_prompt = new + cached = 90 = 20 + 30 + 40 ✓
|
|
self.assertEqual(tl.n_points, 3)
|
|
last_overall = tl.overall_rates()
|
|
# wall_time = 104 - 100 = 4
|
|
self.assertAlmostEqual(tl.wall_time, 4.0)
|
|
# total / wall = 90 / 4 = 22.5
|
|
self.assertAlmostEqual(last_overall[0], 22.5)
|
|
# new / wall = 35 / 4 = 8.75
|
|
self.assertAlmostEqual(last_overall[1], 8.75)
|
|
# cached / wall = 55 / 4 = 13.75
|
|
self.assertAlmostEqual(last_overall[2], 13.75)
|
|
# completion / wall = 45 / 4 = 11.25
|
|
self.assertAlmostEqual(last_overall[3], 11.25)
|
|
|
|
def test_wall_start_locked_on_first_feed(self):
|
|
"""A later request with a smaller start_time must not retroactively
|
|
shift wall_start; otherwise already-appended ``t`` offsets would
|
|
silently become wrong (they are never rewritten).
|
|
"""
|
|
tl = WorkloadTimeline()
|
|
tl.feed(_ok(start=100.0, completed=101.0, prompt=10, completion=5))
|
|
# Second request started *earlier* but completed later (worker race).
|
|
tl.feed(_ok(start=99.0, completed=102.0, prompt=10, completion=5))
|
|
# wall_start stayed at 100; second point's t = 102 - 100 = 2.0.
|
|
# Old "min" logic would have made wall_start=99 and silently broken
|
|
# the first point's already-stored t (left at 1.0 instead of 2.0).
|
|
self.assertEqual(tl._wall_start, 100.0)
|
|
self.assertAlmostEqual(tl._points[0].t, 1.0)
|
|
self.assertAlmostEqual(tl._points[1].t, 2.0)
|
|
|
|
def test_new_prompt_clamped_when_cached_exceeds_prompt(self):
|
|
tl = WorkloadTimeline()
|
|
# Pathological: server reports cached > prompt (chat-template inflation).
|
|
# WorkloadTimeline must clamp new_prompt to >= 0 to keep rates non-negative.
|
|
tl.feed(_ok(start=1.0, completed=2.0, prompt=10, completion=5, cached=50))
|
|
rates = tl.overall_rates()
|
|
self.assertGreaterEqual(rates[1], 0.0) # new_prompt non-negative
|
|
|
|
|
|
class TestSteadyStateAndLastWindow(unittest.TestCase):
|
|
|
|
def test_last_window_falls_back_to_overall_when_run_shorter(self):
|
|
tl = WorkloadTimeline()
|
|
tl.feed(_ok(start=100.0, completed=101.0, prompt=20, completion=10))
|
|
tl.feed(_ok(start=100.5, completed=105.0, prompt=20, completion=10))
|
|
# wall = 5s < 30s; last_window_rates(30) should equal overall_rates
|
|
self.assertEqual(tl.last_window_rates(30.0), tl.overall_rates())
|
|
|
|
def test_last_window_uses_anchor_inside_run(self):
|
|
tl = WorkloadTimeline()
|
|
# Build a 10s timeline so a 5s tail window discards the first half.
|
|
# Each point is well-separated by 1s for deterministic anchor selection.
|
|
tl.feed(_ok(start=0.0, completed=1.0, prompt=10, completion=10)) # t=1
|
|
tl.feed(_ok(start=0.5, completed=3.0, prompt=10, completion=10)) # t=3
|
|
tl.feed(_ok(start=1.0, completed=6.0, prompt=10, completion=10)) # t=6
|
|
tl.feed(_ok(start=1.5, completed=10.0, prompt=10, completion=10)) # t=10
|
|
# wall = 10s. Tail window = last 5s -> from t=5 anchor (latest with t<=5 is t=3).
|
|
# delta_t = 10 - 3 = 7; delta_completion = 40 - 20 = 20
|
|
rates = tl.last_window_rates(5.0)
|
|
self.assertAlmostEqual(rates[3], 20 / 7, places=4)
|
|
|
|
def test_steady_state_drops_warmup(self):
|
|
tl = WorkloadTimeline()
|
|
tl.feed(_ok(start=0.0, completed=1.0, prompt=10, completion=10)) # t=1
|
|
tl.feed(_ok(start=0.5, completed=2.0, prompt=10, completion=10)) # t=2
|
|
tl.feed(_ok(start=1.0, completed=8.0, prompt=10, completion=10)) # t=8
|
|
tl.feed(_ok(start=1.5, completed=10.0, prompt=10, completion=10)) # t=10
|
|
# wall = 10s, warmup_frac = 0.2 -> cutoff_t = 2s.
|
|
# Anchor = latest point with t<=2 = the t=2 point. delta_t = 10-2 = 8;
|
|
# delta_completion = 40 - 20 = 20.
|
|
rates = tl.steady_state_rates(warmup_frac=0.2)
|
|
self.assertAlmostEqual(rates[3], 20 / 8, places=4)
|
|
|
|
def test_steady_state_falls_back_to_overall_when_anchor_is_last(self):
|
|
# Single point can't carve a steady-state window.
|
|
tl = WorkloadTimeline()
|
|
tl.feed(_ok(start=0.0, completed=1.0, prompt=10, completion=10))
|
|
self.assertEqual(tl.steady_state_rates(0.2), tl.overall_rates())
|
|
|
|
|
|
class TestThroughputSnapshot(unittest.TestCase):
|
|
|
|
def test_to_summary_emits_four_rows(self):
|
|
tl = WorkloadTimeline()
|
|
tl.feed(_ok(start=0.0, completed=1.0, prompt=10, completion=10))
|
|
tl.feed(_ok(start=0.5, completed=2.0, prompt=20, completion=20, cached=10))
|
|
snap = tl.to_summary(last_window_s=30.0, warmup_frac=0.2)
|
|
self.assertEqual(snap.n_samples, 2)
|
|
self.assertEqual(len(snap.rows), 4)
|
|
metrics = [r.metric for r in snap.rows]
|
|
self.assertEqual(
|
|
metrics,
|
|
['Total Prompt tok/s', 'New Prompt tok/s', 'Cached Prompt tok/s', 'Completion tok/s'],
|
|
)
|
|
|
|
def test_to_table_labels_reflect_window_and_warmup(self):
|
|
tl = WorkloadTimeline()
|
|
tl.feed(_ok(start=0.0, completed=1.0, prompt=10, completion=10))
|
|
snap = tl.to_summary(last_window_s=60.0, warmup_frac=0.25)
|
|
table = snap.to_table()
|
|
self.assertIn('Overall', table)
|
|
self.assertIn('Last 60s', table)
|
|
self.assertIn('Steady (drop 25%)', table)
|
|
|
|
def test_empty_summary_is_empty(self):
|
|
snap = WorkloadThroughput()
|
|
self.assertTrue(snap.is_empty())
|
|
self.assertEqual(snap.to_table(), '')
|
|
|
|
def test_raw_points_dict_shape(self):
|
|
tl = WorkloadTimeline()
|
|
tl.feed(_ok(start=0.0, completed=1.5, prompt=20, completion=10, cached=5))
|
|
d = tl.to_raw_points_dict()
|
|
self.assertIn('points', d)
|
|
self.assertEqual(len(d['points']), 1)
|
|
p = d['points'][0]
|
|
for key in ('t', 'cum_completion', 'cum_new_prompt', 'cum_cached_prompt'):
|
|
self.assertIn(key, p)
|
|
self.assertAlmostEqual(p['t'], 1.5)
|
|
self.assertEqual(p['cum_completion'], 10)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(buffer=False)
|