# How to Handle Concurrent Event Dispatch Without Race Conditions in bubus

> Safely handle concurrent event dispatch in bubus. Learn how bubus uses a global lock and custom asyncio queue to prevent race conditions while enabling parallel handler execution for robust event processing.

- Repository: [Browser Use/bubus](https://github.com/browser-use/bubus)
- Tags: how-to-guide
- Published: 2026-02-26

---

**bubus prevents race conditions during concurrent event dispatch by using a single global re-entrant lock combined with a custom asyncio queue that serializes event processing while allowing parallel handler execution.**

When building event-driven architectures with browser-use/bubus, handling concurrent event dispatch without race conditions is critical for system stability. The library implements a sophisticated concurrency model centered in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py) that ensures thread-safe event processing even when multiple coroutines call `dispatch()` simultaneously.

## Core Concurrency Architecture

### The Global Re-entrant Lock

At the heart of bubus concurrency safety is a **global re-entrant lock** implemented via the `ReentrantLock` class (lines 47-84 in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py)). This lock is acquired at the start of every event processing cycle within the `_run_loop` method (lines 45-52), specifically during the `step()` → `process_event()` chain.

The lock is **re-entrant across different asyncio tasks**, meaning a handler can safely dispatch new events without deadlocking itself. This is crucial for complex event chains where handlers need to trigger subsequent events.

### The Custom Asyncio Queue

bubus uses a `CleanShutdownQueue` (lines 56-84 in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py)) instead of a standard `asyncio.Queue`. This custom queue overrides `put()` and `get()` to raise a dedicated `QueueShutDown` exception during graceful shutdown, allowing the run-loop to exit cleanly without "queue closed" warnings while ensuring all pending events are processed before shutdown completes.

## Race Condition Prevention Mechanisms

### Serializing Event Processing

The `_run_loop` task repeatedly calls `step()`, which acquires the global lock before processing any event. While the lock is held, `process_event()` gathers all applicable handlers and creates pending `EventResult` objects. This serialization ensures that `event.results` is never written to by multiple handlers simultaneously, even when `parallel_handlers=True` allows handlers to execute concurrently.

### Preventing Infinite Forwarding Loops

When handlers forward events to other buses, the `_would_create_loop()` method (lines 332-348 in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py)) checks the current event's `event_path` attribute. If the target bus already appears in the path, the forward is skipped, preventing cycles like A→B→A that could cause race conditions or infinite recursion.

### Blocking Recursive Handler Dispatches

The `_handler_dispatched_ancestor()` method (lines 889-901 in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py)) walks the event ancestry chain and aborts if recursion depth exceeds two levels. If a handler attempts to recursively dispatch the same event type that triggered it, bubus raises a clear `RuntimeError`, preventing stack overflow and state corruption.

### Deadlock Detection for Long-Running Handlers

Each handler execution spawns a `deadlock_monitor` task (lines 1100-1116 in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py)) that logs a warning after 15 seconds. This helps identify handlers that block the global lock for too long, allowing developers to refactor blocking code before it causes system-wide latency.

### Context Tracking for Child Events

Context variables (`_current_event_context`, `inside_handler_context`, `_current_handler_id_context`) defined in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py) (lines 37-45) track the parent event and handler ID. When handlers create child events, these contexts ensure proper parent-child linking and automatic cleanup when the parent completes, preventing orphaned events and memory leaks.

## Practical Code Examples

### Concurrent Event Dispatch

```python
import asyncio
from bubus import EventBus, BaseEvent

class PingEvent(BaseEvent):
    event_type = "Ping"

bus = EventBus(parallel_handlers=True)          # handlers run concurrently

@bus.on("*")
async def logger(event: BaseEvent):
    print(f"🪵 Received {event}")

@bus.on(PingEvent)
async def handle_ping(event: PingEvent):
    await asyncio.sleep(0.1)                    # simulate work

    return "pong"

async def fire():
    # fire 10 events at once – they are queued safely

    tasks = [bus.dispatch(PingEvent()) for _ in range(10)]
    results = await asyncio.gather(*tasks)      # each result is a completed PingEvent

    print([r.event_result() for r in results]) # ["pong", …]

asyncio.run(fire())

```

*Why it’s race-free:* all `dispatch()` calls share the same `CleanShutdownQueue`; the run-loop processes them one-by-one while holding the global lock, so `event.event_results` is never written concurrently.

