tests/spa/orchestration/test_orchestration_cpu_offload.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """The CPU offload judge: one hot worker slims by one user per beat.16 17 The stage is the one ``test_orchestration_group_handler`` builds — real child18 processes under a real group and a real vertex — and both the CPU and the19 per-user recent work are DECLARED, written straight into the photo the judge20 reads. Implementation tests: they photograph the experimental policy and go21 with it. A CPU-closed worker past ``cpu_offload_percent`` cedes its least22 busy active user through the ordered freeze; a single active user is never23 transferred; the standing conditions reach the journal once, not every beat.24 """25 26 from __future__ import annotations27 28 import json29 import time30 31 import pytest32 33 from genro_asgi_multiworker_spa.orchestration.group_policy import GroupPolicy, GroupPolicyError34 35 from .test_orchestration_group_handler import known_at_the_vertex36 from .test_orchestration_group_handler import commander # noqa: F40137 from .test_orchestration_group_handler import group_settings # noqa: F40138 from .test_orchestration_group_handler import instance_root # noqa: F40139 from .test_orchestration_group_handler import make_group # noqa: F40140 41 DECISIONS_LOGGER = "genro_asgi.orchestration.decisions"42 43 44 def declare_cpu(worker_handler, cpu_temperature_percent: float) -> None:45 """Declare the CPU channel directly; policy tests do not test its clock."""46 worker_handler.cpu_temperature_percent = cpu_temperature_percent47 worker_handler.cpu_temperature_sampled_at = time.monotonic()48 worker_handler.cpu_temperature_interval_seconds = 0.149 worker_handler.get_cpu_temperature_percent = lambda: cpu_temperature_percent50 51 52 async def offload_group(make_group, commander, users, **policies):53 """One worker past the offload threshold, its users placed and declared.54 55 The scripted child carries the users as RESIDENTS (no transfer flag: a56 flag is read at the vertex as a hold), and every identity is known at the57 vertex before the child registers.58 """59 group = make_group(60 users=list(users),61 transfer_flag=None,62 cpu_admission_close_percent=50.0,63 cpu_offload_percent=75.0,64 **policies,65 )66 for user in users:67 known_at_the_vertex(commander, f"c_{user}", user)68 worker_handler = await group.start_worker()69 for user in users:70 assert await group.assign_user(user) == worker_handler.name71 worker_handler.hosted_users.add(user)72 worker_handler.cpu_admission_open = False73 declare_cpu(worker_handler, 80.0)74 return group, worker_handler75 76 77 def declare_service(worker_handler, user, *, seconds=0.0, calls=0, pending=0):78 """Write one user's recent work into the photo the judge reads."""79 item = worker_handler.worker_snapshot["users"][user]["item"]80 item["recent_service_seconds"] = seconds81 item["recent_call_count"] = calls82 item["pending_call_count"] = pending83 84 85 def offload_decisions(caplog):86 return [87 json.loads(record.getMessage())88 for record in caplog.records89 if record.name == DECISIONS_LOGGER90 and json.loads(record.getMessage())["decision"] == "cpu_offload"91 ]92 93 94 # --- who is ceded, and when nobody is ---------------------------------------95 96 97 async def test_below_the_threshold_nobody_is_ceded(make_group, commander):98 group, worker_handler = await offload_group(make_group, commander, ["mario", "lucia"])99 declare_cpu(worker_handler, 60.0)100 declare_service(worker_handler, "mario", seconds=5.0, calls=3)101 declare_service(worker_handler, "lucia", seconds=3.0, calls=1)102 103 await group.check_cpu_offload()104 105 assert group.user_worker_map["mario"] == worker_handler.name106 assert group.user_worker_map["lucia"] == worker_handler.name107 108 109 async def test_with_the_policy_off_the_judge_is_inert(make_group, commander, caplog):110 group, worker_handler = await offload_group(make_group, commander, ["mario", "lucia"])111 declare_service(worker_handler, "mario", seconds=5.0, calls=3)112 declare_service(worker_handler, "lucia", seconds=3.0, calls=1)113 group.apply_policy(114 GroupPolicy.from_settings(group.policy.to_settings() | {"cpu_offload_percent": None}),115 [],116 )117 118 with caplog.at_level("INFO", logger=DECISIONS_LOGGER):119 await group.check_cpu_offload()120 121 assert group.user_worker_map["mario"] == worker_handler.name122 assert group.user_worker_map["lucia"] == worker_handler.name123 assert offload_decisions(caplog) == []124 125 126 async def test_one_beat_cedes_the_least_busy_material_alone(make_group, commander, caplog):127 group, worker_handler = await offload_group(128 make_group, commander, ["mario", "lucia", "pia"]129 )130 declare_service(worker_handler, "mario", seconds=5.0, calls=9)131 declare_service(worker_handler, "lucia", seconds=1.5, calls=2)132 declare_service(worker_handler, "pia", seconds=2.0, calls=4)133 134 with caplog.at_level("INFO", logger=DECISIONS_LOGGER):135 await group.check_cpu_offload()136 137 # S=8.5, N=3, threshold ~1.417: all three material, lucia the least busy.138 assert commander.user_is_frozen("lucia") is True139 assert group.user_worker_map["lucia"] is None140 assert group.user_worker_map["mario"] == worker_handler.name141 assert group.user_worker_map["pia"] == worker_handler.name142 rows = offload_decisions(caplog)143 reasons = [row["reason"] for row in rows]144 assert reasons == ["cpu_offload_threshold", "cpu_offload_user_selected", "cpu_offload_completed"]145 # Every row carries the numbers that rebuild the judgment.146 threshold_row = rows[0]147 assert threshold_row["numbers"]["cpu_temperature_percent"] == 80.0148 assert threshold_row["numbers"]["window_service_seconds"] == pytest.approx(8.5)149 assert threshold_row["numbers"]["active_users"] == 3150 assert threshold_row["numbers"]["material_threshold"] == pytest.approx(8.5 / 6)151 assert threshold_row["numbers"]["material_contributors"] == 3152 assert threshold_row["numbers"]["cedible_contributors"] == 3153 selected_row = rows[1]154 assert selected_row["numbers"]["recent_service_seconds"] == 1.5155 assert selected_row["numbers"]["recent_call_count"] == 2156 assert selected_row["numbers"]["pending_call_count"] == 0157 158 159 async def test_a_call_in_flight_keeps_a_user_from_the_head(make_group, commander):160 """recent seconds 0 with a pending call is a call just begun, not idleness."""161 group, worker_handler = await offload_group(162 make_group, commander, ["mario", "lucia", "pia"]163 )164 declare_service(worker_handler, "mario", seconds=10.0, calls=9)165 declare_service(worker_handler, "lucia", seconds=0.0, calls=0, pending=1)166 declare_service(worker_handler, "pia", seconds=3.0, calls=4)167 168 await group.check_cpu_offload()169 170 assert commander.user_is_frozen("pia") is True171 assert group.user_worker_map["lucia"] == worker_handler.name172 173 174 async def test_all_material_busy_defers_and_nobody_is_frozen(make_group, commander, caplog):175 group, worker_handler = await offload_group(make_group, commander, ["mario", "lucia"])176 declare_service(worker_handler, "mario", seconds=6.0, calls=5, pending=1)177 declare_service(worker_handler, "lucia", seconds=4.0, calls=3, pending=1)178 179 with caplog.at_level("INFO", logger=DECISIONS_LOGGER):180 await group.check_cpu_offload()181 await group.check_cpu_offload()182 183 # Both material, both mid-call: no freeze at all, one deferred row (dedup).184 assert commander.user_is_frozen("mario") is False185 assert commander.user_is_frozen("lucia") is False186 assert group.user_worker_map["lucia"] == worker_handler.name187 rows = offload_decisions(caplog)188 assert [row["reason"] for row in rows] == ["cpu_offload_deferred_pending_calls"]189 assert rows[0]["numbers"]["material_contributors"] == 2190 assert rows[0]["numbers"]["cedible_contributors"] == 0191 192 193 async def test_the_next_beat_cedes_the_contributor_that_came_free(make_group, commander):194 group, worker_handler = await offload_group(make_group, commander, ["mario", "lucia"])195 declare_service(worker_handler, "mario", seconds=6.0, calls=5, pending=1)196 declare_service(worker_handler, "lucia", seconds=4.0, calls=3, pending=1)197 await group.check_cpu_offload()198 assert commander.user_is_frozen("lucia") is False199 200 declare_service(worker_handler, "lucia", seconds=4.0, calls=3, pending=0)201 await group.check_cpu_offload()202 203 assert commander.user_is_frozen("lucia") is True204 assert group.user_worker_map["mario"] == worker_handler.name205 206 207 async def test_the_inactive_user_is_no_candidate(make_group, commander, caplog):208 """Nothing recent and nothing in flight belongs to the idle freeze, not here."""209 group, worker_handler = await offload_group(make_group, commander, ["mario", "lucia"])210 declare_service(worker_handler, "mario", seconds=5.0, calls=3)211 declare_service(worker_handler, "lucia")212 213 with caplog.at_level("INFO", logger=DECISIONS_LOGGER):214 await group.check_cpu_offload()215 216 assert commander.user_is_frozen("lucia") is False217 assert commander.user_is_frozen("mario") is False218 assert [row["reason"] for row in offload_decisions(caplog)] == ["single_user_overload"]219 220 221 async def test_negligible_activity_never_makes_a_candidate(make_group, commander, caplog):222 """Noise-level users are alive but immaterial: the dominant one stays alone."""223 group, worker_handler = await offload_group(224 make_group, commander, ["mario", "lucia", "pia"]225 )226 declare_service(worker_handler, "mario", seconds=10.0, calls=9)227 declare_service(worker_handler, "lucia", seconds=0.05, calls=1)228 declare_service(worker_handler, "pia", seconds=0.05, calls=1)229 230 with caplog.at_level("INFO", logger=DECISIONS_LOGGER):231 await group.check_cpu_offload()232 233 # S=10.1, N=3, threshold ~1.68: only mario is material.234 assert commander.user_is_frozen("lucia") is False235 assert commander.user_is_frozen("pia") is False236 rows = offload_decisions(caplog)237 assert [row["reason"] for row in rows] == ["single_user_overload"]238 assert rows[0]["subject"] == "mario"239 assert rows[0]["numbers"]["active_users"] == 3240 assert rows[0]["numbers"]["material_contributors"] == 1241 242 243 async def test_exact_equality_with_the_threshold_is_material(make_group, commander):244 """s == S/(2N) is material: a strict comparison would leave mario alone."""245 group, worker_handler = await offload_group(make_group, commander, ["mario", "lucia"])246 declare_service(worker_handler, "mario", seconds=3.0, calls=3)247 declare_service(worker_handler, "lucia", seconds=1.0, calls=1)248 249 await group.check_cpu_offload()250 251 # S=4, N=2, threshold exactly 1.0: lucia is material and the least busy.252 assert commander.user_is_frozen("lucia") is True253 254 255 # --- the black sheep and the journal's silence -------------------------------256 257 258 async def test_a_single_active_user_is_said_once_and_never_moved(259 make_group, commander, caplog260 ):261 group, worker_handler = await offload_group(make_group, commander, ["mario"])262 declare_service(worker_handler, "mario", seconds=9.0, calls=9)263 264 with caplog.at_level("INFO", logger=DECISIONS_LOGGER):265 await group.check_cpu_offload()266 await group.check_cpu_offload()267 await group.check_cpu_offload()268 269 assert group.user_worker_map["mario"] == worker_handler.name270 rows = offload_decisions(caplog)271 assert [row["reason"] for row in rows] == ["single_user_overload"]272 assert rows[0]["subject"] == "mario"273 274 275 async def test_the_condition_speaks_again_after_it_fell(make_group, commander, caplog):276 group, worker_handler = await offload_group(make_group, commander, ["mario"])277 declare_service(worker_handler, "mario", seconds=9.0, calls=9)278 279 with caplog.at_level("INFO", logger=DECISIONS_LOGGER):280 await group.check_cpu_offload()281 declare_cpu(worker_handler, 40.0)282 await group.check_cpu_offload()283 declare_cpu(worker_handler, 80.0)284 await group.check_cpu_offload()285 286 assert [row["reason"] for row in offload_decisions(caplog)] == [287 "single_user_overload",288 "single_user_overload",289 ]290 291 292 async def test_nobody_active_is_said_once(make_group, commander, caplog):293 group, worker_handler = await offload_group(make_group, commander, ["mario", "lucia"])294 declare_service(worker_handler, "mario")295 declare_service(worker_handler, "lucia")296 297 with caplog.at_level("INFO", logger=DECISIONS_LOGGER):298 await group.check_cpu_offload()299 await group.check_cpu_offload()300 301 assert [row["reason"] for row in offload_decisions(caplog)] == [302 "cpu_offload_no_active_candidate"303 ]304 305 306 # --- the departure's roads ----------------------------------------------------307 308 309 async def test_a_refused_freeze_releases_the_hold_and_says_so(310 make_group, commander, caplog311 ):312 group, worker_handler = await offload_group(313 make_group, commander, ["mario", "lucia"], freeze_refused=True314 )315 declare_service(worker_handler, "mario", seconds=5.0, calls=3)316 declare_service(worker_handler, "lucia", seconds=3.0, calls=1)317 318 with caplog.at_level("INFO", logger=DECISIONS_LOGGER):319 await group.check_cpu_offload()320 321 assert commander.user_is_frozen("lucia") is False322 assert commander.user_map["lucia"]["on_hold"] is None323 assert group.user_worker_map["lucia"] == worker_handler.name324 assert offload_decisions(caplog)[-1]["reason"] == "cpu_offload_refused"325 326 327 async def test_the_offloaded_user_cannot_return_to_the_closed_worker(328 make_group, commander329 ):330 group, worker_handler = await offload_group(make_group, commander, ["mario", "lucia"])331 declare_service(worker_handler, "mario", seconds=5.0, calls=3)332 declare_service(worker_handler, "lucia", seconds=3.0, calls=1)333 await group.check_cpu_offload()334 assert group.user_worker_map["lucia"] is None335 336 placed = await group.assign_user("lucia")337 338 assert placed != worker_handler.name339 assert len(group.living_workers) == 2340 341 342 async def test_the_hottest_of_the_closed_workers_is_the_target(make_group, commander):343 group, first = await offload_group(make_group, commander, ["mario", "lucia"])344 declare_service(first, "mario", seconds=5.0, calls=3)345 declare_service(first, "lucia", seconds=3.0, calls=1)346 second = await group.start_worker()347 for user in ("carla", "nino"):348 known_at_the_vertex(commander, f"c_{user}", user)349 group.user_worker_map[user] = second.name350 second.hosted_users.add(user)351 second.cpu_admission_open = False352 declare_cpu(second, 90.0)353 # The second child's photo carries no users of its own: hand it the rows.354 second.worker_snapshot["users"] = {355 user: {"transfer_flag": None, "item": {"state": "active"}} for user in ("carla", "nino")356 }357 declare_service(second, "carla", seconds=4.0, calls=2)358 declare_service(second, "nino", seconds=2.0, calls=1)359 360 await group.check_cpu_offload()361 362 # The 90% worker cedes nino; the 80% one keeps everybody this beat.363 assert group.user_worker_map["nino"] is None364 assert group.user_worker_map["lucia"] == first.name365 366 367 async def test_a_cession_stamps_the_cpu_pressure_clock(make_group, commander):368 """Coherence of the pressure history: the retirement is already suspended369 while a worker is CPU-closed; the stamp keeps the quiet after it true."""370 group, worker_handler = await offload_group(make_group, commander, ["mario", "lucia"])371 declare_service(worker_handler, "mario", seconds=5.0, calls=3)372 declare_service(worker_handler, "lucia", seconds=3.0, calls=1)373 group._cpu_pressure_monotonic = None374 375 await group.check_cpu_offload()376 377 assert group._cpu_pressure_monotonic is not None378 379 380 # --- the policy road ------------------------------------------------------------381 382 383 async def test_the_offload_threshold_applies_live(make_group, commander):384 group, worker_handler = await offload_group(make_group, commander, ["mario", "lucia"])385 declare_service(worker_handler, "mario", seconds=5.0, calls=3)386 declare_service(worker_handler, "lucia", seconds=3.0, calls=1)387 group.apply_policy(388 GroupPolicy.from_settings(group.policy.to_settings() | {"cpu_offload_percent": 90.0}),389 [],390 )391 392 await group.check_cpu_offload()393 assert commander.user_is_frozen("lucia") is False394 395 declare_cpu(worker_handler, 95.0)396 await group.check_cpu_offload()397 assert commander.user_is_frozen("lucia") is True398 assert worker_handler is group.worker_handler_map[worker_handler.name]399 400 401 def test_a_group_built_with_offload_but_no_admission_does_not_exist(make_group):402 with pytest.raises(GroupPolicyError) as caught:403 make_group(cpu_offload_percent=75.0)404 assert "requires cpu_admission_close_percent" in str(caught.value)