Merge branch 'sleepct-pause' into v0.5.0

README.md

@@ -57,6 +57,8 @@ The bot requires the following permissions:
 * Voice Channel Speak
 * Voice Channel Use Voice Activity
 
+Permission integer 36768768.
+
 ### Finding Discord CID and GID
 
 Discord GID is a unique ID linked to one Discord Server, also called Guild. CID is similarly a unique ID for a Discord Channel. To find these you need to set Discord into developer Mode.

internal/bridge/discord.go

@@ -100,7 +100,7 @@ func (dd *DiscordDuplex) discordSendPCM(ctx context.Context, wg *sync.WaitGroup,
 		default:
 		}
 
-		promTimerDiscordSend.Observe(float64(sleepTick.SleepNextTarget()))
+		promTimerDiscordSend.Observe(float64(sleepTick.SleepNextTarget(true)))
 
 		if (len(pcm) > 1 && streaming) || (len(pcm) > dd.Bridge.BridgeConfig.DiscordStartStreamingCount && !streaming) {
 			if !streaming {
@@ -136,7 +136,7 @@ func (dd *DiscordDuplex) discordSendPCM(ctx context.Context, wg *sync.WaitGroup,
 				// We want to do this after alerting the user of possible short speaking cycles
 				for i := 0; i < 5; i++ {
 					internalSend(opusSilence)
-					promTimerDiscordSend.Observe(float64(sleepTick.SleepNextTarget()))
+					promTimerDiscordSend.Observe(float64(sleepTick.SleepNextTarget(true)))
 				}
 
 				dd.Bridge.DiscordVoice.Speaking(false)
@@ -235,7 +235,7 @@ func (dd *DiscordDuplex) discordReceivePCM(ctx context.Context, wg *sync.WaitGro
 		dd.fromDiscordMap[p.SSRC] = s
 		dd.discordMutex.Unlock()
 
-		p.PCM, err = s.decoder.Decode(p.Opus, deltaT*2, false)
+		p.PCM, err = s.decoder.Decode(p.Opus, deltaT, false)
 		if err != nil {
 			OnError("Error decoding opus data", err)
 			continue
@@ -285,7 +285,7 @@ func (dd *DiscordDuplex) fromDiscordMixer(ctx context.Context, wg *sync.WaitGrou
 		default:
 		}
 
-		promTimerDiscordMixer.Observe(float64(sleepTick.SleepNextTarget()))
+		promTimerDiscordMixer.Observe(float64(sleepTick.SleepNextTarget(true)))
 
 		dd.discordMutex.Lock()
 
@@ -364,7 +364,7 @@ func (dd *DiscordDuplex) fromDiscordMixer(ctx context.Context, wg *sync.WaitGrou
 
 			for i := 0; i < 5; i++ {
 				mumbleTimeoutSend(mumbleSilence)
-				promTimerDiscordMixer.Observe(float64(sleepTick.SleepNextTarget()))
+				promTimerDiscordMixer.Observe(float64(sleepTick.SleepNextTarget(false)))
 			}
 
 			toMumbleStreaming = false

internal/bridge/mumble.go

@@ -64,7 +64,7 @@ func (m MumbleDuplex) fromMumbleMixer(ctx context.Context, wg *sync.WaitGroup, t
 		default:
 		}
 
-		promTimerMumbleMixer.Observe(float64(sleepTick.SleepNextTarget()))
+		promTimerMumbleMixer.Observe(float64(sleepTick.SleepNextTarget(true)))
 
 		mutex.Lock()
 

pkg/sleepct/sleepct.go

@@ -2,19 +2,21 @@ package sleepct
 
 import (
 	"fmt"
-	"sync"
 	"time"
 )
 
-// SleepCT - Sleep constant time step crates a sleep based ticker
-// designed maintain a sleep/tick interval
+// SleepCT - Sleep constant time step crates a sleep based ticker.
+// designed maintain a consistent sleep/tick interval.
+// The sleeper can be paused waiting to be signaled from another go routine.
+// This allows for the pausing of loops that do not have work to complete
 type SleepCT struct {
-	sync.Mutex
-	d time.Duration // duration
+	d      time.Duration // desired duration between targets
 	t      time.Time     // last time target
+	resume chan bool
 }
 
 func (s *SleepCT) Start(d time.Duration) {
+	s.resume = make(chan bool, 1)
 	if s.t.IsZero() {
 		s.d = d
 		s.t = time.Now()
@@ -23,12 +25,15 @@ func (s *SleepCT) Start(d time.Duration) {
 	}
 }
 
-func (s *SleepCT) SleepNextTarget() int64 {
-	s.Lock()
+// Sleep to the next target duration.
+// If pause it set to true will sleep the duration and wait to be notified.
+// The notification channel will be cleared when the thread wakes.
+// SleepNextTarget should not be call more than once concurrently.
+func (s *SleepCT) SleepNextTarget(pause bool) int64 {
+	var last time.Time
 
 	now := time.Now()
 
-	var last time.Time
 	if s.t.IsZero() {
 		fmt.Println("SleepCT reset")
 		last = now.Add(-s.d)
@@ -36,17 +41,33 @@ func (s *SleepCT) SleepNextTarget() int64 {
 		last = s.t
 	}
 
-	// Next Target
+	// Sleep to Next Target
 	s.t = last.Add(s.d)
 
-	d := s.t.Sub(now)
+	d := time.Until(s.t)
 
 	time.Sleep(d)
 
-	// delta := now.Sub(s.t)
-	// fmt.Println("delta", delta, d, time.Since(s.t))
+	if pause {
+		// wait until resume
+		if len(s.resume) == 0 {
+			<-s.resume
+			// if we did pause set the last sleep target to now
+			last = time.Now()
+		}
+	}
 
-	s.Unlock()
+	// Drain the resume channel
+	select {
+	case <-s.resume:
+	default:
+	}
 
 	return now.Sub(s.t).Microseconds()
 }
+
+// Notify attempts to resume a paused sleeper.
+// It is safe to call notify from other processes and as often as desired.
+func (s *SleepCT) Notify() {
+	s.resume <- true
+}

test/timing_test.go

@@ -13,12 +13,12 @@ import (
 	"github.com/stieneee/tickerct"
 )
 
-const testCount int64 = 10000
+const testCount int64 = 1000
 const maxSleepInterval time.Duration = 15 * time.Millisecond
 const tickerInterval time.Duration = 10 * time.Millisecond
 const testDuration time.Duration = time.Duration(testCount * 10 * int64(time.Millisecond))
 
-func testTickerBaseCase(wg *sync.WaitGroup) {
+func testTickerBaseCase(wg *sync.WaitGroup, test *testing.T) {
 	wg.Add(1)
 	go func(interval time.Duration) {
 		now := time.Now()
@@ -39,7 +39,7 @@ func testTickerBaseCase(wg *sync.WaitGroup) {
 func TestTickerBaseCase(t *testing.T) {
 	wg := sync.WaitGroup{}
 
-	testTickerBaseCase(&wg)
+	testTickerBaseCase(&wg, t)
 
 	wg.Wait()
 }
@@ -115,7 +115,7 @@ func testSleepCT(wg *sync.WaitGroup) {
 			if i+1 < testCount {
 				time.Sleep(time.Duration(float64(maxSleepInterval) * rand.Float64()))
 			}
-			s.SleepNextTarget()
+			s.SleepNextTarget(false)
 		}
 		fmt.Println("SleepCT (loaded) after", testDuration, "drifts", time.Since(start)-testDuration)
 		wg.Done()
@@ -130,6 +130,35 @@ func TestSleepCT(t *testing.T) {
 	wg.Wait()
 }
 
+func testSleepCTPause(wg *sync.WaitGroup) {
+	wg.Add(1)
+	go func(interval time.Duration) {
+		now := time.Now()
+		start := now
+		// start the ticker
+		s := sleepct.SleepCT{}
+		s.Start(interval)
+		var i int64
+		for i = 0; i < testCount; i++ {
+			if i+1 < testCount {
+				time.Sleep(time.Duration(float64(maxSleepInterval) * rand.Float64()))
+			}
+			s.Notify()
+			s.SleepNextTarget(true)
+		}
+		fmt.Println("SleepCT Pause (loaded) after", testDuration, "drifts", time.Since(start)-testDuration)
+		wg.Done()
+	}(tickerInterval)
+}
+
+func TestSleepCTPause(t *testing.T) {
+	wg := sync.WaitGroup{}
+
+	testSleepCTPause(&wg)
+
+	wg.Wait()
+}
+
 func TestIdleJitter(t *testing.T) {
 	wg := sync.WaitGroup{}