diff --git a/pkg/network/webrtc/webrtc.go b/pkg/network/webrtc/webrtc.go index b342b6b5..5f5abdbf 100644 --- a/pkg/network/webrtc/webrtc.go +++ b/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 diff --git a/pkg/worker/media/gstreamer.go b/pkg/worker/media/gstreamer.go index 4a6171fd..c5fe8ec5 100644 --- a/pkg/worker/media/gstreamer.go +++ b/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.