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

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, 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:

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

The LightningStoreThreaded class, defined in 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 (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 (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. 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:

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, 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.

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, 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.

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