aaronrene / scooling-lab public
test_t3_stress.py python
145 lines 5.0 KB
Raw
sha256:fc4c9ad652d1fff3dc508cb6ea02ee710ee6dfc4cb3761291d9900b5e029ea8a feat(slice-7): T3 dataset review lifecycle, job queue, prov… Human minor ⚠ breaking 41 days ago
1 """Stress tier tests — T3 concurrent submissions, queue concurrency bound."""
2
3 from __future__ import annotations
4
5 import concurrent.futures
6 import threading
7 import unittest
8
9 from scooling_lab_helpers import valid_payload
10
11 from scooling_lab.dataset_review import DatasetStore
12 from scooling_lab.errors import ApiError, ErrorCode
13 from scooling_lab.service import TrainingApiService
14 from scooling_lab.store import TrainingJobStore
15
16
17 def _make_service(
18 *,
19 queue_limit: int = 256,
20 max_concurrent_running: int = 1,
21 auto_run: bool = True,
22 dataset_store: DatasetStore | None = None,
23 ) -> TrainingApiService:
24 store = TrainingJobStore(
25 queue_limit=queue_limit,
26 max_concurrent_running=max_concurrent_running,
27 )
28 return TrainingApiService(
29 store, auto_run_worker=auto_run, dataset_store=dataset_store
30 )
31
32
33 class T3StressConcurrencyBoundTests(unittest.TestCase):
34 """Stress tests verifying the running-job concurrency bound holds under load."""
35
36 def test_stress_t3_concurrent_submissions_never_exceed_running_bound(self) -> None:
37 """At most max_concurrent_running jobs are in running state at once.
38
39 We track the peak concurrent running count by hooking into the store
40 check; since the fake worker is synchronous the semaphore ensures the
41 bound is never exceeded.
42 """
43
44 max_running = 2
45 service = _make_service(
46 max_concurrent_running=max_running, queue_limit=64, auto_run=True
47 )
48 observed_running: list[int] = []
49 lock = threading.Lock()
50 original_run_job = service._worker.run_job
51
52 def instrumented_run(job_id: str) -> object: # type: ignore[misc]
53 snapshot = service._store.running_count()
54 with lock:
55 observed_running.append(snapshot)
56 return original_run_job(job_id)
57
58 service._worker.run_job = instrumented_run # type: ignore[method-assign]
59
60 with concurrent.futures.ThreadPoolExecutor(max_workers=12) as pool:
61 list(
62 pool.map(
63 lambda i: service.create_training_job(valid_payload(f"conc-{i}")),
64 range(24),
65 )
66 )
67
68 self.assertTrue(len(observed_running) > 0)
69 peak = max(observed_running)
70 self.assertLessEqual(
71 peak,
72 max_running,
73 msg=f"Peak concurrent running {peak} exceeded bound {max_running}",
74 )
75
76 def test_stress_t3_concurrent_dataset_registrations_are_safe(self) -> None:
77 """Concurrent registrations for distinct datasets are all stored correctly."""
78
79 ds_store = DatasetStore()
80 dataset_ids = [f"stress-ds-{i:04d}" for i in range(50)]
81
82 with concurrent.futures.ThreadPoolExecutor(max_workers=10) as pool:
83 results = list(
84 pool.map(lambda did: ds_store.register(did), dataset_ids)
85 )
86
87 self.assertEqual(len(results), 50)
88 for did in dataset_ids:
89 self.assertEqual(ds_store.get(did).dataset_id, did)
90
91 def test_stress_t3_concurrent_review_decisions_are_stable(self) -> None:
92 """Only the first review decision for a dataset wins; retries are idempotent."""
93
94 ds_store = DatasetStore()
95 ds_store.register("concurrent-review-ds")
96 ds_store.submit_for_review("concurrent-review-ds")
97
98 errors: list[ApiError] = []
99 successes: list[str] = []
100 lock = threading.Lock()
101
102 def try_approve() -> None:
103 try:
104 record = ds_store.approve("concurrent-review-ds")
105 with lock:
106 successes.append(record.status.value)
107 except ApiError as exc:
108 with lock:
109 errors.append(exc)
110
111 with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
112 list(pool.map(lambda _: try_approve(), range(16)))
113
114 final = ds_store.get("concurrent-review-ds")
115 self.assertEqual(final.status.value, "approved")
116
117 def test_stress_t3_queue_limit_still_enforced_under_concurrent_load(self) -> None:
118 """QUEUE_LIMIT_EXCEEDED is returned reliably under concurrent pressure."""
119
120 service = _make_service(
121 queue_limit=3, max_concurrent_running=3, auto_run=False
122 )
123 service.create_training_job(valid_payload("ql-a"))
124 service.create_training_job(valid_payload("ql-b"))
125 service.create_training_job(valid_payload("ql-c"))
126
127 limit_hits: list[bool] = []
128 lock = threading.Lock()
129
130 def try_create(suffix: str) -> None:
131 try:
132 service.create_training_job(valid_payload(f"ql-overflow-{suffix}"))
133 except ApiError as exc:
134 if exc.code == ErrorCode.QUEUE_LIMIT_EXCEEDED:
135 with lock:
136 limit_hits.append(True)
137
138 with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
139 list(pool.map(try_create, [str(i) for i in range(16)]))
140
141 self.assertGreater(len(limit_hits), 0)
142
143
144 if __name__ == "__main__":
145 unittest.main()
File History 1 commit
sha256:fc4c9ad652d1fff3dc508cb6ea02ee710ee6dfc4cb3761291d9900b5e029ea8a feat(slice-7): T3 dataset review lifecycle, job queue, prov… Human minor ⚠ 41 days ago