> ## Documentation Index
> Fetch the complete documentation index at: https://developer.sanas.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Processing Multiple Streams

> Run many concurrent audio processors on a single SDK instance, and benchmark throughput, concurrency, and memory.

A single SDK instance can drive many audio processors at once. Because `process_frame` releases the GIL, streams running in separate threads process **in parallel**. This tutorial follows `multi_stream_example.py`, a benchmark that creates one SDK, runs N processors concurrently (one per thread), and reports success/failure counts, peak concurrency, per-stream wall time, and optional memory growth.

<Note>
  Reuse one `Sdk` object across all streams — you activate once, then create a processor per thread. Do not create an SDK per stream.
</Note>

## Prerequisites

| Variable            | Required | Default     | Purpose         |
| ------------------- | -------- | ----------- | --------------- |
| `SANAS_API_KEY`     | Yes      | —           | Licence key     |
| `SANAS_STORAGE_DIR` | No       | `./storage` | SDK data + logs |

```bash theme={null}
export SANAS_API_KEY="your-key"

python multi_stream_example.py --streams 8 --concurrency 4 --rate 2 \
    --model VI_G_SE --input test_input.wav
```

### Command-line flags

| Flag                      | Default          | Meaning                                              |
| ------------------------- | ---------------- | ---------------------------------------------------- |
| `--streams`               | `4`              | Total number of streams to run                       |
| `--concurrency`           | `= --streams`    | Max streams running at once                          |
| `--rate`                  | `0`              | Stream start rate per second (0 = all at once)       |
| `--model`                 | `VI_G_SE`        | Model key                                            |
| `--input`                 | `test_input.wav` | Input WAV path                                       |
| `--max-throughput`        | off              | Feed as fast as possible instead of real-time pacing |
| `--pcm16`                 | off              | Negotiate raw 16-bit PCM (cloud inference only)      |
| `--save-wav-out-period N` | `0`              | Save every Nth stream's output for spot-checking     |

<Tip>
  Install `psutil` (`pip install psutil`) to also report process memory growth — useful for spotting leaks in long soak runs. The benchmark runs fine without it; the memory lines are simply omitted.
</Tip>

## Step 1: Create and activate one SDK

```python theme={null}
import sanas

sdk = sanas.create_sdk(
    sanas.InitParams(storage_dir=os.environ.get("SANAS_STORAGE_DIR", "./storage"))
)
res = sdk.activate_api_key(os.environ["SANAS_API_KEY"])
if not res.success:
    sys.exit(f"Activation failed: {res.message}")
print(f"[activation] OK (SDK {sdk.version})")

samples, sample_rate, channels = load_samples(args.input)
```

## Step 2: Define a per-stream worker

Each stream builds its own `PipelineWaiter`, processor, and cloud params, then feeds the shared samples. Exceptions are caught per stream so one failure doesn't take down the others. Note `collect_output` is only turned on when you actually intend to save audio — leaving it off keeps the memory-growth metric honest.

```python theme={null}
def _run_stream(sdk, idx, samples, sample_rate, channels, model,
                realtime, use_pcm16, save_period, stats):
    result = {"idx": idx, "ok": False, "error": None, "wall_s": 0.0, "saved": None}
    stats.stream_started()
    t0 = time.monotonic()
    try:
        cloud_params = sanas.CloudInferencingParams()
        cloud_params.use_pcm16 = use_pcm16
        waiter = PipelineWaiter()
        attrs = sanas.ProcessorAttributes(
            audio_attributes=sanas.AudioAttributes(
                sampling_rate=sample_rate,
                channels=channels,
                model_name=model,
                cloud_inferencing_params=cloud_params,
                audio_pipeline_state_notify=waiter.callback,
            )
        )
        with sdk.create_audio_processor(attrs) as proc:
            waiter.wait_until_running(timeout=30.0)
            drained = feed_and_drain(
                proc, samples, sample_rate, channels,
                realtime=realtime, collect_output=save_period > 0,
            )
        result["wall_s"] = time.monotonic() - t0
        if save_period and idx % save_period == 0 and drained["output"]:
            out_path = f"stream_{idx}.wav"
            save_wav(out_path, drained["output"], sample_rate, channels)
            result["saved"] = out_path
        result["ok"] = True
    except Exception as e:   # report per-stream, keep others going
        result["error"] = f"{type(e).__name__}: {e}"
        result["wall_s"] = time.monotonic() - t0
    finally:
        stats.stream_finished(result["ok"])
    return result
```

## Step 3: Track live statistics

