# Ensuring Thread-Safety and Async Safety in Distributed Training with Agent-Lightning

> Learn how Agent-Lightning ensures thread-safety and async safety in distributed training using LightningStoreThreaded, cooperative cancellation, and process-isolated Ray actors.

- Repository: [Microsoft/agent-lightning](https://github.com/microsoft/agent-lightning)
- Tags: best-practices
- Published: 2026-04-01

---

**Agent-Lightning guarantees safe concurrent execution through `LightningStoreThreaded` for thread-safe shared state, cooperative cancellation via `asyncio.Task` watchers, and process-isolated Ray actors for distributed rollouts.**

Agent-Lightning is a Microsoft open-source framework designed for distributed reinforcement learning training that separates algorithm logic from environment interaction workers. When scaling to multi-threaded and asynchronous execution patterns, ensuring **thread-safety and async safety in distributed training** becomes critical to prevent data corruption and event-loop blocking. The framework addresses these concerns through a layered architecture involving execution strategies, lock-protected stores, and cooperative shutdown mechanisms.

## Thread-Safe Shared State with LightningStoreThreaded

When using the **SharedMemoryExecutionStrategy**, Agent-Lightning automatically wraps the data store to serialize concurrent access. In [`agentlightning/execution/shared_memory.py`](https://github.com/microsoft/agent-lightning/blob/main/agentlightning/execution/shared_memory.py), the strategy checks the `managed_store` flag (controlled by the environment variable `AGL_MANAGED_STORE` and defaulting to `True`) to determine whether to apply thread-safety wrapping:

```python
if self.managed_store:
    thread_safe_store = LightningStoreThreaded(store)
else:
    thread_safe_store = store

```

The `LightningStoreThreaded` class, defined in [`agentlightning/store/threading.py`](https://github.com/microsoft/agent-lightning/blob/main/agentlightning/store/threading.py), protects every public method—including `get`, `set`, and `append`—with a `threading.RLock`. This ensures that when multiple runner threads read rollout data or the algorithm thread writes updated weights, the underlying data structures remain consistent. Because the lock is reentrant, nested calls within the same thread do not cause deadlocks.

## Async-Safe Coroutine Execution

Agent-Lightning prevents event-loop blocking through the `_run_until_completed_or_canceled` helper in [`agentlightning/execution/shared_memory.py`](https://github.com/microsoft/agent-lightning/blob/main/agentlightning/execution/shared_memory.py) (lines 70-124). This function manages **async safety** by wrapping coroutine bundles in `asyncio.Task` objects and watching for shutdown signals without consuming the event loop.

The mechanism works in four steps:

1. **Task Creation** – The coroutine is scheduled via `asyncio.create_task(coro)`.
2. **Watcher Polling** – A separate watcher coroutine polls the `stop_evt` (a `threading.Event`) every `poll_interval` seconds using `asyncio.sleep`, never calling the blocking `Event.wait()` method.
3. **Graceful Cancellation** – When `stop_evt` is set, the watcher grants the bundle `graceful_delay` seconds to complete naturally. If the bundle remains active, the watcher calls `task.cancel()` to trigger a `CancelledError`.
4. **Exception Propagation** – Both the main task and watcher are awaited with `asyncio.wait(FIRST_COMPLETED)`, ensuring that exceptions bubble up to the caller for proper handling.

This design keeps the event loop responsive while allowing cooperative cancellation—bundles should periodically check `event.is_set()` to exit cleanly before the hard cancellation deadline.

## Coordinated Shutdown and Error Handling

The framework uses a single shared `threading.Event` instance (`stop_evt`) to coordinate stopping across all threads. In [`agentlightning/execution/shared_memory.py`](https://github.com/microsoft/agent-lightning/blob/main/agentlightning/execution/shared_memory.py) (lines 158-182), the execution strategy handles three shutdown scenarios:

- **KeyboardInterrupt** – The main thread catches `Ctrl+C` and immediately sets `stop_evt`, triggering the graceful shutdown sequence in all runner threads.
- **Bundle Exceptions** – If any algorithm or runner bundle raises an exception, the error is placed onto a `SimpleQueue` named `thread_exceptions`, `stop_evt` is set, and all other threads observe the stop condition.
- **Timeout Joins** – After signaling stop, the strategy joins each background thread with a configurable `join_timeout`. Stragglers that exceed this timeout are logged but left as daemon threads, preventing process hangs while allowing logging for debugging.

## Distributed Rollout Isolation with Async vLLM Server

For distributed training, Agent-Lightning offloads heavy inference to a **Ray-remote async server** (`PatchedvLLMServer`) located in [`agentlightning/verl/async_server.py`](https://github.com/microsoft/agent-lightning/blob/main/agentlightning/verl/async_server.py). Running as a separate process (`@ray.remote(num_cpus=1)`), this server provides complete memory isolation from the training loop's threads.

Key async-safety features include:

- **Non-blocking Endpoints** – The `chat_completion` method parses requests with `await raw_request.json()` and returns `StreamingResponse` objects when `request.stream` is true, maintaining async compatibility.
- **Process Boundaries** – Because the server exists in its own Ray process, it cannot block the main training loop's event loop or corrupt the shared-memory store.

The main training process communicates with this server through the `AgentModeDaemon`, which forwards data asynchronously without shared-memory contention.

## Practical Configuration Example

To configure a training run with guaranteed thread-safety and async safety, instantiate the `Trainer` with explicit strategy parameters:

```python
from agentlightning.trainer import Trainer
from agentlightning.execution.shared_memory import SharedMemoryExecutionStrategy
from agentlightning.store import InMemoryStore

# Create the underlying store

store = InMemoryStore()

# Configure the execution strategy with thread/async safety

strategy = SharedMemoryExecutionStrategy(
    n_runners=4,               # Number of rollout worker threads

    main_thread="algorithm",   # Run algorithm on main thread, runners in background

    graceful_delay=2.0,        # Seconds to wait before forcing cancellation

)

# Build and run the trainer

trainer = Trainer(
    store=store,
    strategy=strategy,
    # Additional config: model, tokenizer, reward function, etc.

)

trainer.fit()

```

In this configuration, the `store` object is automatically wrapped in `LightningStoreThreaded`, four runner threads execute rollouts concurrently, and pressing **Ctrl+C** triggers the cooperative shutdown sequence across all components.

## Summary

- **Thread-safety** is enforced by `LightningStoreThreaded` using an `RLock` to serialize access to shared training data across multiple runner threads.
- **Async-safety** is maintained through the watcher-pattern in `_run_until_completed_or_canceled`, which avoids blocking the event loop and implements graceful, cooperative cancellation.
- **Process isolation** for distributed rollouts is achieved via the Ray-remote `PatchedvLLMServer`, preventing memory corruption and event-loop interference from heavy inference workloads.
- **Coordinated shutdown** uses a shared `ThreadingEvent` to propagate stop signals and exceptions across threads, with configurable timeouts for clean process termination.

## Frequently Asked Questions

### How does Agent-Lightning prevent race conditions when multiple runners access shared data?

Agent-Lightning prevents race conditions by automatically wrapping the `LightningStore` in `LightningStoreThreaded` when using `SharedMemoryExecutionStrategy`. This wrapper acquires a `threading.RLock` around every read and write operation, ensuring that only one thread can modify the store at a time. According to the implementation in [`agentlightning/store/threading.py`](https://github.com/microsoft/agent-lightning/blob/main/agentlightning/store/threading.py), this applies to all public methods including `get`, `set`, and `append`, making concurrent access from multiple runners safe by default.

### What happens when I press Ctrl+C during training?

When you send a KeyboardInterrupt (Ctrl+C), the main thread catches the exception and sets the shared `stop_evt` (a `threading.Event` instance). This signals all background threads to initiate shutdown. Each bundle receives a `graceful_delay` period to complete its current work, after which the watcher coroutine cancels the async task if it hasn't finished. The strategy then joins all threads with a timeout before allowing the process to exit, as implemented in the `execute()` method of [`agentlightning/execution/shared_memory.py`](https://github.com/microsoft/agent-lightning/blob/main/agentlightning/execution/shared_memory.py).

### Can I use Agent-Lightning with async coroutines for the algorithm bundle?

Yes, Agent-Lightning fully supports async algorithm bundles through the `_run_until_completed_or_canceled` mechanism in `SharedMemoryExecutionStrategy`. The strategy creates an `asyncio.Task` from your coroutine and monitors it with a non-blocking watcher that polls the stop event. This ensures that async algorithms—such as those using Ray for distributed communication—can run without blocking the event loop, while still responding to shutdown signals and propagating exceptions correctly.

### Why does the vLLM server run as a separate Ray actor instead of a thread?

The `PatchedvLLMServer` runs as a Ray actor (`@ray.remote`) in its own process to provide complete isolation from the main training loop's memory and event loop. Heavy LLM inference operations in vLLM can block threads and consume significant memory; by placing these in a separate process via [`agentlightning/verl/async_server.py`](https://github.com/microsoft/agent-lightning/blob/main/agentlightning/verl/async_server.py), Agent-Lightning ensures that inference latency or crashes cannot corrupt the shared training state or stall the async event loop. Communication happens through Ray's async-compatible remote methods, preserving both thread-safety and async-safety.