# Managing Pipeline Thread Pools and Queue Sizes for Concurrent Sessions in Hugging Face Speech-to-Speech

> Optimize concurrent speech-to-speech sessions by managing pipeline thread pools and queue sizes. Learn how to configure --num_pipelines for efficient handler threads in Hugging Face speech-to-speech.

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

---

**To manage concurrent speech-to-speech sessions, the repository uses a `ThreadManager` to orchestrate handler threads and creates a pool of independent `PipelineUnit` objects—each with dedicated queues—controlled by the `--num_pipelines` CLI argument.**

The huggingface/speech-to-speech repository implements a real-time speech-to-speech pipeline capable of serving multiple isolated WebSocket sessions simultaneously. Understanding how to configure thread pools and queue sizes is essential for deploying this system at scale without memory exhaustion or cross-session interference.

## Architecture of Concurrent Session Management

### Thread Pool Orchestration with ThreadManager

The **ThreadManager** class in [`src/speech_to_speech/utils/thread_manager.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/utils/thread_manager.py) manages the lifecycle of all OS threads. It receives every pipeline handler—VAD, STT, LLM, TTS, the server, and the client—as a flat list and creates a `threading.Thread` for each. 

During startup, the manager calls `Thread.start()` on each handler. During shutdown, it signals termination by setting each handler’s `stop_event`, then calls `Thread.join()` with a 5‑second timeout to ensure graceful termination. If any thread fails to exit within the timeout, the manager emits a warning but forces the process to continue, preventing zombie threads from blocking deployment.

### Pipeline Isolation via the PipelineUnit Pool

Concurrency is achieved through a pool of **PipelineUnit** objects rather than shared resources. The `build_pipeline` function in [`src/speech_to_speech/s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py) constructs `module_kwargs.num_pipelines` separate units by repeatedly invoking `_build_pipeline_unit`.

