File size: 3,662 Bytes
a2e086b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """The bounded analysis pool: cap, ordering, position and ETA.
The pool is what turns a launch-day burst from "everything fell over" into
"you're 4th in line, ~10 min". These lock its behaviour.
"""
import os
import sys
import threading
import time
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from backend.analysis_queue import AnalysisPool
def _blocking_job(started, release, done):
def run():
started.set()
release.wait(timeout=5)
done.set()
return run
def test_pool_caps_concurrency_at_worker_count():
pool = AnalysisPool(workers=2)
peak = {"n": 0}
lock = threading.Lock()
live = {"n": 0}
release = threading.Event()
def job():
with lock:
live["n"] += 1
peak["n"] = max(peak["n"], live["n"])
release.wait(timeout=5)
with lock:
live["n"] -= 1
for i in range(6):
pool.submit(f"job-{i}", job)
time.sleep(0.4) # let the workers pick up
assert peak["n"] <= 2, "never more than `workers` analyses run at once"
assert peak["n"] == 2, "both workers should be busy under load"
release.set()
def test_waiting_job_reports_its_place_in_line():
pool = AnalysisPool(workers=1)
release = threading.Event()
first_started = threading.Event()
def first():
first_started.set()
release.wait(timeout=5)
def noop():
release.wait(timeout=5)
pool.submit("first", first)
first_started.wait(timeout=5) # occupies the single worker
pool.submit("second", noop)
pool.submit("third", noop)
time.sleep(0.2)
assert pool.position("first") == 0 # running
assert pool.position("second") == 1 # next
assert pool.position("third") == 2 # after that
assert pool.position("unknown") is None
release.set()
def test_finished_job_is_no_longer_in_the_pool():
pool = AnalysisPool(workers=1)
done = threading.Event()
pool.submit("solo", lambda: done.set())
assert done.wait(timeout=5)
time.sleep(0.1)
assert pool.position("solo") is None
def test_eta_grows_with_queue_depth():
pool = AnalysisPool(workers=2)
release = threading.Event()
started = threading.Event()
def hold():
started.set()
release.wait(timeout=5)
for i in range(6):
pool.submit(f"j{i}", hold)
started.wait(timeout=5)
time.sleep(0.2)
etas = [pool.eta_seconds(f"j{i}") for i in range(6)]
etas = [e for e in etas if e is not None]
# A job deeper in the queue must never show a shorter wait than one ahead.
assert etas == sorted(etas), f"ETA must be monotonic by depth, got {etas}"
release.set()
def test_snapshot_counts_running_and_queued():
pool = AnalysisPool(workers=2)
release = threading.Event()
started = threading.Semaphore(0)
def hold():
started.release()
release.wait(timeout=5)
for i in range(5):
pool.submit(f"s{i}", hold)
started.acquire(timeout=5)
started.acquire(timeout=5) # two are running
time.sleep(0.2)
snap = pool.snapshot()
assert snap["workers"] == 2
assert snap["running"] == 2
assert snap["queued"] == 3
release.set()
def test_a_crashing_job_frees_its_slot():
pool = AnalysisPool(workers=1)
ran = threading.Event()
def boom():
raise RuntimeError("analysis blew up")
def after():
ran.set()
pool.submit("boom", boom)
pool.submit("after", after)
# If the crash leaked the worker, `after` would never run.
assert ran.wait(timeout=5), "a crashed job must not wedge the worker"
|