feat(qwen3-tts-cpp): batch concurrent requests via qwentts max_batch

qwentts.cpp batches concurrent qt_synthesize calls on one context when it
is initialized with max_batch > 1 (its compute worker coalesces them into
batched decode steps; qt_synthesize stays blocking and thread safe). The
LocalAI wrapper never set max_batch and serialized every request with
base.SingleThread, so concurrent /v1/audio/speech requests always ran one
after another.

- new model option `max_batch:N` (alias `parallel:N`, default 1 = current
  behavior), passed to qt_init_params.max_batch via qt3_load
- Locking() keeps LocalAI's per-backend request lock only for max_batch 1;
  with batching TTS/TTSStream reach the native worker concurrently
- TTSStream routes chunks per request: a fresh id travels as the C
  callback's user_data (one shared purego callback, as before), each request
  has its own queue drained into its channel, so the native worker never
  blocks on a slow consumer and concurrent streams cannot mix
- TTSLive rejects max_batch > 1 (qwentts incremental mode requires a
  max_batch=1 context, docs/INCREMENTAL.md)
- README: option, measurements, TTSLive limit, ROCm/gfx1201 hipBLASLt note
- unit tests: option parsing, Locking, concurrent stream routing (fails if
  requests were serialized), TTSLive rejection

Measured with qwentts' own tts-server and the 1.7B Base Q4_K_M talker,
four concurrent streaming requests, time to first PCM: RTX 5060 Ti (CUDA)
0.02/0.36/0.70/1.04 s at max_batch 1 vs 0.06 s for all four at 4;
Radeon AI PRO R9700 (ROCm, with ROCBLAS_USE_HIPBLASLT=0) 0.03/0.71/1.36/
2.02 s vs <= 0.11 s.

Signed-off-by: nexxtmobile.de <git@nexxtmobile.de>
Assisted-by: Claude:claude-opus-5-5
(cherry picked from commit da0c5435b5)

backend/go/qwen3-tts-cpp/README.md

@@ -50,6 +50,31 @@ results and rejects saved `localai://voice-profiles/...` references. It does
 not disable Qwen's named-speaker or VoiceDesign modes. Setting it to `true`
 cannot add cloning to a backend that lacks LocalAI's reference-audio contract.
 
+### Concurrent requests (`max_batch`)
+
+By default one request is synthesized at a time. `max_batch:N` (alias
+`parallel:N`) initializes qwentts with `max_batch = N`: its compute worker
+coalesces up to N concurrent `/v1/audio/speech` requests (streaming or not)
+into batched decode steps, and LocalAI stops serializing requests to the
+backend. Further requests queue FIFO inside qwentts.
+
+```yaml
+options:
+  - max_batch:4
+```
+
+Measured with the 1.7B Base Q4_K_M talker and four concurrent streaming
+requests (time to the first PCM chunk): RTX 5060 Ti 0.02/0.36/0.70/1.04 s with
+`max_batch:1`, 0.06 s for all four with `max_batch:4`. VRAM grows with N
+(about 2.6 GB at 1, 6 GB at 4, 10.5 GB at 8 for that talker).
+
+- Incremental synthesis (`TTSLive`, the realtime pipeline with
+  `streaming.tts: true`) requires `max_batch:1` and is rejected otherwise
+  (qwentts `docs/INCREMENTAL.md`).
+- ROCm on gfx1201 (RDNA4) hangs the GPU in hipBLASLt for batched matmuls;
+  `env: {ROCBLAS_USE_HIPBLASLT: "0"}` in the model config routes them to
+  rocBLAS. CUDA needs no workaround.
+
 Request precedence is: a request `voice`, then `tts.voice`, then
 `tts.audio_path`. A saved profile supplies its private WAV and exact transcript
 for that request without changing the model YAML.

backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.cpp

