feat(parakeet-cpp): dynamic batching for concurrent transcription requests

backend/go/parakeet-cpp/Makefile

@@ -1,6 +1,6 @@
 # parakeet-cpp backend Makefile.
 #
-# Upstream pin lives below as PARAKEET_VERSION?=cb45f68068081af01e7092e91b038ee353eb56be
+# Upstream pin lives below as PARAKEET_VERSION?=8a7c48209d7882a7ce79a6b306270e4703194543
 # (.github/bump_deps.sh) can find and update it - matches the
 # whisper.cpp / ds4 / vibevoice-cpp convention.
 #
@@ -15,7 +15,7 @@
 # That's what the L0 smoke test uses. The default target below does the
 # proper clone-at-pin + cmake build so CI doesn't need a side-checkout.
 
-PARAKEET_VERSION?=cb45f68068081af01e7092e91b038ee353eb56be
+PARAKEET_VERSION?=8a7c48209d7882a7ce79a6b306270e4703194543
 PARAKEET_REPO?=https://github.com/mudler/parakeet.cpp
 
 GOCMD?=go

backend/go/parakeet-cpp/batcher.go

@@ -0,0 +1,79 @@
+package main
+
+import "time"
+
+// batchRequest is one in-flight unary transcription waiting to be batched.
+// In production pcm/decoder are set; tag is an opaque marker used by tests.
+type batchRequest struct {
+	pcm     []float32
+	decoder int32
+	tag     string
+	reply   chan batchReply
+}
+
+// batchReply carries one per-item JSON object string (an element of the C-API's
+// JSON array) or an error back to the waiting handler goroutine.
+type batchReply struct {
+	json string
+	err  error
+}
+
+// batcher coalesces concurrent batchRequests into batched runBatch calls. A
+// single run() goroutine is the sole caller of runBatch, so runBatch (which in
+// production calls the thread-unsafe C engine) is never entered concurrently.
+type batcher struct {
+	submit   chan *batchRequest
+	maxSize  int
+	maxWait  time.Duration
+	runBatch func(reqs []*batchRequest) // must deliver a reply to every req
+}
+
+func newBatcher(maxSize int, maxWait time.Duration, runBatch func([]*batchRequest)) *batcher {
+	if maxSize < 1 {
+		maxSize = 1
+	}
+	return &batcher{
+		submit:   make(chan *batchRequest),
+		maxSize:  maxSize,
+		maxWait:  maxWait,
+		runBatch: runBatch,
+	}
+}
+
+// run is the dispatcher loop: accumulate submitted requests until either maxSize
+// is reached or maxWait elapses since the first queued request, then dispatch.
+// Exits when stop is closed (draining any partially-filled batch first).
+func (b *batcher) run(stop <-chan struct{}) {
+	for {
+		var first *batchRequest
+		select {
+		case first = <-b.submit:
+		case <-stop:
+			return
+		}
+		batch := []*batchRequest{first}
+
+		// maxSize==1 disables batching: dispatch immediately (passthrough).
+		if b.maxSize == 1 {
+			b.runBatch(batch)
+			continue
+		}
+
+		timer := time.NewTimer(b.maxWait)
+	fill:
+		for len(batch) < b.maxSize {
+			select {
+			case r := <-b.submit:
+				batch = append(batch, r)
+			case <-timer.C:
+				break fill
+			case <-stop:
+				timer.Stop()
+				b.runBatch(batch)
+				return
+			}
+		}
+		timer.Stop()
+		b.runBatch(batch)
+	}
+}

backend/go/parakeet-cpp/batcher_test.go

@@ -0,0 +1,108 @@
+package main
+
+import (
+	"sync"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("batcher", func() {
+	echoReply := func(reqs []*batchRequest) {
+		for _, r := range reqs {
+			r.reply <- batchReply{json: r.tag}
+		}
+	}
+
+	It("coalesces concurrent submits into batches", func() {
+		var mu sync.Mutex
+		var sizes []int
+		run := func(reqs []*batchRequest) {
+			mu.Lock()
+			sizes = append(sizes, len(reqs))
+			mu.Unlock()
+			echoReply(reqs)
+		}
+		b := newBatcher(4, 50*time.Millisecond, run)
+		stop := make(chan struct{})
+		go b.run(stop)
+		defer close(stop)
+
+		const N = 4
+		var wg sync.WaitGroup
+		got := make([]string, N)
+		for i := 0; i < N; i++ {
+			wg.Add(1)
+			go func(i int) {
+				defer wg.Done()
+				rep := make(chan batchReply, 1)
+				b.submit <- &batchRequest{tag: string(rune('a' + i)), reply: rep}
+				got[i] = (<-rep).json
+			}(i)
+		}
+		wg.Wait()
+
+		mu.Lock()
+		defer mu.Unlock()
+		total, maxBatch := 0, 0
+		for _, s := range sizes {
+			total += s
+			if s > maxBatch {
+				maxBatch = s
+			}
+		}
+		Expect(total).To(Equal(N))
+		Expect(maxBatch).To(BeNumerically(">=", 2), "expected at least one batch to coalesce >1 request")
+	})
+
+	It("dispatches when max size is reached", func() {
+		dispatched := make(chan int, 8)
+		run := func(reqs []*batchRequest) {
+			dispatched <- len(reqs)
+			echoReply(reqs)
+		}
+		b := newBatcher(2, time.Hour, run) // huge window: only size can trigger
+		stop := make(chan struct{})
+		go b.run(stop)
+		defer close(stop)
+		for i := 0; i < 2; i++ {
+			rep := make(chan batchReply, 1)
+			b.submit <- &batchRequest{tag: "x", reply: rep}
+			go func(rep chan batchReply) { <-rep }(rep)
+		}
+		Eventually(dispatched, "2s").Should(Receive(Equal(2)))
+	})
+
+	It("dispatches when the wait window elapses", func() {
+		dispatched := make(chan int, 8)
+		run := func(reqs []*batchRequest) {
+			dispatched <- len(reqs)
+			echoReply(reqs)
+		}
+		b := newBatcher(8, 20*time.Millisecond, run) // size unreachable; window fires
+		stop := make(chan struct{})
+		go b.run(stop)
+		defer close(stop)
+		rep := make(chan batchReply, 1)
+		b.submit <- &batchRequest{tag: "x", reply: rep}
+		go func() { <-rep }()
+		Eventually(dispatched, "2s").Should(Receive(Equal(1)))
+	})
+
+	It("bypasses batching when max size is 1", func() {
+		dispatched := make(chan int, 8)
+		run := func(reqs []*batchRequest) {
+			dispatched <- len(reqs)
+			echoReply(reqs)
+		}
+		b := newBatcher(1, time.Hour, run) // size 1 => immediate dispatch
+		stop := make(chan struct{})
+		go b.run(stop)
+		defer close(stop)
+		rep := make(chan batchReply, 1)
+		b.submit <- &batchRequest{tag: "x", reply: rep}
+		go func() { <-rep }()
+		Eventually(dispatched, "2s").Should(Receive(Equal(1)))
+	})
+})