Skip to content

fix(taskworker): Widen taskworker busy segment to cover the full task cycle #780

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

Merged
enochtangg merged 1 commit into from
Aug 12, 2026
Merged

fix(taskworker): Widen taskworker busy segment to cover the full task cycle #780

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
23 changes: 20 additions & 3 deletions clients/python/src/taskbroker_client/worker/workerchild.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,6 +239,7 @@ def run_worker(
) -> None:
processed_task_count = 0
pending_task_futures: list[ActivationWithPendingFutures] = []
is_busy = False

def handle_alarm(signum: int, frame: FrameType | None) -> None:
"""
Expand Down Expand Up @@ -372,6 +373,13 @@ def check_task_future_completion(
exit_initiated: float | None = None

while not shutdown_event.is_set() and not local_shutdown.is_set():
# Close the busy segment opened by the previous iteration. Everything
# 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"))
is_busy = False

if max_task_count and processed_task_count >= max_task_count:
if exit_initiated is None:
metrics.incr(
Expand Down Expand Up @@ -409,6 +417,11 @@ def check_task_future_completion(
)
continue

# 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"))
is_busy = True

task_func = _get_known_task(inflight.activation)
if not task_func:
metrics.incr(
Expand Down Expand Up @@ -465,7 +478,6 @@ def check_task_future_completion(
next_state = TASK_ACTIVATION_STATUS_FAILURE
# Use time.time() so we can measure against activation.received_at
execution_start_time = time.time()
messages.put_nowait(ChildMessage(child_id, "busy"))
try:
with timeout_alarm(inflight.activation.processing_deadline_duration, handle_alarm):
_execute_activation(task_func, inflight.activation, app.context_hooks)
Expand Down Expand Up @@ -539,8 +551,6 @@ def check_task_future_completion(
and next_state != TASK_ACTIVATION_STATUS_RETRY
):
_log_task_failed(inflight.activation, err, processing_pool_name)
finally:
messages.put_nowait(ChildMessage(child_id, "idle"))

clear_current_task()
processed_task_count += 1
Expand Down Expand Up @@ -593,6 +603,13 @@ def check_task_future_completion(
)
pending_task_futures.append(pending_task)

# The loop condition is only re-checked between iterations, so a shutdown
# 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"))
is_busy = False

# Once we get the shutdown signal, drain any pending futures
_future_completion_thread.join()
for task in pending_task_futures.copy():
Expand Down
Loading