### Safe Bus Forwarding

```python
bus_a = EventBus(name="A")
bus_b = EventBus(name="B")

@bus_a.on("*")
async def forward_to_b(event):
    # safe forwarding – the path check in _would_create_loop() prevents A→B→A cycles

    await bus_b.dispatch(event)

@bus_b.on("*")
async def log_b(event):
    print(f"B got {event}")

await bus_a.dispatch(PingEvent())

```

If `bus_b` later tried to forward the same event back to `bus_a`, `_would_create_loop()` would detect `bus_a` already in `event_path` and skip the handler, avoiding an infinite loop.

### Waiting for Idle State

```python
await bus.wait_until_idle(timeout=5)   # blocks until the queue is empty and all handlers finished

```

The idle waiter yields to the event loop (`await asyncio.sleep(0)`) so any *child* events created during a handler are processed before the function returns, guaranteeing a clean, race-free state.

## Key Implementation Files

| File | What it Provides (relevant sections) |
|------|--------------------------------------|
| [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py) | Core `EventBus` implementation, `CleanShutdownQueue`, `ReentrantLock`, global lock handling, event dispatch, loop, race-prevention helpers (`_would_create_loop`, `_handler_dispatched_ancestor`). |
| [`bubus/models.py`](https://github.com/browser-use/bubus/blob/main/bubus/models.py) | `BaseEvent` and `EventResult` definitions, context-variable tracking, completion signalling, child-event hierarchy. |
| [`bubus/helpers.py`](https://github.com/browser-use/bubus/blob/main/bubus/helpers.py) | Utility helpers (logging, pretty-printing) used by the bus – useful for debugging concurrent scenarios. |

These files together provide the complete concurrency safety model in bubus.

## Summary

- **Global serialization**: A single re-entrant lock in `_run_loop` ensures only one event is processed at a time, preventing concurrent writes to shared state.
- **Safe forwarding**: The `_would_create_loop()` check prevents infinite forwarding cycles between buses that could cause deadlocks.
- **Recursion guards**: `_handler_dispatched_ancestor()` blocks recursive event dispatches that would otherwise corrupt the event stack.
- **Graceful shutdown**: `CleanShutdownQueue` allows the bus to finish processing queued events before shutting down, preventing dropped events.
- **Parallel execution**: While event processing is serialized, handlers can run in parallel when `parallel_handlers=True`, with results safely merged under the global lock.

## Frequently Asked Questions

### How does bubus prevent race conditions when multiple tasks dispatch events simultaneously?

bubus serializes all event processing through a global re-entrant lock acquired in the `_run_loop` method of [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py). While multiple coroutines can call `dispatch()` concurrently, adding events to the `CleanShutdownQueue`, the run-loop processes them one-by-one under the lock. This ensures that `event.results` and other shared state are never modified by multiple handlers simultaneously, even when `parallel_handlers=True` allows concurrent handler execution.

### What prevents infinite loops when forwarding events between multiple buses?

The `_would_create_loop()` method in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py) (lines 332-348) tracks the `event_path` attribute of each event. Before forwarding an event to another bus, it checks if the target bus name already exists in the path. If a cycle is detected (e.g., Bus A → Bus B → Bus A), the forward is skipped, preventing infinite recursion and the race conditions that would occur from unbounded event chaining.

### How does bubus handle graceful shutdown while events are still processing?

bubus uses a `CleanShutdownQueue` class (lines 56-84 in [`bubus/service.py`](https://github.com/browser-use/bubus/blob/main/bubus/service.py)) that overrides standard `asyncio.Queue` methods. When shutdown is initiated, the queue raises a dedicated `QueueShutDown` exception instead of the standard "queue closed" warning, allowing the `_run_loop` to exit cleanly. The bus waits for all currently executing handlers to complete and for child events to finish before shutting down, ensuring no events are dropped or left in an inconsistent state.

### Can handlers run in parallel, and how is thread safety maintained?

Yes, handlers can run in parallel when the `EventBus` is initialized with `parallel_handlers=True`. However, thread safety is maintained because the global lock in `_run_loop` serializes the *dispatch* and *collection* phases—gathering handlers and merging results—while only the handler *execution* runs concurrently. The `event_completed_signal` is set only when all handlers and child events finish, and context variables track parent-child relationships, ensuring that parallel execution never corrupts shared event state.