# How the Speech-to-Speech Server Routes WebSocket Connections to Pipeline Instances in the Pool

> Learn how the speech-to-speech server routes WebSocket connections to available pipeline instances. Discover the efficient unit claiming and session binding process.

- Repository: [Hugging Face/speech-to-speech](https://github.com/huggingface/speech-to-speech)
- Tags: internals
- Published: 2026-08-08

---

**The server maintains a pool of `PipelineUnit` objects and routes each incoming WebSocket connection to the first available unit by atomically claiming it through `_claim_unit`, binding a `WebSocketTransport` to the unit's session state, and dedicating an async send loop to that specific client until the session ends.**

The huggingface/speech-to-speech repository implements a real-time inference API that handles concurrent speech-to-speech conversations through isolated pipeline instances. When a client connects to the `/v1/realtime` WebSocket endpoint, the system must map that connection to a specific STT→LM→TTS pipeline while ensuring exclusive access and clean resource isolation. This routing mechanism relies on a deterministic pool-based architecture with atomic claiming and per-unit lifecycle management.

## Pipeline Pool Initialization

At server startup, the `RealtimeServer` class constructs a fixed-size pool of independent pipeline units. In [`src/speech_to_speech/api/openai_realtime/server.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/server.py) (lines 14‑48), the `run()` method instantiates multiple `PipelineUnit` objects—each representing a complete speech-to-speech pipeline with its own queues, service instance, and cancellation scope—and passes this collection to the FastAPI application factory.

```python
from threading import Event
from speech_to_speech.api.openai_realtime.server import RealtimeServer
from speech_to_speech.api.openai_realtime.pipeline_unit import PipelineUnit

# Build a pool of N isolated pipelines

pipeline_pool = [PipelineUnit(index=i) for i in range(N)]

stop = Event()
server = RealtimeServer(stop_event=stop, pool=pipeline_pool, host="0.0.0.0", port=8765)
server.run()          # Blocks until `stop.set()` is called

```

Each `PipelineUnit` maintains internal queues for audio and text data, creating isolated data pathways that prevent cross-contamination between concurrent sessions.

## Atomic Unit Claiming and Session Binding

When a WebSocket connection arrives, the server must acquire an idle pipeline unit exclusively. The `_claim_unit` helper function in [`src/speech_to_speech/api/openai_realtime/websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py) (lines 46‑57) implements this by scanning the pool for the first unit where `unit.session is None`.

Because the function executes the search loop without any `await` statements before returning the unit, the claim operation is **atomic**—no other coroutine can interleave and claim the same unit simultaneously. Upon finding an idle unit, the function creates a fresh `SessionState` object containing the `WebSocketTransport` and assigns it to the unit.

```python
@app.websocket("/v1/realtime")
async def realtime_endpoint(ws: WebSocket):
    await ws.accept()
    transport = WebSocketTransport(ws)

    # Try to grab the first idle pipeline unit

    unit = _claim_unit(transport)
    if unit is None:
        await send_ws_event(ws, build_error_event("All session slots are in use"))
        await ws.close(code=1008, reason="All session slots are in use")
        return

    # … register session, drain queues, then start the communication loop …

```

If no idle units exist, the endpoint returns error code `1008` with the message "All session slots are in use" and closes the connection immediately.

## WebSocket Endpoint and Connection Routing

The `/v1/realtime` WebSocket route in [`src/speech_to_speech/api/openai_realtime/websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py) (lines 59‑78) orchestrates the binding process. After claiming a unit, the server registers a unique session ID with the unit's service, invokes `_clean_unit` to drain any stale queue data, and establishes bidirectional communication.

The routing mechanism creates a dedicated pathway: client events flow into the pipeline unit's input queues, while the unit's output flows back through the WebSocket via an async send loop. This 1:1 mapping between WebSocket connections and pipeline units persists for the entire session lifecycle.

## Session Lifecycle and Per-Unit Send Loops

Each claimed pipeline unit runs an independent async task (`_send_loop_for`) defined in [`src/speech_to_speech/api/openai_realtime/websocket_router.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/websocket_router.py) (lines 22‑34 and 122‑146). This loop continuously polls the unit's output queues and forwards audio chunks and text events to the bound `WebSocketTransport`.

```python
async def _send_loop_for(unit: PipelineUnit):
    while not stop_event.is_set():
        session = unit.session                  # Snapshot the current session

        transport = session.transport if session else None
        session_id = session.session_id if session else None

        # Pull text events first

        try:
            text_msg = unit.text_output_queue.get_nowait()
            if isinstance(text_msg, PipelineEvent) and transport:
                events = unit.service.dispatch_pipeline_event(session_id, text_msg)
                await transport.send_events(events)
        except Empty:
            pass

        # Pull audio chunks and send them

        try:
            audio_chunk = unit.output_queue.get_nowait()
            await transport.send_audio_chunk(unit.service, session_id, audio_chunk)
        except Empty:
            pass

        await asyncio.sleep(0.01)

```

When the WebSocket disconnects or the client hangs up, the `_release_session` function (lines 81‑108) handles cleanup. It flushes remaining events from the unit's queues, enqueues a `SESSION_END` sentinel message to signal pipeline termination, and finally sets `unit.session = None`, making the unit available for the next incoming connection.

## Summary

- **Pool-based architecture**: The server pre-instantiates a fixed number of `PipelineUnit` objects at startup, each representing an isolated STT→LM→TTS pipeline with dedicated queues.
- **Atomic claiming**: The `_claim_unit` function scans for units with `session is None` without yielding control, ensuring exclusive, race-free assignment of pipeline instances to WebSocket connections.
- **Transport binding**: Each claimed unit receives a `SessionState` containing a `WebSocketTransport`, creating a dedicated communication channel that persists for the session duration.
- **Graceful rejection**: When the pool is exhausted, the server returns WebSocket close code `1008` and rejects new connections without consuming resources.
- **Clean release**: The `_release_session` mechanism drains queues, sends a `SESSION_END` sentinel, and resets the unit's session state to `None`, returning it to the available pool.

## Frequently Asked Questions

### What happens when all pipeline units are occupied?

When every `PipelineUnit` in the pool has an active `SessionState` (meaning `unit.session is not None`), the `_claim_unit` function returns `None`. The WebSocket endpoint responds by sending a non-chargeable error event and closing the connection with WebSocket code `1008` and the reason "All session slots are in use".

### How does the server prevent race conditions when claiming units?

The `_claim_unit` function executes its search loop synchronously without any `await` statements before returning the claimed unit. Because Python's async event loop cannot interrupt a coroutine until it hits an `await`, the scan-and-claim operation completes atomically, preventing two simultaneous connections from acquiring the same pipeline instance.

### How is a pipeline unit released after a WebSocket disconnect?

Upon disconnect, the `_release_session` function drains the unit's input and output queues, enqueues a `SESSION_END` control message to terminate any ongoing pipeline processing, and sets `unit.session = None`. This sequence ensures all pending audio chunks are processed or discarded before the unit becomes available for new connections.

### Can multiple WebSocket connections share a single pipeline unit?

No. The routing architecture enforces strict 1:1 isolation between WebSocket connections and pipeline units. Each unit maintains exclusive session state and transport references, and the atomic claiming mechanism guarantees that once a unit is assigned to a connection, no other client can access its queues or processing threads until the session is explicitly released.