From 9eff180b072728478fe779fb1ffc65e0846cd8c1 Mon Sep 17 00:00:00 2001 From: Evan Hicks Date: Wed, 4 Nov 2026 14:24:20 -0500 Subject: [PATCH] feat(taskworker): Count child startup as busy in occupancy A warming child can't take work, but occupancy left it out of both busy time and the ceiling, so respawns shrank capacity without moving the metric. The parent now bills spawn-to-running as busy. Co-Authored-By: Claude Opus 5.5 --- .../taskbroker_client/worker/childtiming.py | 30 +++++++++++++--- .../src/taskbroker_client/worker/worker.py | 23 ++++++------ clients/python/tests/worker/test_worker.py | 36 ++++++++++++++----- 3 files changed, 64 insertions(+), 25 deletions(-) diff --git a/clients/python/src/taskbroker_client/worker/childtiming.py b/clients/python/src/taskbroker_client/worker/childtiming.py index 772906e9..8f5ed320 100644 --- a/clients/python/src/taskbroker_client/worker/childtiming.py +++ b/clients/python/src/taskbroker_client/worker/childtiming.py @@ -150,13 +150,22 @@ class ChildTimeAccounting: _prev_wait: float = 0.0 _accounted: bool = False _measured_from: float = 0.0 + _starting: bool = False + _startup_busy: float = 0.0 + + def mark_spawned(self, now: float) -> None: + """Count a warming child as busy: its slot cannot take work yet.""" + self._accounted = True + self._starting = True + self._measured_from = now def mark_running(self, now: float) -> None: - """Start counting this child, baselining against the slot as it stands. + """Start reading this child's slot, baselining against it as it stands. Baselining rather than zeroing discards whatever the child banked before - the parent saw its `running` message, which is the same window over - which it is absent from the pool's running count. + the parent saw its `running` message. A spawned child already billed + that window as startup busy; carry the unsampled tail into the next + sample and leave `_measured_from` alone so eligible covers it too. """ reading = self._read(now) if reading is None: @@ -167,8 +176,13 @@ def mark_running(self, now: float) -> None: else: self._prev_busy, self._prev_wait = reading + if self._starting: + self._startup_busy = max(0.0, now - self._measured_from) + self._starting = False + else: + self._startup_busy = 0.0 + self._measured_from = now self._accounted = True - self._measured_from = now def mark_stopped(self) -> None: """Stop counting, so an exiting child's tail misses the live pool.""" @@ -179,6 +193,11 @@ def sample(self, now: float) -> SampleResult: if not self._accounted: return SampleResult() + if self._starting: + elapsed = max(0.0, now - self._measured_from) + self._measured_from = now + return SampleResult(elapsed, 0.0, elapsed) + reading = self._read(now) if reading is None: # Advance nothing. The next sample then covers both windows, and its @@ -186,7 +205,8 @@ def sample(self, now: float) -> SampleResult: return SampleResult() busy_now, wait_now = reading - busy = max(0.0, busy_now - self._prev_busy) + busy = max(0.0, busy_now - self._prev_busy) + self._startup_busy + self._startup_busy = 0.0 wait = max(0.0, wait_now - self._prev_wait) eligible = max(0.0, now - self._measured_from) diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py index cdd71f71..32815c3d 100644 --- a/clients/python/src/taskbroker_client/worker/worker.py +++ b/clients/python/src/taskbroker_client/worker/worker.py @@ -970,13 +970,13 @@ def ready_count(self) -> int: return sum(1 for c in self._children.values() if c.state == "running") def _accounting_log( - self, busy_time: float, wait_time: float, ceiling: float, running_count: int + self, busy_time: float, wait_time: float, ceiling: float, measured_count: int ) -> dict[str, float | int | str]: return { "busy_time": busy_time, "wait_time": wait_time, "ceiling": ceiling, - "running_count": running_count, + "measured_count": measured_count, "processing_pool": self._processing_pool_name, } @@ -1034,15 +1034,15 @@ def _emit_periodic_metrics(self) -> None: busy_time = 0.0 wait_time = 0.0 eligible_time = 0.0 - running_count = 0 + measured_count = 0 for child in self._children.values(): state_counts[child.state] += 1 - # Running only: the numerator must match occupancy's divisor. - if child.state != "running": + # Pending children count too: their startup is billed as busy. + if child.state == "exiting": continue - running_count += 1 + measured_count += 1 result = child.timing.sample(time.monotonic()) busy_time += result.busy wait_time += result.wait @@ -1070,9 +1070,9 @@ def _emit_periodic_metrics(self) -> None: max(0.0, wait_time) ) - if running_count > 0 and eligible_time > 0: + if measured_count > 0 and eligible_time > 0: # Sum of the windows each child was measured over. `elapsed * - # running_count` would instead bill a child baselined part-way + # measured_count` would instead bill a child baselined part-way # through this flush as if it had been here all along, which reads # as idle time the pool never had. ceiling = eligible_time @@ -1086,7 +1086,7 @@ def _emit_periodic_metrics(self) -> None: ) logger.warning( "taskworker.worker.occupancy.accounting_overflow", - extra=self._accounting_log(busy_time, wait_time, ceiling, running_count), + extra=self._accounting_log(busy_time, wait_time, ceiling, measured_count), ) elif accounted < ceiling * 0.9: self._metrics.incr( @@ -1095,7 +1095,7 @@ def _emit_periodic_metrics(self) -> None: ) logger.warning( "taskworker.worker.occupancy.accounting_deficit", - extra=self._accounting_log(busy_time, wait_time, ceiling, running_count), + extra=self._accounting_log(busy_time, wait_time, ceiling, measured_count), ) occupancy = min(busy_time / ceiling, 1.0) @@ -1302,7 +1302,7 @@ def spawn_children_thread() -> None: continue - # Baseline here, where it also enters `running_count`. + # Hand off from startup busy to reading the slot. if message.event == "running": child.state = "running" child.timing.mark_running(time.monotonic()) @@ -1379,6 +1379,7 @@ def spawn_children_thread() -> None: release=release, timing=ChildTimeAccounting(shm=timing_shm), ) + child.timing.mark_spawned(time.monotonic()) self._children[child_id] = child except Exception as e: diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py index 56079841..cefd8a27 100644 --- a/clients/python/tests/worker/test_worker.py +++ b/clients/python/tests/worker/test_worker.py @@ -1340,7 +1340,10 @@ def _make_tracked_child( # Baseline against the zeroed slot, so the seeding below lands in sample 1. # Defaults to the start of the [10.0, 11.0] interval the occupancy tests # use, so a child is measurable throughout unless told otherwise. - timing.mark_running(measured_from) + if state == "pending": + timing.mark_spawned(measured_from) + else: + timing.mark_running(measured_from) if busy_since is not None: kind, start = KIND_BUSY, busy_since @@ -1381,7 +1384,8 @@ def _gauge_calls(metrics: mock.Mock, name: str) -> list[Any]: def test_emit_periodic_metrics_skips_occupancy_during_warmup() -> None: - pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4) + # A fresh pod has no task yet, so the first-task gate holds the gauge back. + pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4, received_task=False) pool._metrics = mock.Mock() # A freshly started pod has only pending children; none are consuming yet. @@ -1433,7 +1437,7 @@ def test_emit_periodic_metrics_time_weights_busy_over_the_interval() -> None: assert child.timing.sample(12.0).busy == pytest.approx(0.0) -def test_emit_periodic_metrics_divides_by_running_children() -> None: +def test_emit_periodic_metrics_bills_warming_children_as_busy() -> None: # Two children busy for the whole 1s interval, one idle, one still warming. pool = _make_result_thread_pool(_SendResultCapture(), concurrency=8) pool._metrics = mock.Mock() @@ -1442,7 +1446,7 @@ def test_emit_periodic_metrics_divides_by_running_children() -> None: pool._children[uuid4()] = _make_tracked_child("running", busy_accumulated=1.0) pool._children[uuid4()] = _make_tracked_child("running", busy_accumulated=1.0) pool._children[uuid4()] = _make_tracked_child("running", busy_accumulated=0.0) - # Excluded from both numerator and denominator. + # Cannot take work, so busy in both numerator and denominator. pool._children[uuid4()] = _make_tracked_child("pending") with mock.patch("taskbroker_client.worker.worker.time.monotonic", return_value=11.0): @@ -1450,8 +1454,7 @@ def test_emit_periodic_metrics_divides_by_running_children() -> None: occupancy_calls = _gauge_calls(pool._metrics, "taskworker.worker.occupancy") assert len(occupancy_calls) == 1 - # 2 busy-child-seconds over 3 running slots for a 1s interval. - assert occupancy_calls[0].args[1] == pytest.approx(2 / 3) + assert occupancy_calls[0].args[1] == pytest.approx(3 / 4) def test_emit_periodic_metrics_clamps_occupancy_and_flags_the_overflow() -> None: @@ -1649,7 +1652,7 @@ def test_sample_keeps_counting_wait_when_the_fold_overshoots_a_close() -> None: assert second.eligible == pytest.approx(1.0) -def test_emit_periodic_metrics_counters_exclude_non_running_children() -> None: +def test_emit_periodic_metrics_counters_exclude_exiting_children() -> None: # The counters must sum over the same population occupancy divides by. pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4) pool._metrics = mock.Mock() @@ -1663,7 +1666,7 @@ def test_emit_periodic_metrics_counters_exclude_non_running_children() -> None: pool._emit_periodic_metrics() busy = _distribution_calls(pool._metrics, "taskworker.worker.child_busy_seconds") - assert busy[0].args[1] == pytest.approx(1.0) + assert busy[0].args[1] == pytest.approx(2.0) assert _gauge_calls(pool._metrics, "taskworker.worker.occupancy")[0].args[1] == pytest.approx( 1.0 ) @@ -1867,6 +1870,21 @@ def test_child_timing_excludes_time_banked_before_the_parent_saw_running() -> No assert _bw(reader.sample(6.0)) == pytest.approx((0.0, 1.0)) +def test_child_timing_bills_startup_as_busy_across_the_running_handoff() -> None: + writer, reader = _writer_and_reader() + reader.mark_spawned(0.0) + + first = reader.sample(1.0) + assert (first.busy, first.wait, first.eligible) == pytest.approx((1.0, 0.0, 1.0)) + + writer.mark_running(1.2) # child's own wait before the parent saw it is dropped + reader.mark_running(1.5) + second = reader.sample(2.0) + assert (second.busy, second.wait, second.eligible) == pytest.approx((0.5, 0.5, 1.0)) + + assert _bw(reader.sample(3.0)) == pytest.approx((0.0, 1.0)) + + def test_emit_periodic_metrics_emits_busy_and_wait_seconds() -> None: # Interval [10.0, 11.0]: one child busy throughout, one waiting 0.25s. pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4) @@ -1895,7 +1913,7 @@ def test_emit_periodic_metrics_emits_busy_and_wait_seconds() -> None: def test_emit_periodic_metrics_emits_counters_during_warmup() -> None: # Unlike occupancy: zero separates "idle" from "not reporting". - pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4) + pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4, received_task=False) pool._metrics = mock.Mock() with pool._children_lock: