Skip to content

feat(taskworker): Count child startup as busy in occupancy #844

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
evanh wants to merge 1 commit into from
Closed

feat(taskworker): Count child startup as busy in occupancy #844

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 25 additions & 5 deletions clients/python/src/taskbroker_client/worker/childtiming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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."""
Expand All @@ -179,14 +193,20 @@ 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
# busy and eligible grow together instead of one outrunning the other.
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)

Expand Down
23 changes: 12 additions & 11 deletions clients/python/src/taskbroker_client/worker/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
}

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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(
Expand All @@ -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)
Expand Down Expand Up @@ -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())
Expand Down Expand Up @@ -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:
Expand Down
36 changes: 27 additions & 9 deletions clients/python/tests/worker/test_worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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()
Expand All @@ -1442,16 +1446,15 @@ 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):
pool._emit_periodic_metrics()

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:
Expand Down Expand Up @@ -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()
Expand All @@ -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
)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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:
Expand Down
Loading