Serialize OpenAL playback on a dedicated thread

gumble/gumbleopenal/stream.go

@@ -5,14 +5,15 @@ import (
 	"errors"
 	"math"
 	"os/exec"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"git.stormux.org/storm/barnard/audio"
-	"git.stormux.org/storm/barnard/log"
 	"git.stormux.org/storm/barnard/gumble/go-openal/openal"
 	"git.stormux.org/storm/barnard/gumble/gumble"
+	"git.stormux.org/storm/barnard/log"
 	"git.stormux.org/storm/barnard/noise"
 )
 
@@ -55,6 +56,11 @@ func beep() {
 	}
 }
 
+type renderCommand struct {
+	fn   func()
+	done chan struct{}
+}
+
 type Stream struct {
 	client *gumble.Client
 	link   gumble.Detacher
@@ -68,6 +74,8 @@ type Stream struct {
 
 	deviceSink  *openal.Device
 	contextSink *openal.Context
+	renderCh    chan renderCommand
+	renderDone  chan struct{}
 
 	noiseProcessor      NoiseProcessor
 	noiseProcessorRight NoiseProcessor
@@ -147,11 +155,40 @@ func New(client *gumble.Client, inputDevice *string, outputDevice *string, test
 		s.Destroy()
 		return nil, ErrOutputDevice
 	}
-	s.contextSink.Activate()
+	// OpenAL contexts are current to an OS thread. Move ownership to one
+	// dedicated render thread before any source or buffer is created.
+	openal.NullContext.Activate()
+	s.startRenderer()
 
 	return s, nil
 }
 
+func (s *Stream) startRenderer() {
+	s.renderCh = make(chan renderCommand)
+	s.renderDone = make(chan struct{})
+	ready := make(chan struct{})
+	go func() {
+		runtime.LockOSThread()
+		defer runtime.UnlockOSThread()
+		s.contextSink.Activate()
+		close(ready)
+		defer close(s.renderDone)
+		for command := range s.renderCh {
+			command.fn()
+			close(command.done)
+		}
+		openal.NullContext.Activate()
+	}()
+	<-ready
+}
+
+// render executes fn on the sole OS thread that owns the OpenAL context.
+func (s *Stream) render(fn func()) {
+	command := renderCommand{fn: fn, done: make(chan struct{})}
+	s.renderCh <- command
+	<-command.done
+}
+
 func (s *Stream) AttachStream(client *gumble.Client) {
 	s.link = client.Config.AttachAudio(s)
 }
@@ -169,6 +206,20 @@ func (s *Stream) GetFilePlayer() FilePlayer {
 	return s.filePlayer
 }
 
+// UpdateUserGain applies a user's current mute and volume state on the
+// renderer thread.
+func (s *Stream) UpdateUserGain(user *gumble.User) {
+	s.render(func() {
+		if source := user.AudioSource(); source != nil {
+			if user.LocallyMuted() {
+				source.SetGain(0)
+			} else {
+				source.SetGain(user.Volume())
+			}
+		}
+	})
+}
+
 // SetErrorFunc sets a callback that is invoked when the microphone
 // capture device fails to provide audio data.
 func (s *Stream) SetErrorFunc(f func(error)) {
@@ -197,9 +248,13 @@ func (s *Stream) Destroy() {
 		s.deviceSource = nil
 	}
 	if s.deviceSink != nil {
-		s.contextSink.Destroy()
+		if s.contextSink != nil {
+			s.render(func() { s.contextSink.Destroy() })
+			close(s.renderCh)
+			<-s.renderDone
+			s.contextSink = nil
+		}
 		s.deviceSink.CloseDevice()
-		s.contextSink = nil
 		s.deviceSink = nil
 	}
 }
@@ -257,32 +312,34 @@ func (s *Stream) SetMicVolume(change float32, relative bool) {
 func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 	go func(e *gumble.AudioStreamEvent) {
 		log.Info("audio stream started for user %s", e.User.Name)
-		var source = openal.NewSource()
-		e.User.SetAudioSource(&source)
-
-		// Set initial gain based on volume and mute state
-		if e.User.LocallyMuted() {
-			source.SetGain(0)
-		} else {
-			source.SetGain(e.User.Volume())
-		}
-
-		bufferCount := e.Client.Config.Buffers
-		if bufferCount < 64 {
-			bufferCount = 64
-		}
-		emptyBufs := openal.NewBuffers(bufferCount)
+		var source openal.Source
+		var emptyBufs openal.Buffers
+		var raw [maxBufferSize]byte
+		s.render(func() {
+			source = openal.NewSource()
+			e.User.SetAudioSource(&source)
+			if e.User.LocallyMuted() {
+				source.SetGain(0)
+			} else {
+				source.SetGain(e.User.Volume())
+			}
+			bufferCount := e.Client.Config.Buffers
+			if bufferCount < 64 {
+				bufferCount = 64
+			}
+			emptyBufs = openal.NewBuffers(bufferCount)
+		})
 
 		reclaim := func() {
-			if n := source.BuffersProcessed(); n > 0 {
-				reclaimedBufs := make(openal.Buffers, n)
-				source.UnqueueBuffers(reclaimedBufs)
-				emptyBufs = append(emptyBufs, reclaimedBufs...)
-			}
+			s.render(func() {
+				if n := source.BuffersProcessed(); n > 0 {
+					reclaimedBufs := make(openal.Buffers, n)
+					source.UnqueueBuffers(reclaimedBufs)
+					emptyBufs = append(emptyBufs, reclaimedBufs...)
+				}
+			})
 		}
 
-		var raw [maxBufferSize]byte
-
 		// Jitter buffer: collects incoming packets, reorders by
 		// sequence number, and releases them in order after a small
 		// initial delay (3-5 packets / 30-50ms).
@@ -354,7 +411,9 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 					break
 				}
 				reclaim()
-				emptyBufs = s.processAudioPacket(pkt, e.User, &source, emptyBufs, &raw)
+				s.render(func() {
+					emptyBufs = s.processAudioPacket(pkt, e.User, &source, emptyBufs, &raw)
+				})
 			}
 		}
 
@@ -368,12 +427,17 @@ func (s *Stream) OnAudioStream(e *gumble.AudioStreamEvent) {
 			}
 			if pkt != nil {
 				reclaim()
-				emptyBufs = s.processAudioPacket(pkt, e.User, &source, emptyBufs, &raw)
+				s.render(func() {
+					emptyBufs = s.processAudioPacket(pkt, e.User, &source, emptyBufs, &raw)
+				})
 			}
 		}
 		reclaim()
-		emptyBufs.Delete()
-		source.Delete()
+		s.render(func() {
+			emptyBufs.Delete()
+			source.Delete()
+			e.User.SetAudioSource(nil)
+		})
 		log.Debug("audio stream ended for user %s", e.User.Name)
 	}(e)
 }