Concurrency & Locking

Relevant source files

The following files were used as context for generating this wiki page:

• bubus/event_bus.py
• bubus/lock_manager.py
• bubus-ts/src/lock_manager.ts
• tests/test_semaphores.py
• bubus-ts/examples/concurrency_options.ts
• TODO_python.md

Concurrency and locking in the bubus ecosystem provide a multi-layered mechanism to control event execution flow and handler parallelism. The system is designed to ensure FIFO (First-In-First-Out) processing while allowing for sophisticated execution policies such as global serialization, bus-local serialization, and parallel handler execution. It employs semaphores and re-entrant locks to manage resource access across both local threads and distributed processes.

Sources: bubus/event_bus.py:108-115, bubus-ts/src/lock_manager.ts:50-55, TODO_python.md:14-25

Core Concurrency Architecture

The architecture distinguishes between Event Concurrency (how many events a bus processes at once) and Handler Concurrency (how many handlers for a single event run at once). This management is centralized in the LockManager class, which interacts with the EventBus to enforce these limits.

The diagram above illustrates the relationship between the EventBus, its LockManager, and the various semaphore scopes used to gate execution.

Sources: bubus/event_bus.py:133-145, bubus-ts/src/lock_manager.ts:89-105

Concurrency Modes

The system supports several modes that can be configured at the EventBus level or overridden per BaseEvent.

Mode Category	Option	Description
Event Concurrency	global-serial	Only one event is processed across all bus instances sharing the same lock.
	bus-serial	The bus processes one event at a time from its queue.
	parallel	Multiple events from the queue are processed simultaneously.
Handler Concurrency	serial	Handlers for a single event run one after another in registration order.
	parallel	All handlers for a single event are dispatched concurrently.

Sources: bubus/event_bus.py:118-124, bubus-ts/src/lock_manager.ts:20-30, bubus-ts/examples/concurrency_options.ts:23-50

The Lock Manager

The LockManager (available in both Python and TypeScript) is responsible for yielding semaphores based on the event's policy. It tracks active handler results to facilitate "inside handler" detection, which is critical for queue-jumping and parent-child event tracking.

Synchronization Components

1. AsyncSemaphore: A standard asynchronous semaphore used to limit concurrency to a specific size. In TypeScript, it manages a waiters queue of resolvers.
2. ReentrantLock: Used primarily in Python to prevent deadlocks during nested execution where the same task might attempt to acquire the same lock multiple times.
3. HandlerLock: Tracks a single handler's ownership of a semaphore. It provides methods like yieldHandlerLockForChildRun() and reclaimHandlerLockIfRunning() to support queue-jumping semantics.

Sources: bubus/lock_manager.py:10-40, bubus-ts/src/lock_manager.ts:32-48, bubus-ts/src/lock_manager.ts:108-140

Queue-Jumping and Re-entrancy

Queue-jumping occurs when a handler awaits the completion of a child event it just dispatched. To prevent deadlocks in serial modes, the parent handler must yield its lock to allow the child event to process immediately on the same bus.

The sequence above shows the coordination required to allow a child event to "jump the queue" while the parent handler is still technically active.

Sources: bubus-ts/src/lock_manager.ts:114-135, TODO_python.md:55-65

Multiprocess Semaphores

For scenarios involving multiple Python processes, the @retry decorator provides semaphore_scope='multiprocess'. This implementation uses file-based locking (via portalocker) in a designated directory to ensure limits are respected across OS process boundaries.

• Lax Mode: If semaphore_lax=True, the function proceeds even if the semaphore acquisition times out.
• Death Handling: The system is designed to handle "process death" where a process dies without releasing its slot. Subsequent attempts or cleanup logic detect stale locks.

Sources: tests/test_semaphores.py:15-50, tests/test_semaphores.py:205-240

Summary

The concurrency and locking system in bubus provides a flexible yet robust framework for managing asynchronous execution. By combining bus-level defaults with per-event overrides and specialized handlers for queue-jumping and multiprocess environments, the system ensures data integrity and FIFO ordering without sacrificing the performance benefits of parallel processing.

Sources: bubus/event_bus.py:1120-1150, bubus-ts/examples/concurrency_options.ts:145-160