proxy: optimize FlushEncoder for better performance

Author: spinlock

config/proxy.toml

@@ -60,7 +60,7 @@ backend_send_bufsize = "128kb"
 backend_send_timeout = "30s"
 
 # Set backend pipeline buffer size.
-backend_max_pipeline = 1024
+backend_max_pipeline = 20480
 
 # Set backend never read replica groups, default is false
 backend_primary_only = false

pkg/proxy/backend.go

@@ -337,7 +337,7 @@ func (bc *BackendConn) loopWriter(round int) (err error) {
 
 	p := c.FlushEncoder()
 	p.MaxInterval = time.Millisecond
-	p.MaxBuffered = math2.MinInt(256, cap(tasks))
+	p.MaxBuffered = cap(tasks) / 2
 
 	for r := range bc.input {
 		if r.IsReadOnly() && r.IsBroken() {

pkg/proxy/config.go

@@ -76,7 +76,7 @@ backend_send_bufsize = "128kb"
 backend_send_timeout = "30s"
 
 # Set backend pipeline buffer size.
-backend_max_pipeline = 1024
+backend_max_pipeline = 20480
 
 # Set backend never read replica groups, default is false
 backend_primary_only = false

pkg/proxy/session.go

@@ -197,11 +197,14 @@ func (s *Session) loopWriter(tasks *RequestChan) (err error) {
 		s.flushOpStats(true)
 	}()
 
-	var breakOnFailure = s.config.SessionBreakOnFailure
+	var (
+		breakOnFailure = s.config.SessionBreakOnFailure
+		maxPipelineLen = s.config.SessionMaxPipeline
+	)
 
 	p := s.Conn.FlushEncoder()
 	p.MaxInterval = time.Millisecond
-	p.MaxBuffered = 256
+	p.MaxBuffered = maxPipelineLen / 2
 
 	return tasks.PopFrontAll(func(r *Request) error {
 		resp, err := s.handleResponse(r)