diff --git a/backend/go/qwen3-tts-cpp/README.md b/backend/go/qwen3-tts-cpp/README.md index 247c39c3c3..b354b6ab97 100644 --- a/backend/go/qwen3-tts-cpp/README.md +++ b/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. diff --git a/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.cpp b/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.cpp index d1b49883c8..8bc348f7db 100644 --- a/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.cpp +++ b/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) { diff --git a/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.h b/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.h index 351592ba32..1eedcae676 100644 --- a/backend/go/qwen3-tts-cpp/cpp/goqwen3ttscpp.h +++ b/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 diff --git a/backend/go/qwen3-tts-cpp/e2e_test.go b/backend/go/qwen3-tts-cpp/e2e_test.go index 8c2cd8d294..3d1fb9eecf 100644 --- a/backend/go/qwen3-tts-cpp/e2e_test.go +++ b/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 } }) diff --git a/backend/go/qwen3-tts-cpp/goqwen3ttscpp.go b/backend/go/qwen3-tts-cpp/goqwen3ttscpp.go index e8b3b0feb7..b4e0a9cf92 100644 --- a/backend/go/qwen3-tts-cpp/goqwen3ttscpp.go +++ b/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 { diff --git a/backend/go/qwen3-tts-cpp/live.go b/backend/go/qwen3-tts-cpp/live.go index 644e974e4d..29cafd64e3 100644 --- a/backend/go/qwen3-tts-cpp/live.go +++ b/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 } diff --git a/backend/go/qwen3-tts-cpp/live_native_test.go b/backend/go/qwen3-tts-cpp/live_native_test.go index a1b4fb37a6..1f3b8f351e 100644 --- a/backend/go/qwen3-tts-cpp/live_native_test.go +++ b/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() diff --git a/backend/go/qwen3-tts-cpp/options.go b/backend/go/qwen3-tts-cpp/options.go index 67837db0ec..e50b6972db 100644 --- a/backend/go/qwen3-tts-cpp/options.go +++ b/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 diff --git a/backend/go/qwen3-tts-cpp/qwen3ttscpp_test.go b/backend/go/qwen3-tts-cpp/qwen3ttscpp_test.go index 79fffc68fe..7b35a193d8 100644 --- a/backend/go/qwen3-tts-cpp/qwen3ttscpp_test.go +++ b/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"))) + }) +})