`LiveStats` is a thread-safe counter aggregated across streams. Its `active` and `peak_concurrent` fields reveal how many streams actually overlapped — something per-stream result dicts can't show on their own.

```python theme={null}
class LiveStats:
    def __init__(self):
        self._lock = threading.Lock()
        self.started = self.completed = self.failed = self.active = 0
        self.peak_concurrent = 0
        self.start_time = time.monotonic()

    def stream_started(self):
        with self._lock:
            self.started += 1
            self.active += 1
            self.peak_concurrent = max(self.peak_concurrent, self.active)

    def stream_finished(self, ok):
        with self._lock:
            self.active -= 1
            if ok:
                self.completed += 1
            else:
                self.failed += 1
```

A background `MemorySampler` thread records peak RSS every 0.5 s (a no-op without `psutil`). Sampling — rather than reading once at the end — captures the true high-water mark while streams overlap, so a leak or transient spike shows up in the report.

## Step 4: Launch streams on a thread pool

Use a `ThreadPoolExecutor` sized to `--concurrency`. Optionally stagger starts with `--rate` to model a ramp-up instead of a thundering herd.

```python theme={null}
from concurrent.futures import ThreadPoolExecutor, as_completed

with ThreadPoolExecutor(max_workers=concurrency) as pool:
    futures = []
    for i in range(args.streams):
        futures.append(pool.submit(
            _run_stream, sdk, i, samples, sample_rate, channels,
            args.model, realtime, args.pcm16, args.save_wav_out_period, stats,
        ))
        if args.rate > 0 and i + 1 < args.streams:
            time.sleep(1.0 / args.rate)   # stagger stream starts
    for fut in as_completed(futures):
        r = fut.result()
        tag = "ok" if r["ok"] else f"FAILED ({r['error']})"
        print(f"[stream {r['idx']}] {tag}  wall={r['wall_s']:.2f}s")
```

## Step 5: Read the report

```text theme={null}
============================================================
Benchmark summary
============================================================
streams        : 8  (ok=8, failed=0)
success rate   : 100.0%
peak concurrent: 4
uptime         : 12.34s
memory (RSS)   : start=95.2 MB  peak=180.6 MB  growth=+85.4 MB
```

Interpreting it: **success rate** flags whether the configured concurrency is sustainable; **peak concurrent** confirms streams actually overlapped up to your `--concurrency` limit; **memory growth** should stay flat across a long soak run — steady climbing suggests a leak.

## Tuning tips

Use `--max-throughput` to measure how fast the pipeline *can* process (no real-time pacing) versus the default real-time feed that models a live call. Raise `--concurrency` gradually and watch the success rate and memory growth to find a safe ceiling for your hardware and model. Leave `--save-wav-out-period` at `0` for benchmarking; set it to spot-check output quality on a sample of streams.

## Next Steps

<CardGroup cols={2}>
  <Card title="Processing a Single Stream" icon="waveform-lines" href="/Docs/Tutorials/Single-Stream">
    The core flow this benchmark scales up.
  </Card>

  <Card title="API Reference" icon="book" href="/API-Reference/Overview">
    Full SDK documentation for classes, enums, and callbacks.
  </Card>
</CardGroup>

## Next Steps

<CardGroup cols={2}>
  <Card title="Language Translation" icon="language" href="/Docs/Tutorials/Language-Translation">
    Translate speech between languages and receive audio plus transcripts.
  </Card>

  <Card title="Processing Multiple Streams" icon="layer-group" href="/Docs/Tutorials/Processing-Multiple-Streams">
    Run many concurrent processors on one SDK instance.
  </Card>
</CardGroup>

## References

<CardGroup cols={2}>
  <Card title="SDK" icon="server" href="/API-Reference/Reference/Core-Classes/RemoteSDK">
    SDK lifecycle methods including create\_sdk, activate\_api\_key, and create\_audio\_processor.
  </Card>

  <Card title="AudioProcessor" icon="waveform-lines" href="/API-Reference/Reference/Core-Classes/AudioProcessor">
    process\_frame method for audio processing.
  </Card>
</CardGroup>

***

## Need Help?

<CardGroup cols={2}>
  <Card title="Email Support" icon="envelope" href="mailto:support@sanas.ai">
    [support@sanas.ai](mailto:support@sanas.ai)

    Response time: 1 business day
  </Card>

  <Card title="Support Portal" icon="ticket" href="https://support.sanas.ai">
    Raise a [support ticket](https://help.sanas.ai/) for urgent issues
  </Card>
</CardGroup>