Processing Multiple Streams - Sanas Developer Hub

Documentation Index

Fetch the complete documentation index at: /llms.txt

Use this file to discover all available pages before exploring further.

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.

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

Prerequisites

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

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.

Step 1: Create and activate one SDK

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.

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.

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.

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

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