Each **PipelineUnit** (defined in [`src/speech_to_speech/api/openai_realtime/pipeline_unit.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/pipeline_unit.py)) encapsulates its own `VADHandler`, STT handler, LLM handler, TTS handler, and dedicated `queue.Queue` instances. Because no queue is shared between units, audio chunks and text tokens from one session cannot leak into another. The `RealtimeServer` 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) assigns a free `PipelineUnit` to each incoming WebSocket connection and returns it to the pool upon disconnect, making the maximum concurrent session count directly proportional to the configured pool size.

## Configuring Concurrent Sessions

### Setting the Number of Pipelines via CLI

The entry point for concurrency configuration is the `ModuleArguments` class in [`src/speech_to_speech/arguments_classes/module_arguments.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/arguments_classes/module_arguments.py). The `--num_pipelines` argument defaults to `1` and determines how many independent pipeline units are instantiated. Increasing this value allows the `RealtimeServer` to accept that many simultaneous WebSocket connections.

```bash
python -m speech_to_speech.cli serve --num_pipelines 4 --log_level debug

```

*This command creates a pool of four pipelines, allowing up to four simultaneous WebSocket sessions.*

### Queue Types and Handler Contracts

All inter-handler communication is type-checked through aliases defined in [`src/speech_to_speech/pipeline/queue_types.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/pipeline/queue_types.py). These include `AudioInItem`, `VADOutItem`, `STTOutItem`, and others. These types serve as the contract between handlers, ensuring that a VAD handler outputs items compatible with the STT handler’s input queue.

## Tuning Queue Sizes for Production

### Bounded Microphone Input Queue

The local audio client uses a bounded queue to prevent unbounded memory growth when network latency spikes. In [`src/speech_to_speech/api/openai_realtime/audio_client.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/api/openai_realtime/audio_client.py), the microphone queue is initialized as:

```python
from queue import Queue

mic_queue: Queue[bytes] = Queue(maxsize=128)

```

When the queue reaches capacity, excess chunks are discarded and a debug message is logged. To increase tolerance for high-latency networks, modify `maxsize` before instantiating the client:

```python
mic_queue: Queue[bytes] = Queue(maxsize=256)  # Increase from 128

```

### Internal Pipeline Queue Sizing and Back-Pressure

By default, internal queues created inside `_build_pipeline_unit` are unbounded (`Queue()`). This design delegates back-pressure to the handlers themselves—typically by dropping chunks when downstream stages lag. If your deployment requires strict memory caps, replace these with bounded queues and handle the `Full` exception explicitly:

```python
from queue import Queue, Full

# Inside _build_pipeline_unit, replace queue definition:

recv_audio_chunks_queue: Queue[AudioInItem] = Queue(maxsize=32)

# In the VAD handler's callback:

try:
    recv_audio_chunks_queue.put_nowait(chunk)
except Full:
    logger.debug("Dropping VAD chunk – queue full")

```

*This caps the amount of raw audio that can be in-flight, preventing memory blow-up when the LLM or TTS handlers experience latency.*

## Graceful Shutdown and Thread Lifecycle

The shutdown sequence is coordinated through the `stop_event` shared across all handlers. When `ThreadManager.stop()` is invoked, it first sets the event, then joins each thread with a 5‑second timeout. The server’s internal `_watch_stop` thread monitors the same event and forces `uvicorn` to exit by setting `server.should_exit = True`. This coordinated stop ensures all queues are drained and resources such as audio streams and sockets are released cleanly.

You can inspect the active thread pool programmatically:

```python
pipeline_manager = build_pipeline(args, stop_event)
print(f"Running {len(pipeline_manager.threads)} handler threads")
pipeline_manager.start()
pipeline_manager.wait()

```

*This outputs the total number of threads spawned, which equals the number of handlers across all pipeline units plus the server thread.*

## Summary

- **ThreadManager** in [`src/speech_to_speech/utils/thread_manager.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/utils/thread_manager.py) orchestrates all handler threads and provides graceful shutdown with a 5‑second timeout.
- **PipelineUnit** pools in [`src/speech_to_speech/s2s_pipeline.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/s2s_pipeline.py) isolate sessions; the `--num_pipelines` CLI argument controls concurrency.
- **Queue types** in [`src/speech_to_speech/pipeline/queue_types.py`](https://github.com/huggingface/speech-to-speech/blob/main/src/speech_to_speech/pipeline/queue_types.py) define typed contracts between handlers.
- The **microphone queue** in the audio client uses a bounded `maxsize=128` to prevent memory growth on slow networks.
- **Internal pipeline queues** are unbounded by default but can be modified to bounded queues with explicit `Full` handling for custom back-pressure.

## Frequently Asked Questions

### How do I increase the number of concurrent users?

Set the `--num_pipelines` argument to the desired number of simultaneous sessions. Each increment creates a fully isolated `PipelineUnit` with its own handlers and queues, allowing the `RealtimeServer` to assign one unit per WebSocket connection without cross-talk.

### Why are internal pipeline queues unbounded by default?

Unbounded queues (`Queue()`) simplify the handler logic by removing the need for exception handling around `put` operations. Back-pressure is instead implemented at the handler level—typically by dropping audio chunks when the consumer lags—ensuring real-time latency is prioritized over reliability.

### How do I prevent memory issues with slow clients?

For the local audio client, increase the `maxsize` of the microphone queue if you expect network jitter, or decrease it to drop data earlier. For server deployments, modify `_build_pipeline_unit` to use bounded queues with `maxsize` limits and catch `queue.Full` exceptions to implement custom dropping or blocking behavior.

### What happens if a thread doesn't stop during shutdown?

The `ThreadManager` attempts to join each thread for 5 seconds after setting the `stop_event`. If a thread fails to terminate within this window, the manager logs a warning and proceeds with shutdown, ensuring the process does not hang indefinitely while leaving the non-responsive thread behind.