perf: improve user specified flush delay automatically by statistics

Author: rueian

pipe.go

@@ -281,35 +281,68 @@ func (p *pipe) _background() {
 	atomic.StoreInt32(&p.state, 4)
 }
 
+type flush struct {
+	works time.Duration
+	waits time.Duration
+	bytes int
+}
+
+const flushesKeep = 16
+const flushesMask = flushesKeep - 1
+
 func (p *pipe) _backgroundWrite() (err error) {
 	var (
 		ones  = make([]cmds.Completed, 1)
 		multi []cmds.Completed
 		ch    chan RedisResult
 
-		flushDelay = p.maxFlushDelay
-		flushStart = time.Time{}
+		flushes [flushesKeep]flush
+		flushId = 0
+		bytes   = 1 // avoid zero
+		works   = time.Duration(0)
+		waits   = time.Duration(0)
+
+		userDelay = p.maxFlushDelay
 	)
 
+	var ts1 = time.Now()
 	for atomic.LoadInt32(&p.state) < 3 {
 		if ones[0], multi, ch = p.queue.NextWriteCmd(); ch == nil {
-			if flushDelay != 0 {
-				flushStart = time.Now()
-			}
-			if p.w.Buffered() == 0 {
+			buf := p.w.Buffered()
+			if buf == 0 {
 				err = p.Error()
+			} else if userDelay == 0 {
+				err = p.w.Flush()
 			} else {
+				ts2 := time.Now()
 				err = p.w.Flush()
+				dur, gap := time.Since(ts2), ts2.Sub(ts1)
+				bytes = bytes + buf - flushes[flushId].bytes
+				works = works + dur - flushes[flushId].works
+				waits = waits + gap - flushes[flushId].waits
+				flushes[flushId] = flush{bytes: buf, works: dur, waits: gap}
+				flushId = (flushId + 1) & flushesMask
+				ts1 = ts2
 			}
+
 			if err == nil {
 				if atomic.LoadInt32(&p.state) == 1 {
 					ones[0], multi, ch = p.queue.WaitForWrite()
 				} else {
 					runtime.Gosched()
 					continue
 				}
-				if flushDelay != 0 && atomic.LoadInt32(&p.waits) > 1 { // do not delay for sequential usage
-					time.Sleep(flushDelay - time.Since(flushStart)) // ref: https://github.com/rueian/rueidis/issues/156
+				if userDelay != 0 && atomic.LoadInt32(&p.waits) > 1 { // do not delay for sequential usage
+					byteWork := works / time.Duration(bytes)
+					byteWait := waits / time.Duration(bytes)
+					byteCost := byteWait + byteWork
+					capDelay := userDelay
+					if byteCost != 0 {
+						if delay := byteWait * (works / flushesKeep) / byteCost; delay < capDelay {
+							capDelay = delay
+						}
+					}
+					time.Sleep(capDelay) // ref: https://github.com/rueian/rueidis/issues/156
 				}
 			}
 		}