# How SlotWorker Handles Priority Queuing and Auto‑Preemption in Forge

> Discover how SlotWorker uses asyncio PriorityQueue and auto-preemption to manage urgent tasks, ensuring efficient workflow execution in Forge.

- Repository: [Antoine/forge](https://github.com/antoinezambelli/forge)
- Tags: internals
- Published: 2026-05-22

---

**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`](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/slot_worker.py) (lines 34‑36):

```python
(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`](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/slot_worker.py)):

```python
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:

```python
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`:

```python
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`](https://github.com/antoinezambelli/forge/blob/main/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`](https://github.com/antoinezambelli/forge/blob/main/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.