Use async audio converter

pkg/network/webrtc/webrtc.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/giongto35/cloud-game/v3/pkg/logger"
@@ -25,8 +24,6 @@ type Peer struct {
 	onMessage func(data []byte)
 }
 
-var samplePool sync.Pool
-
 var DefaultOfferAnswerOptions = webrtc.OfferAnswerOptions{ICETricklingSupported: true}
 var DefaultOfferOptions = webrtc.OfferOptions{OfferAnswerOptions: DefaultOfferAnswerOptions}
 var DefaultAnswerOptions = webrtc.AnswerOptions{OfferAnswerOptions: DefaultOfferAnswerOptions}
@@ -221,34 +218,19 @@ func (p *Peer) Channel(label string, conf *webrtc.DataChannelInit, onMessage fun
 }
 
 func (p *Peer) SendAudio(dat []byte, dur time.Duration) {
-	if err := p.send(dat, dur, p.a.WriteSample); err != nil {
+	if err := p.a.WriteSample(media.Sample{Data: dat, Duration: dur}); err != nil {
 		p.log.Error().Err(err).Send()
 	}
 }
 
 func (p *Peer) SendVideo(data []byte, dur time.Duration) {
-	if err := p.send(data, dur, p.v.WriteSample); err != nil {
+	if err := p.v.WriteSample(media.Sample{Data: data, Duration: dur}); err != nil {
 		p.log.Error().Err(err).Send()
 	}
 }
 
 func (p *Peer) SendData(data []byte) { _ = p.d.Send(data) }
 
-func (p *Peer) send(data []byte, duration time.Duration, fn func(media.Sample) error) error {
-	sample, _ := samplePool.Get().(*media.Sample)
-	if sample == nil {
-		sample = new(media.Sample)
-	}
-	sample.Data = data
-	sample.Duration = duration
-	err := fn(*sample)
-	if err != nil {
-		return err
-	}
-	samplePool.Put(sample)
-	return nil
-}
-
 func (p *Peer) Disconnect() {
 	if p.c == nil {
 		return

pkg/worker/media/gstreamer.go

@@ -97,6 +97,7 @@ var pixFmtToGst = map[uint32]string{
 	pixFmtBGRA:  "BGRA",
 	pixFmtRGB16: "RGB16",
 }
+var audioBufPool = sync.Pool{New: func() any { b := make([]byte, 4096); return &b }}
 var pixFmtCache = map[string]uint32{}
 
 func init() {
@@ -118,14 +119,13 @@ func init() {
 // Video encoding is done in a single goroutine to avoid races.
 // Audio is pulled from the appsink on GStreamer's own audio thread.
 //
-// Goroutines (3):
+// Goroutines (4):
 //   - video worker x1 (push+pull loop for video encoding)
-//   - bus messages x2 (one per pipeline, bus message logging)
+//   - audio push x1 (audio queue for GStreamer)
+//   - bus messages  x2 (one per pipeline, bus message logging)
 type GstMediaPipe struct {
 	a, v *pipe
 
-	onAudio func([]byte, time.Duration)
-
 	conf config.Encoder
 
 	pixFmt uint32
@@ -148,6 +148,9 @@ type GstMediaPipe struct {
 	kfi      int  // 0=GStreamer auto, >0=force keyframe every N frames
 	aSegSent bool // for Opusenc bug
 
+	audioCh chan []byte
+	onAudio func([]byte, time.Duration)
+
 	// used for reinit
 	videoCh   chan videoJob
 	videoDone chan struct{}
@@ -236,6 +239,8 @@ func (g *GstMediaPipe) initAudio() (err error) {
 		return
 	}
 	g.a.sink.SetCallbacks(&app.SinkCallbacks{NewSampleFunc: g.pullAudio})
+	g.audioCh = make(chan []byte, 2)
+	go g.pushAudio()
 	return g.a.pipeline.SetState(gst.StatePlaying)
 }
 
@@ -261,7 +266,7 @@ func (g *GstMediaPipe) initVideo() (err error) {
 
 	p := g.v
 	fmt := g.vidFmt
-	g.videoCh = make(chan videoJob, 3)
+	g.videoCh = make(chan videoJob, 1)
 	g.videoDone = make(chan struct{})
 	go g.videoWorker(p, fmt, g.videoCh, g.videoDone)
 
@@ -285,6 +290,9 @@ func (g *GstMediaPipe) Destroy() {
 
 	g.reinit.Store(true)
 
+	if g.audioCh != nil {
+		close(g.audioCh)
+	}
 	if g.videoCh != nil {
 		g.v.stop()
 		close(g.videoCh)
@@ -299,11 +307,32 @@ func (g *GstMediaPipe) Destroy() {
 
 func (g *GstMediaPipe) ProcessAudio(audio []byte, cb func([]byte, time.Duration)) {
 	g.onAudio = cb
-	if !g.aSegSent {
-		g.aSegSent = true
-		g.a.srcPad.PushEvent(gst.NewSegmentEvent(cachedSegment))
+	buf := audioBufPool.Get().(*[]byte)
+	if cap(*buf) < len(audio) {
+		*buf = make([]byte, len(audio))
+	}
+	*buf = (*buf)[:len(audio)]
+	copy(*buf, audio)
+	select {
+	case g.audioCh <- *buf:
+	default:
+		audioBufPool.Put(buf)
+	}
+}
+
+func (g *GstMediaPipe) pushAudio() {
+	for data := range g.audioCh {
+		if g.reinit.Load() || g.a == nil {
+			audioBufPool.Put(&data)
+			continue
+		}
+		if !g.aSegSent {
+			g.aSegSent = true
+			g.a.srcPad.PushEvent(gst.NewSegmentEvent(cachedSegment))
+		}
+		C.pushAudioBuf(g.a.src(), unsafe.Pointer(&data[0]), C.gsize(len(data)))
+		audioBufPool.Put(&data)
 	}
-	C.pushAudioBuf(g.a.src(), unsafe.Pointer(&audio[0]), C.gsize(len(audio)))
 }
 
 // pullAudio pulls audio buffers from the appsink when they are available.