diff --git a/clients/python/src/taskbroker_client/worker/childtiming.py b/clients/python/src/taskbroker_client/worker/childtiming.py new file mode 100644 index 00000000..1bf26446 --- /dev/null +++ b/clients/python/src/taskbroker_client/worker/childtiming.py @@ -0,0 +1,227 @@ +"""Shared-memory busy/wait accounting for worker children. + +Children write their own cumulative totals into a slot; the parent diffs them at +flush. Costs O(children) per second rather than O(tasks) per second, and there is +no queue to fall behind. + +Slot layout, five doubles per child:: + + 0 version seqlock; odd means a write is in progress + 1 busy_total cumulative seconds closed into busy + 2 wait_total cumulative seconds closed into wait + 3 segment_start time.monotonic() when the open segment began + 4 segment_kind KIND_NONE, KIND_WAIT or KIND_BUSY + +Values are absolute and cumulative, so a torn read costs one transient sample +that the next flush re-derives. time.monotonic() is CLOCK_MONOTONIC, which is +system-wide, so a child's timestamps are valid in the parent. +""" + +from __future__ import annotations + +import ctypes +from dataclasses import dataclass + +SLOT_VERSION = 0 +SLOT_BUSY_TOTAL = 1 +SLOT_WAIT_TOTAL = 2 +SLOT_SEGMENT_START = 3 +SLOT_SEGMENT_KIND = 4 +SLOT_WIDTH = 5 + +# NONE must be 0.0 so a zeroed slot reads as "nothing open". +KIND_NONE = 0.0 +KIND_WAIT = 1.0 +KIND_BUSY = 2.0 + +# Handed to a child when the pool has no slot left. Reads and writes are no-ops. +NO_SLOT = -1 + +SEQLOCK_READ_ATTEMPTS = 3 + + +@dataclass(frozen=True) +class SampleResult: + """One child's contribution to a flush, plus why it may be short. + + `eligible` is how much of the interval this child could have accrued in at + all. It differs from the interval width for a child baselined part-way + through, which the `elapsed * running_count` ceiling treats as whole. + """ + + busy: float = 0.0 + wait: float = 0.0 + eligible: float = 0.0 + reason: str = "ok" + + +def slot_count(concurrency: int) -> int: + """Twice concurrency, so a generation of unreaped exiting children can + overlap a generation of replacements.""" + return max(1, concurrency * 2) + + +class ChildTimeWriter: + """Child-side writer for one slot. + + Sole writer, so it keeps authoritative totals as plain floats and + republishes the whole slot on each transition. + """ + + __slots__ = ("_shm", "_slot", "_base", "_busy_total", "_wait_total", "_start", "_kind") + + def __init__(self, shm: ctypes.Array[ctypes.c_double] | None, slot: int) -> None: + self._shm = shm + self._slot = NO_SLOT if shm is None else slot + self._base = slot * SLOT_WIDTH + self._busy_total = 0.0 + self._wait_total = 0.0 + self._start = 0.0 + self._kind = KIND_NONE + + def _publish(self) -> None: + if self._slot == NO_SLOT or self._shm is None: + return + + shm = self._shm + base = self._base + + version = shm[base + SLOT_VERSION] + # Odd: a reader that sees this discards what it read. + shm[base + SLOT_VERSION] = version + 1.0 + shm[base + SLOT_BUSY_TOTAL] = self._busy_total + shm[base + SLOT_WAIT_TOTAL] = self._wait_total + shm[base + SLOT_SEGMENT_START] = self._start + shm[base + SLOT_SEGMENT_KIND] = self._kind + shm[base + SLOT_VERSION] = version + 2.0 + + def _close_open(self, now: float) -> None: + if self._kind == KIND_BUSY: + self._busy_total += max(0.0, now - self._start) + elif self._kind == KIND_WAIT: + self._wait_total += max(0.0, now - self._start) + self._kind = KIND_NONE + + def mark_running(self, now: float) -> None: + """Open the wait clock. A warmed-up child with no task yet is waiting.""" + if self._kind != KIND_NONE: + return + + self._start = now + self._kind = KIND_WAIT + self._publish() + + def mark_busy(self, now: float) -> None: + """Close the open wait segment and open a busy one.""" + self._close_open(now) + self._start = now + self._kind = KIND_BUSY + self._publish() + + def mark_idle(self, now: float) -> None: + """Close the open busy segment and open a wait one.""" + self._close_open(now) + self._start = now + self._kind = KIND_WAIT + self._publish() + + def close(self, now: float) -> None: + """Close whatever is open so a departing child stops folding time forward.""" + self._close_open(now) + self._publish() + + +@dataclass +class ChildTimeAccounting: + """Parent-side reader for one child's slot. + + Holds the previous absolute reading and returns deltas, so a child in a long + task contributes to every interval it spans. + """ + + shm: ctypes.Array[ctypes.c_double] | None + slot: int = NO_SLOT + _prev_busy: float = 0.0 + _prev_wait: float = 0.0 + _accounted: bool = False + # When this child started being counted, so a flush can tell a short + # sample apart from a child that was only eligible for part of it. + _baselined_at: float = 0.0 + + def mark_running(self, now: float) -> None: + """Start counting this child, baselining against the slot as it stands. + + Baselining rather than zeroing drops whatever the child banked before the + parent saw its `running` message, which is the same window over which it + is absent from `running_count`. + """ + reading = self._read(now) + if reading is None: + self._prev_busy = 0.0 + self._prev_wait = 0.0 + else: + self._prev_busy, self._prev_wait = reading + + self._accounted = True + self._baselined_at = now + + def mark_stopped(self) -> None: + """Stop counting, so an exiting child's tail misses the live pool.""" + self._accounted = False + + def sample(self, now: float, interval_start: float = 0.0) -> SampleResult: + """Return this child's busy/wait since the previous sample.""" + if not self._accounted: + return SampleResult(reason="not_accounted") + + eligible = max(0.0, now - max(interval_start, self._baselined_at)) + + reading = self._read(now) + if reading is None: + # Baseline untouched, so the next sample covers both intervals. + return SampleResult(eligible=eligible, reason="read_failed") + + busy_now, wait_now = reading + raw_busy = busy_now - self._prev_busy + raw_wait = wait_now - self._prev_wait + # A total going backwards means a torn read or a reused slot, and the + # clamp below silently drops that time. + reason = "clamped" if raw_busy < 0.0 or raw_wait < 0.0 else "ok" + + self._prev_busy = busy_now + self._prev_wait = wait_now + return SampleResult(max(0.0, raw_busy), max(0.0, raw_wait), eligible, reason) + + def _read(self, now: float) -> tuple[float, float] | None: + """Seqlock read of absolute busy/wait, including the segment still open. + + None means the read could not be taken cleanly. + """ + if self.slot == NO_SLOT or self.shm is None: + return None + + shm = self.shm + base = self.slot * SLOT_WIDTH + + for _ in range(SEQLOCK_READ_ATTEMPTS): + version = shm[base + SLOT_VERSION] + if version % 2.0: + continue + + busy = shm[base + SLOT_BUSY_TOTAL] + wait = shm[base + SLOT_WAIT_TOTAL] + start = shm[base + SLOT_SEGMENT_START] + kind = shm[base + SLOT_SEGMENT_KIND] + + if shm[base + SLOT_VERSION] != version: + continue + + # Fold in the segment the child is in right now. + if kind == KIND_BUSY: + busy += max(0.0, now - start) + elif kind == KIND_WAIT: + wait += max(0.0, now - start) + + return (busy, wait) + + return None diff --git a/clients/python/src/taskbroker_client/worker/worker.py b/clients/python/src/taskbroker_client/worker/worker.py index f429f6a2..c2c1803b 100644 --- a/clients/python/src/taskbroker_client/worker/worker.py +++ b/clients/python/src/taskbroker_client/worker/worker.py @@ -1,5 +1,6 @@ from __future__ import annotations +import ctypes import logging import multiprocessing import os @@ -40,6 +41,12 @@ ) from taskbroker_client.metrics import MetricsBackend from taskbroker_client.types import InflightTaskActivation, ProcessingResult +from taskbroker_client.worker.childtiming import ( + NO_SLOT, + SLOT_WIDTH, + ChildTimeAccounting, + slot_count, +) from taskbroker_client.worker.client import ( HealthCheckSettings, HostTemporarilyUnavailable, @@ -132,6 +139,22 @@ def __init__( registry=self.registry, ) + # Additive and unclamped: the scaler sums across pods and divides once. + self.child_busy_seconds = prometheus_client.Counter( + "taskworker_worker_child_busy_seconds", + "Cumulative child-seconds spent executing tasks.", + ["processing_pool"], + registry=self.registry, + ) + + # What occupancy cannot express: free slots with nothing to do. + self.child_wait_seconds = prometheus_client.Counter( + "taskworker_worker_child_wait_seconds", + "Cumulative child-seconds spent blocked waiting for a task to arrive.", + ["processing_pool"], + registry=self.registry, + ) + prometheus_client.start_http_server(port, registry=self.registry) logger.info("taskworker.worker.prometheus_server_started", extra={"port": port}) @@ -211,43 +234,8 @@ class TrackedChild: process: BaseProcess state: ChildState release: Event - # Time-weighted busy tracking - busy_since: float | None = None # monotonic timestamp of the currently-open busy segment - busy_accumulated: float = 0.0 # the busy seconds banked since the last occupancy flush - - def mark_busy(self, now: float) -> None: - """Open a busy segment when the child starts a task. - - `busy`/`idle` strictly alternate per child today, so a segment should - never already be open; the guard is defensive and keeps the original - start time if that invariant ever drifts. - """ - if self.busy_since is None: - self.busy_since = now - - def mark_idle(self, now: float) -> None: - """Close the open busy segment and bank its elapsed seconds. - - Guarded so an unexpected `idle` with no open segment is a no-op rather - than a crash. - """ - if self.busy_since is not None: - self.busy_accumulated += now - self.busy_since - self.busy_since = None - - def drain_busy(self, now: float) -> float: - """Return busy seconds since the last drain and reset the counter. - - Any segment still open is folded in up to `now` and left open (its - start advanced to `now`) so a task spanning multiple intervals keeps - contributing to each one. - """ - if self.busy_since is not None: - self.busy_accumulated += now - self.busy_since - self.busy_since = now - banked = self.busy_accumulated - self.busy_accumulated = 0.0 - return banked + # No default: an accountant with no slot silently measures nothing. + timing: ChildTimeAccounting class PushTaskWorker: @@ -920,6 +908,13 @@ def __init__( ) self._children: Dict[UUID, TrackedChild] = {} self._exiting_children: Deque[UUID] = deque() + + # Two generations: unreaped exiting children overlap their replacements. + self._timing_slots: int = slot_count(concurrency) + self._timing_shm: ctypes.Array[ctypes.c_double] = self._mp_context.RawArray( + "d", SLOT_WIDTH * self._timing_slots + ) + self._free_timing_slots: Deque[int] = deque(range(self._timing_slots)) self._children_lock = threading.Lock() self._last_occupancy_flush_at = time.monotonic() self._shutdown_event = self._mp_context.Event() @@ -929,6 +924,50 @@ def __init__( self._metrics_thread: threading.Thread | None = None self._spawn_children_thread: threading.Thread | None = None + def _acquire_timing_slot(self) -> int: + """Take a zeroed shared-memory slot for a new child. + + `NO_SLOT` means the pool ran out, which the two-generation sizing is + meant to make impossible. That child then contributes to neither the + occupancy numerator nor `running_count`, so the ratio stays consistent + across the children that are accounted for, and the metric below says + the sizing was wrong. + """ + with self._children_lock: + if not self._free_timing_slots: + slot = NO_SLOT + else: + slot = self._free_timing_slots.popleft() + + if slot == NO_SLOT: + logger.error( + "taskworker.child.timing_slot_exhausted", + extra={ + "slots": self._timing_slots, + "processing_pool": self._processing_pool_name, + }, + ) + self._metrics.incr( + "taskworker.worker.child.timing_slot_exhausted", + tags={"processing_pool": self._processing_pool_name}, + ) + return NO_SLOT + + base = slot * SLOT_WIDTH + for offset in range(SLOT_WIDTH): + self._timing_shm[base + offset] = 0.0 + + return slot + + def _release_timing_slot(self, slot: int) -> None: + """Return a slot whose child never started. The reap path returns slots + inline because it already holds `_children_lock`.""" + if slot == NO_SLOT: + return + + with self._children_lock: + self._free_timing_slots.append(slot) + @property def ready_count(self) -> int: """Number of children that have finished warming up and are consuming.""" @@ -970,18 +1009,120 @@ def _emit_periodic_metrics(self) -> None: } busy_time = 0.0 + wait_time = 0.0 + accounted_running = 0 + eligible_time = 0.0 + outcomes: dict[str, int] = {} + interval_start = self._last_occupancy_flush_at for child in self._children.values(): state_counts[child.state] += 1 - busy_time += child.drain_busy(now) + + # Running only: the numerator must match occupancy's divisor. + if child.state != "running": + continue + + # A slotless child reports nothing; counting it would read as idle. + if child.timing.slot == NO_SLOT: + continue + + accounted_running += 1 + result = child.timing.sample(now, interval_start) + busy_time += result.busy + wait_time += result.wait + eligible_time += result.eligible + outcomes[result.reason] = outcomes.get(result.reason, 0) + 1 exiting_children = len(self._exiting_children) elapsed = now - self._last_occupancy_flush_at self._last_occupancy_flush_at = now - running_count = state_counts["running"] + # Emitted during warmup too: zero is correct for a counter. + self._metrics.distribution( + "taskworker.worker.child_busy_seconds", + busy_time, + tags=tags, + ) + self._metrics.distribution( + "taskworker.worker.child_wait_seconds", + wait_time, + tags=tags, + ) + if self._prom is not None: + # inc(0.0) registers the series, so a new pod reads idle not missing. + self._prom.child_busy_seconds.labels(processing_pool=self._processing_pool_name).inc( + max(0.0, busy_time) + ) + self._prom.child_wait_seconds.labels(processing_pool=self._processing_pool_name).inc( + max(0.0, wait_time) + ) + + # Slotless children are out of both sides; the gauge still counts them. + running_count = accounted_running if running_count > 0 and elapsed > 0: - occupancy = busy_time / (elapsed * running_count) + # Physical bound. Exceeding it means the clamp below is hiding a bug. + ceiling = elapsed * running_count + if busy_time > ceiling or wait_time > ceiling: + self._metrics.incr( + "taskworker.worker.occupancy.accounting_overflow", + tags=tags, + ) + logger.warning( + "taskworker.worker.occupancy.accounting_overflow", + extra={ + "busy_time": busy_time, + "wait_time": wait_time, + "ceiling": ceiling, + "running_count": running_count, + "elapsed": elapsed, + "processing_pool": self._processing_pool_name, + }, + ) + + # How much of the ceiling the counters actually account for. The + # overflow guard above is one-sided; this is the other direction, + # where occupancy reads low because time went missing rather than + # because the pool was idle. + self._metrics.gauge( + "taskworker.worker.occupancy.accounting_ratio", + (busy_time + wait_time) / ceiling, + tags=tags, + ) + # Same numerator against the time children were actually eligible + # for. A child baselined mid-interval counts whole in `ceiling` but + # only partly here, so if this reads ~1.0 while the ratio above + # reads low, the denominator is the fault, not the accounting. + if eligible_time > 0: + self._metrics.gauge( + "taskworker.worker.occupancy.eligible_ratio", + (busy_time + wait_time) / eligible_time, + tags=tags, + ) + for reason, count in outcomes.items(): + self._metrics.incr( + "taskworker.worker.occupancy.sample_outcome", + count, + tags={**tags, "reason": reason}, + ) + if busy_time + wait_time < ceiling * 0.9: + self._metrics.incr( + "taskworker.worker.occupancy.accounting_deficit", + tags=tags, + ) + logger.warning( + "taskworker.worker.occupancy.accounting_deficit", + extra={ + "busy_time": busy_time, + "wait_time": wait_time, + "eligible_time": eligible_time, + "running_count": running_count, + "elapsed": elapsed, + "outcomes": outcomes, + "processing_pool": self._processing_pool_name, + }, + ) + + occupancy = busy_time / ceiling occupancy = min(occupancy, 1.0) self._metrics.gauge( "taskworker.worker.occupancy", @@ -1121,6 +1262,18 @@ def spawn_children_thread() -> None: except queue.Empty: break + # Lifecycle-queue lag. A rising line means this thread is behind. + if received: + drain_at = time.monotonic() + self._metrics.distribution( + "taskworker.worker.child_message.age", + drain_at - min(m.timestamp for m in received), + tags={ + "processing_pool": self._processing_pool_name, + "pod_name": self._pod_name, + }, + ) + with self._children_lock: children = list(self._children.items()) @@ -1131,6 +1284,10 @@ def spawn_children_thread() -> None: c.process.join(timeout=0) self._children.pop(cid) + # Not at `exiting`: a released child can still publish once. + if c.timing.slot != NO_SLOT: + self._free_timing_slots.append(c.timing.slot) + logger.info( "taskworker.child.exited", extra={ @@ -1158,23 +1315,15 @@ def spawn_children_thread() -> None: continue - # This child is now running + # Baseline here, where it also enters `running_count`. if message.event == "running": child.state = "running" + child.timing.mark_running(time.monotonic()) # This child wants to exit, but we may not have enough running children to shut down right away elif message.event == "exiting": self._exiting_children.append(message.child_id) - # This child started executing a task: open a busy segment. - elif message.event == "busy": - child.mark_busy(time.monotonic()) - - # This child finished a task: close the open busy segment - # and bank the elapsed time. - elif message.event == "idle": - child.mark_idle(time.monotonic()) - while True: # Compute how many children are still running running = sum(1 for c in self._children.values() if c.state == "running") @@ -1195,6 +1344,7 @@ def spawn_children_thread() -> None: continue child.state = "exiting" + child.timing.mark_stopped() child.release.set() spawned = sum(1 for c in self._children.values() if c.state != "exiting") @@ -1205,6 +1355,7 @@ def spawn_children_thread() -> None: for _ in range(needed): child_id = uuid4() release = self._mp_context.Event() + timing_slot = self._acquire_timing_slot() process = self._mp_context.Process( name=f"taskworker-child-{child_id}", @@ -1222,6 +1373,8 @@ def spawn_children_thread() -> None: self._future_checking_frequency, messages, release, + self._timing_shm, + timing_slot, ), ) @@ -1233,10 +1386,14 @@ def spawn_children_thread() -> None: process=process, state="pending", release=release, + timing=ChildTimeAccounting(shm=self._timing_shm, slot=timing_slot), ) self._children[child_id] = child except Exception as e: + # Never came up, so nothing will write to its slot. + self._release_timing_slot(timing_slot) + logger.exception( "taskworker.child.spawn.failed", extra={ diff --git a/clients/python/src/taskbroker_client/worker/workerchild.py b/clients/python/src/taskbroker_client/worker/workerchild.py index ac029f19..3e71ef8a 100644 --- a/clients/python/src/taskbroker_client/worker/workerchild.py +++ b/clients/python/src/taskbroker_client/worker/workerchild.py @@ -1,6 +1,7 @@ from __future__ import annotations import contextlib +import ctypes import logging import multiprocessing import queue @@ -8,7 +9,7 @@ import threading import time from collections.abc import Callable, Generator, Sequence -from dataclasses import dataclass +from dataclasses import dataclass, field from functools import partial from multiprocessing.synchronize import Event from types import FrameType @@ -40,6 +41,7 @@ from taskbroker_client.state import clear_current_task, current_task, set_current_task from taskbroker_client.task import Task from taskbroker_client.types import ContextHook, InflightTaskActivation, ProcessingResult +from taskbroker_client.worker.childtiming import ChildTimeWriter logger = logging.getLogger(__name__) @@ -169,7 +171,9 @@ def _log_task_retry_exhausted( @dataclass(frozen=True) class ChildMessage: child_id: UUID - event: Literal["running", "exiting", "busy", "idle"] + event: Literal["running", "exiting"] + # compare=False: the timestamp is payload, not identity. + timestamp: float = field(default_factory=time.monotonic, compare=False) def child_process( @@ -185,6 +189,8 @@ def child_process( future_checking_frequency: float, messages: multiprocessing.Queue[ChildMessage], parent_release: Event, + timing_shm: ctypes.Array[ctypes.c_double] | None, + timing_slot: int, ) -> None: """ The entrypoint for spawned worker children. @@ -196,6 +202,9 @@ def child_process( app = import_app(app_module) app.load_modules() metrics = app.metrics + + # Straight to shared memory: `messages` cannot keep up at two events per task. + timing = ChildTimeWriter(timing_shm, timing_slot) # Signals when the parent worker pool terminates the child local_shutdown = threading.Event() @@ -377,7 +386,7 @@ def check_task_future_completion( # the child did since its last dequeue has finished, and what follows # is waiting for the next task. if is_busy: - messages.put_nowait(ChildMessage(child_id, "idle")) + timing.mark_idle(time.monotonic()) is_busy = False if max_task_count and processed_task_count >= max_task_count: @@ -437,7 +446,7 @@ def check_task_future_completion( # Open the busy segment as soon as we have a task. The slot is now # unavailable for new work, whatever stage of handling it is in. - messages.put_nowait(ChildMessage(child_id, "busy")) + timing.mark_busy(time.monotonic()) is_busy = True task_func = _get_known_task(inflight.activation) @@ -625,7 +634,7 @@ def check_task_future_completion( # signal can land while a segment is still open. Close it so a child that # is going away doesn't keep contributing busy time to the pool's occupancy. if is_busy: - messages.put_nowait(ChildMessage(child_id, "idle")) + timing.close(time.monotonic()) is_busy = False # Once we get the shutdown signal, drain any pending futures @@ -672,7 +681,8 @@ def _execute_activation( transaction.set_data("taskworker-task.args", args) transaction.set_data("taskworker-task.kwargs", kwargs) - task_added_time = activation.received_at.ToDatetime().timestamp() + # ToDatetime().timestamp() misreads a naive UTC datetime as local. + task_added_time = activation.received_at.seconds + activation.received_at.nanos / 1e9 # latency attribute needs to be in milliseconds latency = (time.time() - task_added_time) * 1000 @@ -724,9 +734,12 @@ def record_task_execution( taskbroker_host: str, futures_enqueued_time: float | None = None, ) -> None: - task_added_time = activation.received_at.ToDatetime().timestamp() + # ToDatetime().timestamp() reads a naive UTC datetime as local time. + task_added_time = activation.received_at.seconds + activation.received_at.nanos / 1e9 execution_duration = completion_time - start_time execution_latency = completion_time - task_added_time + # Latency minus the task's own cost, so it does not scale with duration. + queue_wait = max(0.0, start_time - task_added_time) futures_duration = time.time() - futures_enqueued_time if futures_enqueued_time else 0 logger.debug( @@ -768,6 +781,16 @@ def record_task_execution( "taskbroker_host": taskbroker_host, }, ) + metrics.distribution( + "taskworker.worker.queue_wait", + queue_wait, + tags={ + "namespace": activation.namespace, + "taskname": activation.taskname, + "processing_pool": processing_pool_name, + "taskbroker_host": taskbroker_host, + }, + ) if futures_duration != 0: metrics.distribution( "taskworker.worker.future_completion_duration", @@ -885,6 +908,7 @@ def _task_execution_complete( ) # Tell the parent that this child has warmed up and is ready to consume tasks + timing.mark_running(time.monotonic()) messages.put_nowait(ChildMessage(child_id, "running")) # Run the worker loop diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py index 9045ab04..33a79c88 100644 --- a/clients/python/tests/worker/test_worker.py +++ b/clients/python/tests/worker/test_worker.py @@ -1,12 +1,14 @@ import contextlib +import itertools import os import queue +import random import signal import threading import time from collections.abc import Iterator, MutableMapping from concurrent.futures import Future -from datetime import datetime +from datetime import datetime, timezone from multiprocessing import Event, get_context from multiprocessing.synchronize import Event as MultiprocessingEvent from pathlib import Path @@ -43,6 +45,21 @@ from taskbroker_client.retry import NoRetriesRemainingError from taskbroker_client.state import current_task from taskbroker_client.types import InflightTaskActivation, ProcessingResult +from taskbroker_client.worker.childtiming import ( + KIND_BUSY, + KIND_NONE, + KIND_WAIT, + NO_SLOT, + SLOT_BUSY_TOTAL, + SLOT_SEGMENT_KIND, + SLOT_SEGMENT_START, + SLOT_VERSION, + SLOT_WAIT_TOTAL, + SLOT_WIDTH, + ChildTimeAccounting, + ChildTimeWriter, + slot_count, +) from taskbroker_client.worker.worker import ( PushTaskWorker, ShutdownSignal, @@ -314,6 +331,8 @@ def child_process( future_checking_frequency, messages, parent_release, + ctx.RawArray("d", SLOT_WIDTH), + 0, ) @@ -410,6 +429,9 @@ def Process(self, *args: Any, **kwargs: Any) -> _FakeProcess: self.processes.append(process) return process + def RawArray(self, typecode: str, size: int) -> Any: + return get_context("fork").RawArray(typecode, size) + def _make_fake_context_pool( fake_context: _FakeContext, @@ -1275,21 +1297,71 @@ def ready_count() -> int: pool_shutdown.assert_called_once_with() +# Independent of any pool: the flush reads through `child.timing.shm`. +_TEST_SLOTS = 512 +_TEST_TIMING_SHM = get_context("fork").RawArray("d", SLOT_WIDTH * _TEST_SLOTS) +_TEST_SLOT_SEQ = itertools.count() + + def _make_tracked_child( state: str, *, busy_since: float | None = None, busy_accumulated: float = 0.0, + wait_since: float | None = None, + wait_accumulated: float = 0.0, + baselined_at: float | None = None, ) -> TrackedChild: + """Seed a child's slot so its next sample reports the given time. + + `*_accumulated` is time the child has already closed; `*_since` leaves a + segment open at that monotonic time, which the parent folds forward at + sample time exactly as it would for a task still running. + """ + slot = next(_TEST_SLOT_SEQ) % _TEST_SLOTS + base = slot * SLOT_WIDTH + + for offset in range(SLOT_WIDTH): + _TEST_TIMING_SHM[base + offset] = 0.0 + + timing = ChildTimeAccounting(shm=_TEST_TIMING_SHM, slot=slot) + # Baseline against the zeroed slot, so the seeding below lands in sample 1. + timing.mark_running(0.0 if baselined_at is None else baselined_at) + + if busy_since is not None: + kind, start = KIND_BUSY, busy_since + elif wait_since is not None: + kind, start = KIND_WAIT, wait_since + else: + kind, start = KIND_NONE, 0.0 + + _TEST_TIMING_SHM[base + SLOT_VERSION] = 2.0 + _TEST_TIMING_SHM[base + SLOT_BUSY_TOTAL] = busy_accumulated + _TEST_TIMING_SHM[base + SLOT_WAIT_TOTAL] = wait_accumulated + _TEST_TIMING_SHM[base + SLOT_SEGMENT_START] = start + _TEST_TIMING_SHM[base + SLOT_SEGMENT_KIND] = kind + return TrackedChild( process=mock.Mock(), state=state, # type: ignore[arg-type] release=mock.Mock(), - busy_since=busy_since, - busy_accumulated=busy_accumulated, + timing=timing, ) +def _bw(result: Any) -> tuple[float, float]: + """The (busy, wait) pair from a SampleResult, dropping the diagnostics.""" + return (result.busy, result.wait) + + +def _distribution_calls(metrics: mock.Mock, name: str) -> list[Any]: + return [c for c in metrics.distribution.call_args_list if c.args[0] == name] + + +def _incr_calls(metrics: mock.Mock, name: str) -> list[Any]: + return [c for c in metrics.incr.call_args_list if c.args[0] == name] + + def _gauge_calls(metrics: mock.Mock, name: str) -> list[Any]: return [c for c in metrics.gauge.call_args_list if c.args[0] == name] @@ -1341,10 +1413,11 @@ def test_emit_periodic_metrics_time_weights_busy_over_the_interval() -> None: assert len(occupancy_calls) == 1 assert occupancy_calls[0].args[1] == pytest.approx(1.24 / 3) - # The open segment is carried into the next interval; banks are drained. - assert pool._children[child_b].busy_since == pytest.approx(11.0) + # B's segment is still open, so the next interval resumes, not re-reports. + assert _bw(pool._children[child_b].timing.sample(12.0)) == pytest.approx((1.0, 0.0)) for child in pool._children.values(): - assert child.busy_accumulated == 0.0 + if child.state == "running": + assert child.timing.sample(12.0).busy == pytest.approx(0.0) assert pool._last_occupancy_flush_at == pytest.approx(11.0) @@ -1370,16 +1443,15 @@ def test_emit_periodic_metrics_divides_by_running_children() -> None: assert occupancy_calls[0].args[1] == pytest.approx(2 / 3) -def test_emit_periodic_metrics_clamps_occupancy_to_one() -> None: - # A draining child can still be mid-task, so busy-time can exceed the running - # capacity for the interval; occupancy must clamp to 1.0. +def test_emit_periodic_metrics_clamps_occupancy_and_flags_the_overflow() -> None: + # 1.5s of busy in a 1s interval is a fault; the clamp must not hide it. pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4) pool._metrics = mock.Mock() pool._last_occupancy_flush_at = 10.0 with pool._children_lock: - pool._children[uuid4()] = _make_tracked_child("running", busy_accumulated=1.0) - pool._children[uuid4()] = _make_tracked_child("exiting", busy_accumulated=1.0) + pool._children[uuid4()] = _make_tracked_child("running", busy_accumulated=1.5) + pool._children[uuid4()] = _make_tracked_child("running", busy_accumulated=1.5) with mock.patch("taskbroker_client.worker.worker.time.monotonic", return_value=11.0): pool._emit_periodic_metrics() @@ -1387,9 +1459,394 @@ def test_emit_periodic_metrics_clamps_occupancy_to_one() -> None: occupancy_calls = _gauge_calls(pool._metrics, "taskworker.worker.occupancy") assert len(occupancy_calls) == 1 assert occupancy_calls[0].args[1] == pytest.approx(1.0) + assert len(_incr_calls(pool._metrics, "taskworker.worker.occupancy.accounting_overflow")) == 1 + + +def test_emit_periodic_metrics_does_not_flag_a_legitimately_full_pool() -> None: + # The guard must not fire on a pool that is simply saturated, or it is noise. + pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4) + pool._metrics = mock.Mock() + pool._last_occupancy_flush_at = 10.0 + + with pool._children_lock: + pool._children[uuid4()] = _make_tracked_child("running", busy_since=10.0) + pool._children[uuid4()] = _make_tracked_child("running", busy_since=10.0) + + with mock.patch("taskbroker_client.worker.worker.time.monotonic", return_value=11.0): + pool._emit_periodic_metrics() + + assert _gauge_calls(pool._metrics, "taskworker.worker.occupancy")[0].args[1] == pytest.approx( + 1.0 + ) + assert _incr_calls(pool._metrics, "taskworker.worker.occupancy.accounting_overflow") == [] + + +def test_emit_periodic_metrics_separates_a_short_sample_from_a_short_interval() -> None: + # The discriminator for the cell-6 deficit. A child baselined mid-interval can + # only accrue over part of it, but `elapsed * running_count` counts it whole, + # so occupancy reads low with no time actually missing. eligible_ratio near 1.0 + # while accounting_ratio reads low means the denominator is at fault. + pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4) + pool._metrics = mock.Mock() + pool._last_occupancy_flush_at = 10.0 + + with pool._children_lock: + pool._children[uuid4()] = _make_tracked_child("running", busy_since=10.0) + pool._children[uuid4()] = _make_tracked_child("running", busy_since=10.5, baselined_at=10.5) + + with mock.patch("taskbroker_client.worker.worker.time.monotonic", return_value=11.0): + pool._emit_periodic_metrics() + + # 1.0 + 0.5 busy-seconds against a ceiling of 2.0. + assert _gauge_calls(pool._metrics, "taskworker.worker.occupancy.accounting_ratio")[0].args[ + 1 + ] == pytest.approx(0.75) + # But only 1.5 child-seconds were ever available. + assert _gauge_calls(pool._metrics, "taskworker.worker.occupancy.eligible_ratio")[0].args[ + 1 + ] == pytest.approx(1.0) + # Nothing went missing, so every sample is clean. + reasons = { + c.kwargs["tags"]["reason"] + for c in _incr_calls(pool._metrics, "taskworker.worker.occupancy.sample_outcome") + } + assert reasons == {"ok"} + + +def test_emit_periodic_metrics_flags_a_deficit_and_names_the_reason() -> None: + # The mirror of accounting_overflow. Occupancy reading low because time went + # missing is the exact failure this project started from, and the overflow + # guard is blind to it. + pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4) + pool._metrics = mock.Mock() + pool._last_occupancy_flush_at = 10.0 + + stalled = _make_tracked_child("running", busy_since=10.0) + stalled.timing.mark_stopped() + + with pool._children_lock: + pool._children[uuid4()] = _make_tracked_child("running", busy_since=10.0) + pool._children[uuid4()] = stalled + + with mock.patch("taskbroker_client.worker.worker.time.monotonic", return_value=11.0): + pool._emit_periodic_metrics() + + assert len(_incr_calls(pool._metrics, "taskworker.worker.occupancy.accounting_deficit")) == 1 + assert _gauge_calls(pool._metrics, "taskworker.worker.occupancy.accounting_ratio")[0].args[ + 1 + ] == pytest.approx(0.5) + reasons = { + c.kwargs["tags"]["reason"] + for c in _incr_calls(pool._metrics, "taskworker.worker.occupancy.sample_outcome") + } + assert reasons == {"ok", "not_accounted"} -def test_spawn_children_tracks_busy_and_idle_transitions() -> None: +def test_emit_periodic_metrics_does_not_flag_a_deficit_when_time_is_all_there() -> None: + pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4) + pool._metrics = mock.Mock() + pool._last_occupancy_flush_at = 10.0 + + with pool._children_lock: + pool._children[uuid4()] = _make_tracked_child("running", busy_since=10.0) + pool._children[uuid4()] = _make_tracked_child("running", wait_since=10.0) + + with mock.patch("taskbroker_client.worker.worker.time.monotonic", return_value=11.0): + pool._emit_periodic_metrics() + + assert _incr_calls(pool._metrics, "taskworker.worker.occupancy.accounting_deficit") == [] + assert _gauge_calls(pool._metrics, "taskworker.worker.occupancy.accounting_ratio")[0].args[ + 1 + ] == pytest.approx(1.0) + + +def test_sample_reports_a_clamped_read_rather_than_hiding_it() -> None: + # A total going backwards means a torn read or a reused slot. The clamp keeps + # the number sane but drops real time, so the drop has to be visible. + writer, reader = _writer_and_reader() + writer.mark_running(0.0) + reader.mark_running(0.0) + writer.mark_busy(0.0) + assert reader.sample(1.0).reason == "ok" + + # Rewind the slot underneath the reader, as slot reuse would. + reader.shm[SLOT_BUSY_TOTAL] = 0.0 # type: ignore[index] + reader.shm[SLOT_SEGMENT_START] = 2.0 # type: ignore[index] + + result = reader.sample(2.0) + assert result.reason == "clamped" + assert result.busy == 0.0 + + +def test_emit_periodic_metrics_excludes_slotless_children_from_occupancy() -> None: + # A slotless child in the denominator would read as idle and halve occupancy. + pool = _make_result_thread_pool(_SendResultCapture(), concurrency=4) + pool._metrics = mock.Mock() + pool._last_occupancy_flush_at = 10.0 + + slotless = _make_tracked_child("running") + slotless.timing.slot = NO_SLOT + + with pool._children_lock: + pool._children[uuid4()] = _make_tracked_child("running", busy_since=10.0) + pool._children[uuid4()] = slotless + + with mock.patch("taskbroker_client.worker.worker.time.monotonic", return_value=11.0): + pool._emit_periodic_metrics() + + # One accounted child, busy for the whole interval. + assert _gauge_calls(pool._metrics, "taskworker.worker.occupancy")[0].args[1] == pytest.approx( + 1.0 + ) + assert _incr_calls(pool._metrics, "taskworker.worker.occupancy.accounting_overflow") == [] + + running_gauges = [ + c + for c in pool._metrics.gauge.call_args_list + if c.args[0] == "taskworker.worker.children" and c.kwargs["tags"]["state"] == "running" + ] + assert running_gauges[0].args[1] == 2.0 + + +def test_emit_periodic_metrics_counters_exclude_non_running_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() + pool._last_occupancy_flush_at = 10.0 + + with pool._children_lock: + pool._children[uuid4()] = _make_tracked_child("running", busy_since=10.0) + pool._children[uuid4()] = _make_tracked_child("exiting", busy_accumulated=1.0) + pool._children[uuid4()] = _make_tracked_child("pending") + + with mock.patch("taskbroker_client.worker.worker.time.monotonic", return_value=11.0): + pool._emit_periodic_metrics() + + busy = _distribution_calls(pool._metrics, "taskworker.worker.child_busy_seconds") + assert busy[0].args[1] == pytest.approx(1.0) + assert _gauge_calls(pool._metrics, "taskworker.worker.occupancy")[0].args[1] == pytest.approx( + 1.0 + ) + + +def test_spawn_children_binds_each_child_to_its_own_timing_slot() -> None: + # If spawn and flush disagree on the slot, the pool measures nothing. + fake_context = _FakeContext() + pool = _make_fake_context_pool(fake_context, concurrency=2) + + pool.start_spawn_children_thread() + try: + _wait_for(lambda: len(fake_context.processes) == 2) + messages = fake_context.queues[-1] + + slots: set[int] = set() + for process in fake_context.processes: + child_id = process.args[0] + shm, slot = process.args[-2], process.args[-1] + slots.add(slot) + + messages.put(ChildMessage(child_id, "running")) + # _wait_for blocks, so the closure resolves inside the iteration. + _wait_for(lambda: pool._children[child_id].state == "running") + assert pool._children[child_id].timing.slot == slot + assert shm is pool._timing_shm + + # Two children, two distinct slots. + assert len(slots) == 2 + finally: + pool.shutdown() + + +def _writer_and_reader(slot: int = 0) -> tuple[ChildTimeWriter, ChildTimeAccounting]: + shm = get_context("fork").RawArray("d", SLOT_WIDTH * (slot + 1)) + return ChildTimeWriter(shm, slot), ChildTimeAccounting(shm=shm, slot=slot) + + +def test_child_timing_round_trips_through_shared_memory() -> None: + # The child records the transition itself; nothing crosses a queue. + writer, reader = _writer_and_reader() + + writer.mark_running(0.0) + reader.mark_running(0.0) + writer.mark_busy(1.0) + + assert _bw(reader.sample(2.0)) == pytest.approx((1.0, 1.0)) + + +def test_child_timing_credits_a_long_task_to_every_interval_it_spans() -> None: + # Without this a 60s task reports zero for 60 flushes, then 60s at once. + writer, reader = _writer_and_reader() + + writer.mark_running(0.0) + reader.mark_running(0.0) + writer.mark_busy(0.0) + + assert _bw(reader.sample(1.0)) == pytest.approx((1.0, 0.0)) + assert _bw(reader.sample(2.0)) == pytest.approx((1.0, 0.0)) + + writer.mark_idle(2.5) + assert _bw(reader.sample(3.0)) == pytest.approx((0.5, 0.5)) + + +def test_child_timing_busy_and_wait_partition_every_interval() -> None: + # The invariant `occupancy.accounting_overflow` guards in production. + writer, reader = _writer_and_reader() + writer.mark_running(0.0) + reader.mark_running(0.0) + + rng = random.Random(20260902) + now = 0.0 + busy = False + total_busy = total_wait = 0.0 + + for i in range(500): + now += rng.uniform(0.001, 0.05) + busy = not busy + (writer.mark_busy if busy else writer.mark_idle)(now) + + if i % 7 == 0: + b, w = _bw(reader.sample(now)) + total_busy += b + total_wait += w + + b, w = _bw(reader.sample(now)) + total_busy += b + total_wait += w + + assert total_busy + total_wait == pytest.approx(now) + + +def test_child_timing_defers_rather_than_drops_a_torn_read() -> None: + # Leaving the baseline alone delays attribution instead of losing it. + writer, reader = _writer_and_reader() + writer.mark_running(0.0) + reader.mark_running(0.0) + writer.mark_busy(0.0) + + assert _bw(reader.sample(1.0)) == pytest.approx((1.0, 0.0)) + + reader.shm[SLOT_VERSION] += 1.0 # type: ignore[index] + assert _bw(reader.sample(2.0)) == pytest.approx((0.0, 0.0)) + + reader.shm[SLOT_VERSION] += 1.0 # type: ignore[index] + assert _bw(reader.sample(3.0)) == pytest.approx((2.0, 0.0)) + + +def test_child_timing_stops_accruing_once_the_child_is_released() -> None: + # Otherwise the segment folds forward forever and a recycling pool looks starved. + writer, reader = _writer_and_reader() + writer.mark_running(0.0) + reader.mark_running(0.0) + + assert reader.sample(0.5).wait == pytest.approx(0.5) + + reader.mark_stopped() + assert _bw(reader.sample(20.0)) == pytest.approx((0.0, 0.0)) + + +def test_child_timing_ignores_a_child_with_no_slot() -> None: + # Degraded mode: report nothing rather than raise. + shm = get_context("fork").RawArray("d", SLOT_WIDTH) + writer = ChildTimeWriter(shm, NO_SLOT) + reader = ChildTimeAccounting(shm=shm, slot=NO_SLOT) + + writer.mark_running(0.0) + writer.mark_busy(1.0) + reader.mark_running(0.0) + + assert _bw(reader.sample(10.0)) == pytest.approx((0.0, 0.0)) + assert shm[SLOT_SEGMENT_KIND] == KIND_NONE + + +def test_child_timing_excludes_time_banked_before_the_parent_saw_running() -> None: + # Numerator and denominator must start together, at the `running` message. + writer, reader = _writer_and_reader() + + writer.mark_running(0.0) + reader.mark_running(5.0) # parent drained the message 5s later + + assert _bw(reader.sample(6.0)) == pytest.approx((0.0, 1.0)) + + +def test_acquire_timing_slot_zeroes_a_recycled_slot() -> None: + # A replacement must not inherit its predecessor's totals. + pool = _make_result_thread_pool(_SendResultCapture(), concurrency=1) + + # Drain the free list: reuse is FIFO, so a release is not reused next. + slot = pool._acquire_timing_slot() + rest = [pool._acquire_timing_slot() for _ in range(slot_count(1) - 1)] + assert NO_SLOT not in rest + + writer = ChildTimeWriter(pool._timing_shm, slot) + writer.mark_running(0.0) + writer.mark_busy(0.0) + writer.mark_idle(30.0) + + pool._release_timing_slot(slot) + recycled = pool._acquire_timing_slot() + assert recycled == slot + + reader = ChildTimeAccounting(shm=pool._timing_shm, slot=recycled) + reader.mark_running(0.0) + assert _bw(reader.sample(1.0)) == pytest.approx((0.0, 0.0)) + + +def test_acquire_timing_slot_reports_exhaustion_instead_of_raising() -> None: + # Should be unreachable; the pool has to keep spawning either way. + pool = _make_result_thread_pool(_SendResultCapture(), concurrency=1) + pool._metrics = mock.Mock() + + taken = [pool._acquire_timing_slot() for _ in range(slot_count(1))] + assert NO_SLOT not in taken + + assert pool._acquire_timing_slot() == NO_SLOT + assert len(_incr_calls(pool._metrics, "taskworker.worker.child.timing_slot_exhausted")) == 1 + + +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) + pool._metrics = mock.Mock() + pool._last_occupancy_flush_at = 10.0 + + with pool._children_lock: + pool._children[uuid4()] = _make_tracked_child("running", busy_since=10.0) + pool._children[uuid4()] = _make_tracked_child( + "running", busy_accumulated=0.75, wait_since=10.75 + ) + + with mock.patch("taskbroker_client.worker.worker.time.monotonic", return_value=11.0): + pool._emit_periodic_metrics() + + busy = _distribution_calls(pool._metrics, "taskworker.worker.child_busy_seconds") + wait = _distribution_calls(pool._metrics, "taskworker.worker.child_wait_seconds") + assert len(busy) == 1 and len(wait) == 1 + assert busy[0].args[1] == pytest.approx(1.75) + assert wait[0].args[1] == pytest.approx(0.25) + + # The scaler divides the pair, needing neither interval nor child count. + assert busy[0].args[1] / (busy[0].args[1] + wait[0].args[1]) == pytest.approx(0.875) + occupancy_calls = _gauge_calls(pool._metrics, "taskworker.worker.occupancy") + assert occupancy_calls[0].args[1] == pytest.approx(1.75 / 2) + + +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._metrics = mock.Mock() + + with pool._children_lock: + pool._children[uuid4()] = _make_tracked_child("pending") + + pool._emit_periodic_metrics() + + assert _gauge_calls(pool._metrics, "taskworker.worker.occupancy") == [] + assert len(_distribution_calls(pool._metrics, "taskworker.worker.child_busy_seconds")) == 1 + assert len(_distribution_calls(pool._metrics, "taskworker.worker.child_wait_seconds")) == 1 + + +def test_spawn_children_reads_transitions_the_child_wrote() -> None: + # End to end through the real handoff, with no message crossing the queue. fake_context = _FakeContext() pool = _make_fake_context_pool(fake_context, concurrency=1) @@ -1397,21 +1854,23 @@ def test_spawn_children_tracks_busy_and_idle_transitions() -> None: try: _wait_for(lambda: len(fake_context.processes) == 1) messages = fake_context.queues[-1] - child_id = fake_context.processes[0].args[0] + process = fake_context.processes[0] + child_id = process.args[0] + writer = ChildTimeWriter(process.args[-2], process.args[-1]) + writer.mark_running(0.0) messages.put(ChildMessage(child_id, "running")) _wait_for(lambda: pool.ready_count == 1) - # "busy" opens a segment. - messages.put(ChildMessage(child_id, "busy")) - _wait_for(lambda: pool._children[child_id].busy_since is not None) + child = pool._children[child_id] + # Rebaseline onto the child's clock; normally done when draining `running`. + child.timing.mark_running(0.0) - # "idle" closes it and banks a positive amount of busy-time. - messages.put(ChildMessage(child_id, "idle")) - _wait_for( - lambda: pool._children[child_id].busy_since is None - and pool._children[child_id].busy_accumulated > 0 - ) + # 2s waiting, 1s of work, then waiting again. + writer.mark_busy(2.0) + writer.mark_idle(3.0) + + assert _bw(child.timing.sample(4.0)) == pytest.approx((1.0, 3.0)) finally: pool.shutdown() @@ -1441,7 +1900,7 @@ def test_spawn_children_releases_draining_child_above_min_concurrency() -> None: messages = fake_context.queues[-1] first_process = fake_context.processes[0] first_child_id = first_process.args[0] - first_release = first_process.args[-1] + first_release = first_process.args[-3] messages.put(ChildMessage(first_child_id, "running")) second_process = fake_context.processes[1] @@ -1467,7 +1926,7 @@ def test_spawn_children_defers_draining_child_at_min_concurrency() -> None: messages = fake_context.queues[-1] first_process = fake_context.processes[0] first_child_id = first_process.args[0] - first_release = first_process.args[-1] + first_release = first_process.args[-3] second_process = fake_context.processes[1] second_child_id = second_process.args[0] @@ -1624,6 +2083,7 @@ def test_child_process_emits_running_message() -> None: ctx = get_context("fork") child_id = uuid4() messages = ctx.Queue() + timing_shm = ctx.RawArray("d", SLOT_WIDTH) parent_release = ctx.Event() parent_release.set() @@ -1641,6 +2101,8 @@ def test_child_process_emits_running_message() -> None: future_checking_frequency=0.1, messages=messages, parent_release=parent_release, + timing_shm=timing_shm, + timing_slot=0, ) # The child signals readiness once warmup is done, before consuming @@ -1658,6 +2120,7 @@ def test_child_process_emits_exiting_once_and_continues_until_release( todo = ctx.Queue() processed = ctx.Queue() messages = ctx.Queue() + timing_shm = ctx.RawArray("d", SLOT_WIDTH) parent_release = ctx.Event() todo.put(SIMPLE_TASK) @@ -1676,25 +2139,19 @@ def test_child_process_emits_exiting_once_and_continues_until_release( 0.1, messages, parent_release, + timing_shm, + 0, ), ) process.start() try: - running_message = messages.get(timeout=5) - busy_message = messages.get(timeout=5) - idle_message = messages.get(timeout=5) - exiting_message = messages.get(timeout=5) - - assert running_message == ChildMessage(child_id, "running") - assert busy_message == ChildMessage(child_id, "busy") - assert idle_message == ChildMessage(child_id, "idle") - assert exiting_message == ChildMessage(child_id, "exiting") + # Lifecycle only now: two per child rather than two per task. + assert messages.get(timeout=5) == ChildMessage(child_id, "running") + assert messages.get(timeout=5) == ChildMessage(child_id, "exiting") assert processed.get(timeout=5).task_id == SIMPLE_TASK.activation.id todo.put(SIMPLE_TASK) assert processed.get(timeout=5).task_id == SIMPLE_TASK.activation.id - assert messages.get(timeout=5) == ChildMessage(child_id, "busy") - assert messages.get(timeout=5) == ChildMessage(child_id, "idle") time.sleep(0.2) assert process.is_alive() @@ -1711,13 +2168,14 @@ def test_child_process_emits_exiting_once_and_continues_until_release( assert mock_capture_checkin.call_count == 0 -def test_child_process_emits_busy_and_idle_messages() -> None: +def test_child_process_records_busy_and_idle_in_its_slot() -> None: todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() ctx = get_context("fork") child_id = uuid4() messages = ctx.Queue() + timing_shm = ctx.RawArray("d", SLOT_WIDTH) parent_release = ctx.Event() parent_release.set() @@ -1735,13 +2193,89 @@ def test_child_process_emits_busy_and_idle_messages() -> None: future_checking_frequency=0.1, messages=messages, parent_release=parent_release, + timing_shm=timing_shm, + timing_slot=0, ) assert messages.get(timeout=1) == ChildMessage(child_id, "running") - assert messages.get(timeout=1) == ChildMessage(child_id, "busy") - assert messages.get(timeout=1) == ChildMessage(child_id, "idle") assert processed.get(timeout=1).task_id == SIMPLE_TASK.activation.id + # One task completed, so a busy segment closed and the wait clock reopened. + assert timing_shm[SLOT_BUSY_TOTAL] > 0.0 + assert timing_shm[SLOT_SEGMENT_KIND] == KIND_WAIT + assert timing_shm[SLOT_VERSION] % 2 == 0 + assert timing_shm[SLOT_SEGMENT_START] > 0.0 + + +def _run_one_task_capturing_metrics(received_at_offset: float) -> mock.Mock: + """Run a single task through a child, with `received_at` set relative to now. + + Returns the mocked metrics backend so callers can assert on what was emitted. + """ + from examples.app import app as example_app + + activation = TaskActivation( + id="queue-wait", + taskname="examples.simple_task", + namespace="examples", + parameters_bytes=msgpack.packb({"args": [], "kwargs": {}}, use_bin_type=True), + processing_deadline_duration=2, + ) + activation.received_at.FromDatetime( + datetime.fromtimestamp(time.time() + received_at_offset, tz=timezone.utc) + ) + + todo: queue.Queue[InflightTaskActivation] = queue.Queue() + processed: queue.Queue[ProcessingResult] = queue.Queue() + todo.put( + InflightTaskActivation(host="localhost:50051", receive_timestamp=0, activation=activation) + ) + + # MagicMock: the child uses metrics.timer() as a context manager. + metrics = mock.MagicMock() + with mock.patch.object(example_app, "metrics", metrics): + child_process( + "examples.app:app", + todo, + processed, + Event(), + 1, + "test", + "fork", + False, + 0.1, + ) + + assert processed.get(timeout=1).task_id == "queue-wait" + return metrics + + +def test_child_process_emits_queue_wait_excluding_execution_time() -> None: + # Latency minus the task's own cost, so one threshold covers every pool. + metrics = _run_one_task_capturing_metrics(received_at_offset=-2.0) + + wait = _distribution_calls(metrics, "taskworker.worker.queue_wait") + latency = _distribution_calls(metrics, "taskworker.worker.execution_latency") + duration = _distribution_calls(metrics, "taskworker.worker.execution_duration") + assert len(wait) == 1 and len(latency) == 1 and len(duration) == 1 + + # The activation was stamped 2s ago and picked up immediately. + assert wait[0].args[1] == pytest.approx(2.0, abs=0.5) + # And it partitions the end-to-end latency with the execution itself. + assert wait[0].args[1] + duration[0].args[1] == pytest.approx(latency[0].args[1], abs=0.01) + + assert wait[0].kwargs["tags"]["processing_pool"] == "test" + assert wait[0].kwargs["tags"]["taskname"] == "examples.simple_task" + + +def test_child_process_queue_wait_clamps_negative_clock_skew() -> None: + # An NTP-skewed pod would otherwise emit a negative sample. + metrics = _run_one_task_capturing_metrics(received_at_offset=+5.0) + + wait = _distribution_calls(metrics, "taskworker.worker.queue_wait") + assert len(wait) == 1 + assert wait[0].args[1] == 0.0 + def test_child_process_remove_start_time_kwargs() -> None: activation = InflightTaskActivation(