142 lines
4.5 KiB
Python
142 lines
4.5 KiB
Python
"""Small synthetic scheduler lifecycle tests; no models, GPU or source IO."""
|
|
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from k1link.perception.streaming_queue import StreamMailbox
|
|
from k1link.perception.streaming_scheduler import SerialGpuStage
|
|
|
|
|
|
def bundle(sequence, size=10):
|
|
return {"sequence": sequence, "payload_bytes": size}
|
|
|
|
|
|
def test_finish_drains_but_cancel_drops_pending_without_freeing_active_owner():
|
|
queue = StreamMailbox()
|
|
first = bundle(0)
|
|
queue.put(first)
|
|
assert queue.take() is first
|
|
queue.put(bundle(1))
|
|
queue.finish()
|
|
assert not queue.put(bundle(2))
|
|
assert queue.bytes == 20 and not queue.quiescent
|
|
queue.cancel()
|
|
assert queue.take() is None and queue.bytes == 10
|
|
assert queue.drop_counts == {"ingress-closed": 1, "cancelled": 1}
|
|
queue.release(first)
|
|
assert queue.quiescent
|
|
with pytest.raises(ValueError, match="ownership"):
|
|
queue.release(first)
|
|
|
|
|
|
def test_oversized_message_does_not_evict_accepted_small_input():
|
|
queue = StreamMailbox(byte_limit=100)
|
|
first = bundle(0)
|
|
queue.put(first)
|
|
assert not queue.put(bundle(1, 101))
|
|
assert queue.take() is first and queue.bytes == 10
|
|
queue.release(first)
|
|
|
|
|
|
def test_diagnostic_history_is_bounded_and_total_drops_remain_exact():
|
|
queue = StreamMailbox()
|
|
# Boundary-sized synthetic fixture, not a local sustained-load benchmark.
|
|
for sequence in range(260):
|
|
queue.put(bundle(sequence))
|
|
assert queue.dropped_count == 258 and len(queue.dropped) == 256
|
|
assert queue.peak_pending == 2 and queue.bytes == 20
|
|
queue.cancel()
|
|
assert queue.dropped_count == 260 and queue.quiescent
|
|
|
|
|
|
@pytest.mark.parametrize("sequence,size", [(True, 1), (-1, 1), (0, 0), (0, True)])
|
|
def test_invalid_admission_is_rejected(sequence, size):
|
|
queue = StreamMailbox()
|
|
with pytest.raises(ValueError):
|
|
queue.put(bundle(sequence, size))
|
|
assert queue.bytes == 0
|
|
|
|
|
|
def test_release_uses_admitted_size_and_does_not_allow_reusing_owned_input():
|
|
queue = StreamMailbox()
|
|
first = bundle(0)
|
|
queue.put(first)
|
|
assert queue.take() is first
|
|
first.update(sequence=1, payload_bytes=1000)
|
|
with pytest.raises(ValueError, match="owned"):
|
|
queue.put(first)
|
|
with pytest.raises(ValueError, match="ownership"):
|
|
queue.release(dict(first))
|
|
queue.release(first)
|
|
assert queue.bytes == 0
|
|
with pytest.raises(ValueError, match="increase"):
|
|
queue.put(bundle(0))
|
|
|
|
|
|
def test_failed_gpu_releases_its_input_and_discards_pending():
|
|
started, proceed = threading.Event(), threading.Event()
|
|
queue = StreamMailbox()
|
|
|
|
def compute(_bundle):
|
|
started.set()
|
|
assert proceed.wait(1)
|
|
raise ValueError("synthetic failure")
|
|
|
|
stage = SerialGpuStage(queue, compute, threading.Event())
|
|
try:
|
|
queue.put(bundle(0))
|
|
assert started.wait(1)
|
|
queue.put(bundle(1))
|
|
proceed.set()
|
|
with pytest.raises(RuntimeError, match="synthetic failure"):
|
|
stage.take()
|
|
assert queue.quiescent and queue.bytes == 0
|
|
assert queue.drop_counts == {"gpu-failed": 2}
|
|
finally:
|
|
proceed.set()
|
|
assert stage.close()
|
|
|
|
|
|
def test_timed_out_stop_retains_ownership_until_callback_actually_exits():
|
|
started, proceed = threading.Event(), threading.Event()
|
|
queue = StreamMailbox()
|
|
|
|
def compute(_bundle):
|
|
started.set()
|
|
assert proceed.wait(1)
|
|
return "must not be published after cancellation"
|
|
|
|
stage = SerialGpuStage(queue, compute, threading.Event())
|
|
try:
|
|
queue.put(bundle(0))
|
|
assert started.wait(1)
|
|
assert not stage.close(timeout=0)
|
|
assert queue.bytes == 10 and not queue.quiescent
|
|
assert stage.take() is None
|
|
proceed.set()
|
|
assert stage.finished.wait(1)
|
|
assert stage.close() and queue.quiescent
|
|
assert queue.drop_counts == {"cancelled": 1}
|
|
finally:
|
|
proceed.set()
|
|
assert stage.close()
|
|
|
|
|
|
def test_stop_discards_completed_result_but_cannot_free_cpu_owned_input():
|
|
queue = StreamMailbox()
|
|
stage = SerialGpuStage(queue, lambda item: item["sequence"], threading.Event())
|
|
try:
|
|
queue.put(bundle(0))
|
|
first, _ = stage.take()
|
|
queue.put(bundle(1))
|
|
with stage.output_condition:
|
|
assert stage.output_condition.wait_for(lambda: stage.completed is not None, timeout=1)
|
|
assert queue.bytes == 20
|
|
assert stage.close()
|
|
assert queue.bytes == 10 and not queue.quiescent
|
|
queue.release(first)
|
|
assert queue.quiescent
|
|
finally:
|
|
assert stage.close()
|