Processing Multiple Streams

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

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

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

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.

Next Steps

Processing a Single Stream

The core flow this benchmark scales up.

API Reference

Full SDK documentation for classes, enums, and callbacks.

Next Steps

Language Translation

Translate speech between languages and receive audio plus transcripts.

Run many concurrent processors on one SDK instance.

References

SDK

SDK lifecycle methods including create_sdk, activate_api_key, and create_audio_processor.

AudioProcessor

process_frame method for audio processing.

Need Help?

Email Support

support@sanas.aiResponse time: 1 business day

Support Portal

Raise a support ticket for urgent issues

Getting Started

Models

Deployment

Tutorials/Examples

Enterprise

Resources

Processing Multiple Streams

Prerequisites

Command-line flags

Step 1: Create and activate one SDK

Step 2: Define a per-stream worker

Step 3: Track live statistics

Step 4: Launch streams on a thread pool

Step 5: Read the report

Tuning tips

Next Steps

Processing a Single Stream

API Reference

Next Steps

Language Translation