optimize time.After

pingcap · Aug 9, 2024 · 31fc263 · 31fc263
1 parent 976ec2a
commit 31fc263
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 4 deletions.
diff --git a/utils/dynstream/dnynamic_stream_bech_test.go b/utils/dynstream/dnynamic_stream_bech_test.go
@@ -20,8 +20,9 @@ func runDynamicStream(pathCount int, eventCount int, times int) {
 	wg := &sync.WaitGroup{}
 	wg.Add(eventCount * pathCount)
 
+	total := &atomic.Int64{}
+
 	sendEvents := func(path Path, wg *sync.WaitGroup) {
-		total := &atomic.Int64{}
 		for i := 0; i < eventCount; i++ {
 			ds.In() <- &inc{times: times, n: total, done: wg, path: path}
 		}
@@ -32,6 +33,11 @@ func runDynamicStream(pathCount int, eventCount int, times int) {
 	}
 
 	wg.Wait()
+
+	if total.Load() != int64(pathCount*eventCount*times) {
+		panic("total != pathCount * eventCount * times")
+	}
+
 	ds.Close()
 }
 

diff --git a/utils/dynstream/dynamic_stream.go b/utils/dynstream/dynamic_stream.go
@@ -138,7 +138,7 @@ type DynamicStream[T Event, D any] struct {
 }
 
 func NewDynamicStreamDefault[T Event, D any](handler Handler[T, D]) *DynamicStream[T, D] {
-	streamCount := max(DefaultStreamCount, runtime.NumCPU()*2)
+	streamCount := max(DefaultStreamCount, runtime.NumCPU())
 	return NewDynamicStream[T, D](handler, DefaultSchedulerInterval, DefaultReportInterval, streamCount)
 }
 
@@ -499,6 +499,7 @@ func (d *DynamicStream[T, D]) scheduler() {
 	}
 
 	nextSchedule := time.Now().Add(d.schedulerInterval)
+	timerChan := time.After(time.Until(nextSchedule))
 Loop:
 	for {
 		select {
@@ -583,8 +584,9 @@ Loop:
 				continue
 			}
 			si.streamStat = stat
-		case <-time.After(time.Until(nextSchedule)):
+		case <-timerChan:
 			nextSchedule = time.Now().Add(d.schedulerInterval)
+			timerChan = time.After(time.Until(nextSchedule))
 			doSchedule(ruleType(scheduleRule.Next()), 0)
 		}
 	}

diff --git a/utils/dynstream/stream.go b/utils/dynstream/stream.go
@@ -230,6 +230,8 @@ func (s *stream[T, D]) reportStatLoop() {
 
 	lastReportTime := time.Now()
 	nextReportTime := lastReportTime.Add(s.reportInterval)
+	reportWait := time.After(time.Until(nextReportTime))
+
 	reportRound := nextReportRound.Add(1)
 
 	handleCount := 0
@@ -273,11 +275,12 @@ func (s *stream[T, D]) reportStatLoop() {
 
 		lastReportTime = time.Now()
 		nextReportTime = lastReportTime.Add(s.reportInterval)
+		reportWait = time.After(time.Until(nextReportTime))
 	}
 
 	for {
 		select {
-		case <-time.After(time.Until(nextReportTime)):
+		case <-reportWait:
 			reportStat()
 		case <-s.reportNow:
 			reportStat()