@@ -25,7 +25,7 @@ static void ggml_log_cb(enum ggml_log_level level, const char *log,
 }
 
 int qt3_load(const char *talker_path, const char *codec_path, int use_fa,
-             int clamp_fp16) {
+             int clamp_fp16, int max_batch) {
     ggml_log_set(ggml_log_cb, nullptr);
     ggml_backend_load_all();
 
@@ -44,9 +44,13 @@ int qt3_load(const char *talker_path, const char *codec_path, int use_fa,
     p.codec_path = codec_path;
     p.use_fa = use_fa != 0;
     p.clamp_fp16 = clamp_fp16 != 0;
+    // > 1: qt's compute worker coalesces concurrent qt_synthesize calls on
+    // this handle into batched decode steps (qt_synthesize stays blocking and
+    // thread safe; streaming callbacks run on that worker thread).
+    p.max_batch = max_batch > 1 ? max_batch : 1;
 
-    fprintf(stderr, "[qwen3-tts-cpp] Loading talker=%s codec=%s\n", talker_path,
-            codec_path);
+    fprintf(stderr, "[qwen3-tts-cpp] Loading talker=%s codec=%s max_batch=%d\n",
+            talker_path, codec_path, p.max_batch);
 
     g_ctx = qt_init(&p);
     if (!g_ctx) {

backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.h

@@ -8,11 +8,13 @@ extern "C" {
 typedef int (*qt3_chunk_cb)(const float *samples, int n_samples,
                             void *user_data);
 
-// Load the talker + codec/tokenizer GGUFs. use_fa / clamp_fp16 map to
-// qt_init_params (the qt ABI exposes no thread count; ggml uses its own
-// default). Returns 0 on success, non-zero on failure.
+// Load the talker + codec/tokenizer GGUFs. use_fa / clamp_fp16 / max_batch
+// map to qt_init_params (the qt ABI exposes no thread count; ggml uses its
+// own default). max_batch > 1 lets concurrent qt3_tts / qt3_tts_stream calls
+// share batched decode steps; 0 or 1 keeps single-sequence synthesis.
+// Returns 0 on success, non-zero on failure.
 int qt3_load(const char *talker_path, const char *codec_path, int use_fa,
-             int clamp_fp16);
+             int clamp_fp16, int max_batch);
 
 // Synthesize to a malloc'd float PCM buffer (caller frees via qt3_pcm_free).
 // The synthesis mode (base / custom_voice / voice_design) is auto-detected by

backend/go/qwen3-tts-cpp/e2e_test.go

@@ -36,7 +36,7 @@ var _ = Describe("qwen3-tts-cpp e2e", Label("e2e"), func() {
 			purego.RegisterLibFunc(&CppTTSStream, h, "qt3_tts_stream")
 			purego.RegisterLibFunc(&CppPCMFree, h, "qt3_pcm_free")
 			purego.RegisterLibFunc(&CppUnload, h, "qt3_unload")
-			Expect(CppLoad(modelPath, codecPath, 1, 0)).To(Equal(0))
+			Expect(CppLoad(modelPath, codecPath, 1, 0, 1)).To(Equal(0))
 			loaded = true
 		}
 	})

backend/go/qwen3-tts-cpp/goqwen3ttscpp.go

@@ -6,6 +6,7 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"unsafe"
 
 	"github.com/ebitengine/purego"
@@ -14,8 +15,8 @@ import (
 )
 
 var (
-	// qt3_load(talker_path, codec_path, use_fa, clamp_fp16) int
-	CppLoad func(talkerPath, codecPath string, useFA, clampFP16 int) int
+	// qt3_load(talker_path, codec_path, use_fa, clamp_fp16, max_batch) int
+	CppLoad func(talkerPath, codecPath string, useFA, clampFP16, maxBatch int) int
 	// qt3_tts(text, lang, instruct, speaker, ref_samples, ref_n, ref_text,
 	//         seed, temperature, top_k, top_p, rep_pen, max_new, out_n) -> float*
 	CppTTS func(text, lang, instruct, speaker string, refSamples unsafe.Pointer,
@@ -35,6 +36,18 @@ type Qwen3TtsCpp struct {
 	// audioPath is the model-config reference voice (tts.audio_path), the
 	// default clone reference when a request omits an audio Voice.
 	audioPath string
+	// maxBatch mirrors opts.maxBatch for Locking(), which the gRPC server
+	// reads on every request (atomic: Load writes it once, requests read it).
+	maxBatch atomic.Int32
+}
+
+// Locking keeps LocalAI's per-backend request lock only for single-sequence
+// synthesis. With max_batch > 1 the native compute worker owns every backend
+// compute on the handle and batches concurrent qt_synthesize calls, so
+// TTS/TTSStream requests must reach it concurrently instead of one at a time.
+// Before Load (maxBatch 0) the lock stays on, so loading is serialized.
+func (q *Qwen3TtsCpp) Locking() bool {
+	return q.maxBatch.Load() <= 1
 }
 
 func (q *Qwen3TtsCpp) Load(opts *pb.ModelOptions) error {
@@ -70,12 +83,13 @@ func (q *Qwen3TtsCpp) Load(opts *pb.ModelOptions) error {
 	useFA := boolToInt(q.opts.useFA)
 	clamp := boolToInt(q.opts.clampFP16)
 
-	fmt.Fprintf(os.Stderr, "[qwen3-tts-cpp] Load talker=%s codec=%s use_fa=%d clamp_fp16=%d\n",
-		model, codec, useFA, clamp)
+	fmt.Fprintf(os.Stderr, "[qwen3-tts-cpp] Load talker=%s codec=%s use_fa=%d clamp_fp16=%d max_batch=%d\n",
+		model, codec, useFA, clamp, q.opts.maxBatch)
 
-	if rc := CppLoad(model, codec, useFA, clamp); rc != 0 {
+	if rc := CppLoad(model, codec, useFA, clamp, q.opts.maxBatch); rc != 0 {
 		return fmt.Errorf("qwen3-tts: failed to load model (rc=%d)", rc)
 	}
+	q.maxBatch.Store(int32(q.opts.maxBatch))
 	return nil
 }
 
@@ -169,26 +183,104 @@ func (q *Qwen3TtsCpp) TTS(req *pb.TTSRequest) error {
 	return writeWAV24k(req.Dst, out)
 }
 
-// streamState carries the active TTSStream channel to the single shared C
-// callback. base.SingleThread serializes TTS/TTSStream, so one global slot is
-// safe and avoids leaking a purego callback per request (purego callbacks
-// cannot be freed and are capped).
+// One purego callback is shared by every TTSStream request (purego callbacks
+// cannot be freed and are capped). Each request registers a streamRoute under
+// a fresh id that travels as the C callback's user_data, so concurrent
+// requests (max_batch > 1) each get their own chunks. streamMu serializes the
+// incremental (TTSLive) path, whose state is global (see live.go).
 var (
 	streamMu     sync.Mutex
-	streamChan   chan []byte
 	streamCbOnce sync.Once
 	streamCbPtr  uintptr
+
+	streamRoutesMu sync.Mutex
+	streamRoutes   = map[uintptr]*streamRoute{}
+	streamNextID   uintptr
 )
 
-// streamCallback is registered once and forwards each PCM chunk to streamChan.
-func streamCallback(samples *float32, nSamples int32, _ uintptr) uintptr {
-	if nSamples <= 0 || samples == nil || streamChan == nil {
+// streamRoute queues one request's PCM chunks. With max_batch > 1 the native
+// worker invokes every request's callback from its single compute thread, so
+// push never blocks on the consumer: a slow client must not stall the other
+// requests of the batch. forward drains the queue into the request's channel.
+type streamRoute struct {
+	mu     sync.Mutex
+	cond   *sync.Cond
+	chunks [][]byte
+	closed bool
+}
+
+func newStreamRoute() *streamRoute {
+	r := &streamRoute{}
+	r.cond = sync.NewCond(&r.mu)
+	return r
+}
+
+func (r *streamRoute) push(b []byte) {
+	r.mu.Lock()
+	r.chunks = append(r.chunks, b)
+	r.mu.Unlock()
+	r.cond.Signal()
+}
+
+func (r *streamRoute) close() {
+	r.mu.Lock()
+	r.closed = true
+	r.mu.Unlock()
+	r.cond.Signal()
+}
+
+// forward sends queued chunks to out until close() and the queue is empty.
+func (r *streamRoute) forward(out chan<- []byte) {
+	for {
+		r.mu.Lock()
+		for len(r.chunks) == 0 && !r.closed {
+			r.cond.Wait()
+		}
+		batch := r.chunks
+		r.chunks = nil
+		done := r.closed
+		r.mu.Unlock()
+		for _, b := range batch {
+			out <- b
+		}
+		if done && len(batch) == 0 {
+			return
+		}
+	}
+}
+
+func registerStreamRoute() (uintptr, *streamRoute) {
+	r := newStreamRoute()
+	streamRoutesMu.Lock()
+	streamNextID++
+	id := streamNextID
+	streamRoutes[id] = r
+	streamRoutesMu.Unlock()
+	return id, r
+}
+
+func unregisterStreamRoute(id uintptr) {
+	streamRoutesMu.Lock()
+	delete(streamRoutes, id)
+	streamRoutesMu.Unlock()
+}
+
+// streamCallback is registered once and routes each PCM chunk to the request
+// identified by user_data.
+func streamCallback(samples *float32, nSamples int32, user uintptr) uintptr {
+	if nSamples <= 0 || samples == nil {
+		return 1 // continue
+	}
+	streamRoutesMu.Lock()
+	r := streamRoutes[user]
+	streamRoutesMu.Unlock()
+	if r == nil {
 		return 1 // continue
 	}
 	src := unsafe.Slice(samples, int(nSamples))
 	cp := make([]float32, int(nSamples)) // copy out of C memory before returning
 	copy(cp, src)
-	streamChan <- floatToPCM16LE(cp)
+	r.push(floatToPCM16LE(cp))
 	return 1 // continue
 }
 
@@ -214,12 +306,17 @@ func (q *Qwen3TtsCpp) TTSStream(req *pb.TTSRequest, results chan []byte) error {
 	// Emit the WAV header first so the HTTP layer gets a self-describing stream.
 	results <- wavHeader24k()
 
-	streamMu.Lock()
-	streamChan = results
+	id, route := registerStreamRoute()
+	forwarded := make(chan struct{})
+	go func() {
+		route.forward(results)
+		close(forwarded)
+	}()
 	rc := CppTTSStream(req.Text, lang, instruct, speaker, refPtr, len(ref), refText,
-		s.seed, s.temperature, s.topK, s.topP, s.repPen, s.maxNew, streamCbPtr, 0)
-	streamChan = nil
-	streamMu.Unlock()
+		s.seed, s.temperature, s.topK, s.topP, s.repPen, s.maxNew, streamCbPtr, id)
+	unregisterStreamRoute(id)
+	route.close()
+	<-forwarded // every chunk is in results before the deferred close
 	runtimeKeepAlive(ref)
 
 	if rc != 0 {

backend/go/qwen3-tts-cpp/live.go

@@ -81,6 +81,11 @@ func (q *Qwen3TtsCpp) TTSLive(ctx context.Context, recv func() (*pb.TTSRequest,
 	if CppTTSLive == nil {
 		return fmt.Errorf("qwen3-tts: incremental synthesis unavailable in native library")
 	}
+	if q.maxBatch.Load() > 1 {
+		// qt's incremental mode requires a max_batch=1 context: a producer
+		// blocked on text owns the compute worker (qwentts docs/INCREMENTAL.md).
+		return fmt.Errorf("qwen3-tts: incremental synthesis (TTSLive) requires max_batch=1; this model is loaded with max_batch=%d", q.maxBatch.Load())
+	}
 	if err := send(wavHeader24k()); err != nil {
 		return err
 	}

backend/go/qwen3-tts-cpp/live_native_test.go

@@ -28,7 +28,7 @@ var _ = Describe("TTSLiveNative", func() {
 		purego.RegisterLibFunc(&CppLoad, lib, "qt3_load")
 		purego.RegisterLibFunc(&CppUnload, lib, "qt3_unload")
 		purego.RegisterLibFunc(&CppTTSLive, lib, "qt3_tts_live")
-		if rc := CppLoad(os.Getenv("QWEN3TTS_MODEL"), os.Getenv("QWEN3TTS_CODEC"), 1, 0); rc != 0 {
+		if rc := CppLoad(os.Getenv("QWEN3TTS_MODEL"), os.Getenv("QWEN3TTS_CODEC"), 1, 0, 1); rc != 0 {
 			Fail(fmt.Sprint("load", rc))
 		}
 		defer CppUnload()

backend/go/qwen3-tts-cpp/options.go

@@ -11,6 +11,9 @@ type loadOptions struct {
 	useFA     bool
 	clampFP16 bool
 	seed      int64
+	// maxBatch > 1 sizes qt's KV cache sets and lets its compute worker batch
+	// concurrent TTS/TTSStream requests on one handle (qt_init_params.max_batch).
+	maxBatch int
 }
 
 // sampling holds per-request generation parameters with qt defaults applied.
@@ -37,7 +40,7 @@ func parseBool(v string) bool { return v == "true" || v == "1" }
 // ignored. Defaults: use_fa true (qt default; CPU still uses the F32 chain),
 // seed -1 (engine random).
 func parseOptions(opts []string) loadOptions {
-	o := loadOptions{useFA: true, seed: -1}
+	o := loadOptions{useFA: true, seed: -1, maxBatch: 1}
 	for _, oo := range opts {
 		key, value, ok := splitOption(oo)
 		if !ok {
@@ -54,6 +57,10 @@ func parseOptions(opts []string) loadOptions {
 			if n, err := strconv.ParseInt(value, 10, 64); err == nil {
 				o.seed = n
 			}
+		case "max_batch", "parallel":
+			if n, err := strconv.Atoi(value); err == nil && n >= 1 {
+				o.maxBatch = n
+			}
 		}
 	}
 	return o

backend/go/qwen3-tts-cpp/qwen3ttscpp_test.go

@@ -2,9 +2,12 @@ package main
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
 	pb "github.com/mudler/LocalAI/pkg/grpc/proto"
+	"sync"
 	"testing"
+	"time"
 	"unsafe"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -152,3 +155,106 @@ var _ = Describe("floatToPCM16LE", func() {
 		Expect(read(8)).To(Equal(int16(-32767))) // clamped from -2.0
 	})
 })
+
+var _ = Describe("max_batch option", func() {
+	It("defaults to 1 (single-sequence synthesis)", func() {
+		Expect(parseOptions(nil).maxBatch).To(Equal(1))
+	})
+	It("reads max_batch and the parallel alias", func() {
+		Expect(parseOptions([]string{"max_batch:4"}).maxBatch).To(Equal(4))
+		Expect(parseOptions([]string{"parallel:2"}).maxBatch).To(Equal(2))
+	})
+	It("ignores values below 1 and non-numbers", func() {
+		Expect(parseOptions([]string{"max_batch:0"}).maxBatch).To(Equal(1))
+		Expect(parseOptions([]string{"max_batch:many"}).maxBatch).To(Equal(1))
+	})
+})
+
+var _ = Describe("Locking", func() {
+	It("keeps the request lock before Load and for max_batch 1", func() {
+		b := &Qwen3TtsCpp{}
+		Expect(b.Locking()).To(BeTrue())
+		b.maxBatch.Store(1)
+		Expect(b.Locking()).To(BeTrue())
+	})
+	It("drops the request lock for max_batch > 1", func() {
+		b := &Qwen3TtsCpp{}
+		b.maxBatch.Store(4)
+		Expect(b.Locking()).To(BeFalse())
+	})
+})
+
+var _ = Describe("TTSStream routing", func() {
+	It("delivers each concurrent request its own chunks", func() {
+		original := CppTTSStream
+		DeferCleanup(func() { CppTTSStream = original })
+		// Both fake syntheses wait until the other one has started, so the test
+		// fails (deadlocks into the timeout) if requests were serialized.
+		var started sync.WaitGroup
+		started.Add(2)
+		CppTTSStream = func(text, _, _, _ string, _ unsafe.Pointer, _ int, _ string, _ int64, _ float32, _ int, _, _ float32, _ int, _ uintptr, user uintptr) int {
+			started.Done()
+			started.Wait()
+			v := float32(0.25)
+			if text == "b" {
+				v = -0.25
+			}
+			samples := []float32{v, v, v}
+			for i := 0; i < 3; i++ {
+				streamCallback(&samples[0], int32(len(samples)), user)
+			}
+			return 0
+		}
+		b := &Qwen3TtsCpp{}
+		b.maxBatch.Store(2)
+		run := func(text string) [][]byte {
+			results := make(chan []byte)
+			errc := make(chan error, 1)
+			go func() { errc <- b.TTSStream(&pb.TTSRequest{Text: text}, results) }()
+			var got [][]byte
+			for chunk := range results {
+				got = append(got, chunk)
+			}
+			Expect(<-errc).To(Succeed())
+			return got
+		}
+		var gotA, gotB [][]byte
+		var wg sync.WaitGroup
+		wg.Add(2)
+		go func() { defer GinkgoRecover(); defer wg.Done(); gotA = run("a") }()
+		go func() { defer GinkgoRecover(); defer wg.Done(); gotB = run("b") }()
+		done := make(chan struct{})
+		go func() { wg.Wait(); close(done) }()
+		Eventually(done).WithTimeout(5 * time.Second).Should(BeClosed())
+
+		expectChunks := func(got [][]byte, want []byte) {
+			Expect(got).To(HaveLen(4)) // WAV header + 3 PCM chunks
+			Expect(got[0]).To(Equal(wavHeader24k()))
+			for _, c := range got[1:] {
+				Expect(c).To(Equal(want))
+			}
+		}
+		expectChunks(gotA, floatToPCM16LE([]float32{0.25, 0.25, 0.25}))
+		expectChunks(gotB, floatToPCM16LE([]float32{-0.25, -0.25, -0.25}))
+		streamRoutesMu.Lock()
+		defer streamRoutesMu.Unlock()
+		Expect(streamRoutes).To(BeEmpty())
+	})
+})
+
+var _ = Describe("TTSLive with max_batch", func() {
+	It("rejects incremental synthesis on a batched context", func() {
+		original := CppTTSLive
+		DeferCleanup(func() { CppTTSLive = original })
+		CppTTSLive = func(_, _, _ string, _ unsafe.Pointer, _ int, _ string, _ int64, _ float32, _ int, _, _ float32, _ int, _, _, _ uintptr) int {
+			Fail("native incremental synthesis must not be reached")
+			return 1
+		}
+		b := &Qwen3TtsCpp{}
+		b.maxBatch.Store(4)
+		err := b.TTSLive(context.Background(),
+			func() (*pb.TTSRequest, error) { return &pb.TTSRequest{Text: "hi"}, nil },
+			func([]byte) error { return nil })
+		Expect(err).To(MatchError(ContainSubstring("requires max_batch=1")))
+	})
+})