How SlotWorker Handles Priority Queuing and Auto‑Preemption in Forge

SlotWorker serializes access to a WorkflowRunner by combining an asyncio.PriorityQueue with a lightweight cancellation protocol that automatically aborts lower‑priority tasks when urgent work arrives.

In the antoinezambelli/forge repository, the SlotWorker class manages concurrent workflow execution by ensuring that higher‑priority tasks always take precedence, even if that means preempting work already in flight. This design guarantees that critical operations are not blocked by background jobs while maintaining FIFO ordering among tasks of equal priority.

The Priority Queue Architecture

The core of SlotWorker is an asyncio.PriorityQueue that stores pending workflow tasks. Because Python’s PriorityQueue orders elements by the first item of a tuple, lower integer values represent higher priority.

Queue Entry Structure

Each item placed into the queue is a six‑element tuple defined in the class constructor at src/forge/core/slot_worker.py (lines 34‑36):

(priority, self._counter, workflow, user_message, prompt_vars, future)

• priority: An integer where smaller values indicate higher urgency.
• self._counter: A monotonically increasing integer that breaks ties.
• workflow, user_message, prompt_vars: Execution context for the runner.
• future: An asyncio.Future that will hold the result or exception.

FIFO Guarantee for Equal Priorities

The second element of the tuple—self._counter—ensures that tasks sharing the same priority value are executed in the order they were submitted. This counter increments with every call to submit(), preventing starvation of older tasks when the queue contains multiple entries with identical priorities.

Auto‑Preemption Mechanics

Auto‑preemption allows SlotWorker to interrupt a running workflow the moment a higher‑priority task enters the system. This is implemented entirely through Python’s asyncio.Event mechanism rather than process‑level signals.

The Preemption Trigger in submit()

Immediately after enqueueing a new task, submit() compares the incoming priority against the currently executing task. If the new task is strictly more important, it signals cancellation (lines 88‑96 of src/forge/core/slot_worker.py):

if (
    self._current_priority is not None
    and priority < self._current_priority
    and self._cancel_event is not None
):
    self._cancel_event.set()

This logic executes synchronously during the submit() call, meaning preemption happens instantly—even before the submitting coroutine yields control.

Cancellation Event Propagation

The _worker() loop (lines 25‑33) creates a fresh asyncio.Event for each task it processes:

self._cancel_event = asyncio.Event()
self._current_priority = priority
result = await self.runner.run(..., cancel_event=self._cancel_event)

When submit() sets _cancel_event, the running WorkflowRunner detects the signal via its cancel_event argument and raises a cancellation exception. The worker catches this, clears its internal state, and immediately begins processing the next highest‑priority item from the queue.

The Worker Processing Loop

The _worker() coroutine runs continuously in the background after start() is called. Its lifecycle follows this pattern:

1. Dequeue: Pull the next tuple from self._queue.get().
2. Prepare: Instantiate a new asyncio.Event() and record the task’s priority in self._current_priority.
3. Execute: Invoke self.runner.run() with the cancellation event.
4. Finalize: Set the task’s future with either the result or the exception, then clear self._current_priority and self._cancel_event.

If the task completes normally, its associated future receives the return value. If it is preempted or fails, the future receives the corresponding exception, allowing the caller to handle the outcome via await worker.submit().

Complete Working Example

The following example demonstrates how a task with priority=1 preempts a running task with priority=5:

import asyncio
from forge.core.runner import WorkflowRunner
from forge.core.slot_worker import SlotWorker
from forge.core.workflow import Workflow

async def main():
    runner = WorkflowRunner(...)
    worker = SlotWorker(runner)

    await worker.start()

    # Low‑priority task

    low_prio = asyncio.create_task(
        worker.submit(Workflow(...), "low‑prio request", priority=5)
    )

    # High‑priority task arrives shortly after

    await asyncio.sleep(0.1)
    high_prio = asyncio.create_task(
        worker.submit(Workflow(...), "high‑prio request", priority=1)
    )

    # The high‑priority task returns successfully

    result = await high_prio
    print("High‑priority result:", result)

    # The low‑priority task raises WorkflowCancelledError

    try:
        await low_prio
    except Exception as exc:
        print("Low‑priority task was pre‑empted:", exc)

    await worker.stop()

asyncio.run(main())

In this scenario, the high‑priority submission triggers self._cancel_event.set(), causing the runner executing the low‑priority workflow to abort and raise an exception that propagates back to the original caller.

Summary

• SlotWorker uses an asyncio.PriorityQueue to order tasks, with lower integers indicating higher priority.
• A monotonic counter ensures FIFO execution among tasks sharing the same priority, preventing starvation.
• Auto‑preemption is triggered inside submit() when a new task’s priority is strictly lower (higher urgency) than _current_priority.
• The cancellation event is passed to WorkflowRunner.run(), allowing graceful abortion of in‑flight work.
• Preempted tasks receive a cancellation exception through their associated future, enabling callers to distinguish between success and preemption.

Frequently Asked Questions

What exception does a preempted task receive?

A preempted task typically receives a WorkflowCancelledError (or similar cancellation exception defined in the forge framework) through its future. This allows the caller to catch the exception and handle cleanup logic or retry the task at a lower priority later.

How does the counter prevent starvation of equal‑priority tasks?

The self._counter value increments monotonically for every submission and serves as the second sort key in the priority queue tuple. Because asyncio.PriorityQueue falls back to the next tuple element when priorities are equal, tasks with identical priority values are dequeued in the exact order they were enqueued, guaranteeing fairness.

Can auto‑preemption be disabled?

The current implementation in src/forge/core/slot_worker.py does not provide a configuration flag to disable auto‑preemption. Preemption is intrinsic to the submit() method’s logic (lines 88‑96). To disable it, you would need to subclass SlotWorker and override submit() to remove the cancellation check.

Which class actually performs the cancellation check?

While SlotWorker creates and sets the asyncio.Event, the actual responsiveness to cancellation depends on WorkflowRunner in src/forge/core/runner.py. The runner must accept the cancel_event parameter and periodically check event.is_set() during long‑running operations to enable cooperative cancellation.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}