Fabio
c36510c21e
Some checks failed
Linux Build / Linux cpu (push) Has been cancelled
Linux Build / Linux cpu-clang (push) Has been cancelled
Linux Build / Linux vulkan (push) Has been cancelled
mac-build / macOS CPU (push) Has been cancelled
Nix Build / Nix (cpu) (push) Has been cancelled
Nix Build / Nix (vulkan) (push) Has been cancelled
windows-build / Windows CPU (push) Has been cancelled
150 lines
8.4 KiB
Python
150 lines
8.4 KiB
Python
"""Compare slot-scheduler busy rejection before/after a fix using real Canary ASR.
|
|
|
|
Requires two server executables, a Canary GGUF/spec and a PCM WAV fixture.
|
|
Uses the standard library. Reports HTTP rejection latency separately from
|
|
inference time and verifies outputs, unload/reload and queue draining.
|
|
"""
|
|
|
|
import argparse
|
|
import concurrent.futures as cf
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import statistics
|
|
import threading
|
|
import time
|
|
import wave
|
|
|
|
from server_test_utils import Server
|
|
|
|
|
|
def wait_active(server, count):
|
|
deadline = time.monotonic() + 30
|
|
while time.monotonic() < deadline:
|
|
state = server.status('canary')
|
|
if state['active_slots'] == count:
|
|
return state
|
|
time.sleep(.001)
|
|
raise AssertionError(f'Expected {count} active slots: {state}')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
for name in ('before-server', 'after-server', 'model', 'spec', 'audio', 'output-dir'):
|
|
parser.add_argument('--' + name, type=Path, required=True)
|
|
parser.add_argument('--parallel-jobs', action='store_true', help='enable the parallel runtime in both compared builds')
|
|
parser.add_argument('--backend', choices=('cuda', 'vulkan'), required=True)
|
|
parser.add_argument('--device', type=int, default=0)
|
|
parser.add_argument('--repeats', type=int, default=5)
|
|
parser.add_argument('--slots', type=int, nargs='+', default=[1], choices=(1, 2, 3, 4),
|
|
help='use 1 for the standalone framework; larger counts require model support')
|
|
args = parser.parse_args()
|
|
if not args.parallel_jobs and args.slots != [1]:
|
|
parser.error("multiple slots require --parallel-jobs")
|
|
for name, value in vars(args).items():
|
|
if isinstance(value, Path):
|
|
setattr(args, name, value.resolve())
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
|
# A longer fixture keeps the warm inference active throughout the rejection
|
|
# probe even on fast GPUs. Preserve all PCM samples; do not resample.
|
|
audio = args.output_dir / 'speech-repeated.wav'
|
|
with wave.open(str(args.audio), 'rb') as source:
|
|
params, frames = source.getparams(), source.readframes(source.getnframes())
|
|
with wave.open(str(audio), 'wb') as dest:
|
|
dest.setparams(params)
|
|
dest.writeframes(frames * 4)
|
|
payload = {'audio': str(audio), 'language': 'en', 'max_tokens': 512}
|
|
model = {'id': 'canary', 'family': 'canary_asr', 'task': 'asr', 'mode': 'offline',
|
|
'path': str(args.model), 'model_spec_override': str(args.spec)}
|
|
records = []
|
|
result_path = args.output_dir / 'results.json'
|
|
metadata = {'backend': args.backend, 'device': args.device, 'timeout_ms': 15,
|
|
'input_duration_s': params.nframes * 4 / params.framerate,
|
|
'model': str(args.model), 'model_spec': str(args.spec),
|
|
'source_audio': str(args.audio),
|
|
'executables': {label: {'path': str(exe), 'sha256': hashlib.sha256(exe.read_bytes()).hexdigest()}
|
|
for label, exe in [('before', args.before_server), ('after', args.after_server)]}}
|
|
|
|
def save():
|
|
result_path.write_text(json.dumps({'metadata': metadata, 'records': records}, indent=2), encoding='utf-8')
|
|
|
|
def timed(server, body):
|
|
started = time.perf_counter()
|
|
status, response = server.api('/v1/tasks/run', body)
|
|
elapsed = (time.perf_counter() - started) * 1000
|
|
timing = response.pop('timing', None)
|
|
return {'status': status, 'response': response, 'timing': timing, 'http_ms': elapsed}
|
|
|
|
references = {}
|
|
for label, exe in [('before', args.before_server), ('after', args.after_server)]:
|
|
for slots in args.slots:
|
|
folder = args.output_dir / f'{label}-{slots}'
|
|
with Server(exe, [{**model, 'slots': slots}] if args.parallel_jobs else [model],
|
|
args.backend, args.device, folder, parallel_jobs=args.parallel_jobs) as server:
|
|
body = {'model': 'canary', 'request': payload}
|
|
cold = timed(server, body)
|
|
warm = timed(server, body)
|
|
assert cold['status'] == warm['status'] == 200, (cold, warm)
|
|
assert cold['response'] == warm['response'], ('Cold/warm Canary reference differs', cold, warm)
|
|
if label == 'before':
|
|
references[slots] = warm['response']
|
|
assert warm['response'] == references[slots]
|
|
records.append({'build': label, 'slots': slots, 'test': 'reference', 'cold': cold, 'warm': warm})
|
|
save()
|
|
# Full pools, plus a partially occupied pool blocked by unload.
|
|
scenarios = [(False, slots), (True, slots)]
|
|
if slots > 1:
|
|
scenarios.append((True, 1))
|
|
with cf.ThreadPoolExecutor(max_workers=slots + 2) as pool:
|
|
for management_waiting, occupied in scenarios:
|
|
for repeat in range(args.repeats):
|
|
barrier = threading.Barrier(occupied + 1)
|
|
|
|
def running():
|
|
barrier.wait(timeout=30)
|
|
return timed(server, body)
|
|
|
|
jobs = [pool.submit(running) for _ in range(occupied)]
|
|
barrier.wait(timeout=30)
|
|
wait_active(server, occupied)
|
|
# Give every occupied lease time to exceed 15 ms.
|
|
time.sleep(.025)
|
|
unload = None
|
|
if management_waiting:
|
|
unload = pool.submit(server.api, '/v1/tasks/unload_all_models', {})
|
|
time.sleep(.01)
|
|
assert not unload.done(), 'Unload finished before rejection probe'
|
|
assert all(not job.done() for job in jobs), 'Fixture inference too short'
|
|
rejection = timed(server, {**body, 'busy_timeout_ms': 15})
|
|
assert rejection['status'] == 503, rejection
|
|
error = json.dumps(rejection['response'])
|
|
expected_error = ('timed out waiting for a slot' if label == 'before' and management_waiting
|
|
else 'slots exceeded busy_timeout_ms')
|
|
assert expected_error in error, (label, management_waiting, rejection)
|
|
outputs = [job.result() for job in jobs]
|
|
assert all(x['status'] == 200 and x['response'] == references[slots] for x in outputs), outputs
|
|
if unload:
|
|
assert unload.result()[0] == 200
|
|
assert not server.status('canary')['loaded']
|
|
reload = timed(server, body)
|
|
assert reload['status'] == 200 and reload['response'] == references[slots], reload
|
|
# Keep each rejection probe on a warm server.
|
|
assert timed(server, body)['response'] == references[slots]
|
|
state = server.status('canary')
|
|
assert state['active_slots'] == state['queued_requests'] == 0, state
|
|
record = {'build': label, 'slots': slots, 'occupied': occupied,
|
|
'management_waiting': management_waiting, 'repeat': repeat,
|
|
'test': 'busy_rejection', 'rejection': rejection,
|
|
'running_outputs': outputs, 'unload_reload_exact': bool(unload)}
|
|
records.append(record)
|
|
save()
|
|
values = [r['rejection']['http_ms'] for r in records if r.get('test') == 'busy_rejection'
|
|
and r['build'] == label and r['slots'] == slots and r['occupied'] == occupied
|
|
and r['management_waiting'] == management_waiting]
|
|
print(f'{args.backend} {label} slots={slots} active={occupied} unload={management_waiting}: '
|
|
f'median rejection={statistics.median(values):.3f} ms, exact outputs', flush=True)
|
|
print('PASS real HTTP overdue rejection, concurrent parity and unload/reload', flush=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|