Optimized RequestChan for Client Session (#1186)

* proxy: rename RequestChan.Len => Buffered

* proxy: set DefaultRequestChanBuffer=128

* proxy: replacing tasks channel with RequestChan, default buffer size=1024

* proxy: set default session_max_pipeline=10000

Author: spinlock

config/proxy.toml

@@ -79,7 +79,7 @@ session_send_timeout = "30s"
 
 # Make sure this is higher than the max number of requests for each pipeline request, or your client may be blocked.
 # Set session pipeline buffer size.
-session_max_pipeline = 1024
+session_max_pipeline = 10000
 
 # Set session tcp keepalive period. (0 to disable)
 session_keepalive_period = "75s"

pkg/proxy/config.go

@@ -95,7 +95,7 @@ session_send_timeout = "30s"
 
 # Make sure this is higher than the max number of requests for each pipeline request, or your client may be blocked.
 # Set session pipeline buffer size.
-session_max_pipeline = 1024
+session_max_pipeline = 10000
 
 # Set session tcp keepalive period. (0 to disable)
 session_keepalive_period = "75s"

pkg/proxy/request.go

@@ -8,7 +8,6 @@ import (
 	"unsafe"
 
 	"github.com/CodisLabs/codis/pkg/proxy/redis"
-	"github.com/CodisLabs/codis/pkg/utils/math2"
 	"github.com/CodisLabs/codis/pkg/utils/sync2/atomic2"
 )
 
@@ -68,15 +67,18 @@ type RequestChan struct {
 	closed bool
 }
 
-const MinRequestChanBuffer = 128
+const DefaultRequestChanBuffer = 128
 
 func NewRequestChan() *RequestChan {
-	return NewRequestChanBuffer(MinRequestChanBuffer)
+	return NewRequestChanBuffer(0)
 }
 
 func NewRequestChanBuffer(n int) *RequestChan {
+	if n <= 0 {
+		n = DefaultRequestChanBuffer
+	}
 	var ch = &RequestChan{
-		buff: make([]*Request, math2.MaxInt(n, MinRequestChanBuffer)),
+		buff: make([]*Request, n),
 	}
 	ch.cond = sync.NewCond(&ch.lock)
 	return ch
@@ -91,7 +93,7 @@ func (c *RequestChan) Close() {
 	c.lock.Unlock()
 }
 
-func (c *RequestChan) Len() int {
+func (c *RequestChan) Buffered() int {
 	c.lock.Lock()
 	n := len(c.data)
 	c.lock.Unlock()
@@ -139,7 +141,7 @@ func (c *RequestChan) lockedPopFront() (*Request, bool) {
 }
 
 func (c *RequestChan) IsEmpty() bool {
-	return c.Len() == 0
+	return c.Buffered() == 0
 }
 
 func (c *RequestChan) PopFrontAll(onRequest func(r *Request) error) error {

pkg/proxy/request_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 func TestRequestChan1(t *testing.T) {
-	var ch = NewRequestChanBuffer(512)
+	var ch = NewRequestChanBuffer(0)
 	for i := 0; i < 8192; i++ {
 		n := ch.PushBack(&Request{UnixNano: int64(i)})
 		assert.Must(n == i+1)
@@ -23,7 +23,7 @@ func TestRequestChan1(t *testing.T) {
 		r, ok := ch.PopFront()
 		assert.Must(ok && r.UnixNano == int64(i))
 	}
-	assert.Must(ch.Len() == 0)
+	assert.Must(ch.Buffered() == 0)
 
 	ch.Close()
 
@@ -39,13 +39,13 @@ func TestRequestChan2(t *testing.T) {
 	}
 	ch.Close()
 
-	assert.Must(ch.Len() == 8192)
+	assert.Must(ch.Buffered() == 8192)
 
 	for i := 0; i < 8192; i++ {
 		r, ok := ch.PopFront()
 		assert.Must(ok && r.UnixNano == int64(i))
 	}
-	assert.Must(ch.Len() == 0)
+	assert.Must(ch.Buffered() == 0)
 
 	_, ok := ch.PopFront()
 	assert.Must(!ok)
@@ -55,10 +55,12 @@ func TestRequestChan3(t *testing.T) {
 	var wg sync.WaitGroup
 	var ch = NewRequestChanBuffer(512)
 
+	const n = 1000 * 1000 * 4
+
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		for i := 0; i < 8192; i++ {
+		for i := 0; i < n; i++ {
 			ch.PushBack(&Request{UnixNano: int64(i)})
 			if i%1024 == 0 {
 				runtime.Gosched()
@@ -69,9 +71,12 @@ func TestRequestChan3(t *testing.T) {
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		for i := 0; i < 8192; i++ {
+		for i := 0; i < n; i++ {
 			r, ok := ch.PopFront()
 			assert.Must(ok && r.UnixNano == int64(i))
+			if i%4096 == 0 {
+				runtime.Gosched()
+			}
 		}
 	}()
 

pkg/proxy/session.go

@@ -15,7 +15,6 @@ import (
 	"github.com/CodisLabs/codis/pkg/proxy/redis"
 	"github.com/CodisLabs/codis/pkg/utils/errors"
 	"github.com/CodisLabs/codis/pkg/utils/log"
-	"github.com/CodisLabs/codis/pkg/utils/math2"
 	"github.com/CodisLabs/codis/pkg/utils/sync2/atomic2"
 )
 
@@ -104,9 +103,8 @@ func (s *Session) CloseWithError(err error) error {
 }
 
 var (
-	ErrTooManySessions = errors.New("too many sessions")
-	ErrRouterNotOnline = errors.New("router is not online")
-
+	ErrRouterNotOnline          = errors.New("router is not online")
+	ErrTooManySessions          = errors.New("too many sessions")
 	ErrTooManyPipelinedRequests = errors.New("too many pipelined requests")
 )
 
@@ -132,7 +130,7 @@ func (s *Session) Start(d *Router) {
 			return
 		}
 
-		tasks := make(chan *Request, math2.MaxInt(1, s.config.SessionMaxPipeline))
+		tasks := NewRequestChanBuffer(1024)
 
 		go func() {
 			s.loopWriter(tasks)
@@ -141,17 +139,20 @@ func (s *Session) Start(d *Router) {
 
 		go func() {
 			s.loopReader(tasks, d)
-			close(tasks)
+			tasks.Close()
 		}()
 	})
 }
 
-func (s *Session) loopReader(tasks chan<- *Request, d *Router) (err error) {
+func (s *Session) loopReader(tasks *RequestChan, d *Router) (err error) {
 	defer func() {
 		s.CloseReaderWithError(err)
 	}()
 
-	var sensitive = s.config.SessionBreakOnFailure
+	var (
+		breakOnFailure = s.config.SessionBreakOnFailure
+		maxPipelineLen = s.config.SessionMaxPipeline
+	)
 
 	for !s.quit {
 		multi, err := s.Conn.DecodeMultiBulk()
@@ -160,6 +161,10 @@ func (s *Session) loopReader(tasks chan<- *Request, d *Router) (err error) {
 		}
 		s.incrOpTotal()
 
+		if tasks.Buffered() > maxPipelineLen {
+			return ErrTooManyPipelinedRequests
+		}
+
 		start := time.Now()
 		s.LastOpUnix = start.Unix()
 		s.Ops++
@@ -170,59 +175,57 @@ func (s *Session) loopReader(tasks chan<- *Request, d *Router) (err error) {
 		r.Database = s.database
 		r.UnixNano = start.UnixNano()
 
-		if len(tasks) == cap(tasks) {
-			return ErrTooManyPipelinedRequests
-		}
 		if err := s.handleRequest(r, d); err != nil {
 			r.Resp = redis.NewErrorf("ERR handle request, %s", err)
-			tasks <- r
-			if sensitive {
+			tasks.PushBack(r)
+			if breakOnFailure {
 				return err
 			}
 		} else {
-			tasks <- r
+			tasks.PushBack(r)
 		}
 	}
 	return nil
 }
 
-func (s *Session) loopWriter(tasks <-chan *Request) (err error) {
+func (s *Session) loopWriter(tasks *RequestChan) (err error) {
 	defer func() {
 		s.CloseWithError(err)
-		for r := range tasks {
+		tasks.PopFrontAllVoid(func(r *Request) {
 			s.incrOpFails(r, nil)
-		}
+		})
 		s.flushOpStats(true)
 	}()
 
-	var sensitive = s.config.SessionBreakOnFailure
+	var breakOnFailure = s.config.SessionBreakOnFailure
 
 	p := s.Conn.FlushEncoder()
 	p.MaxInterval = time.Millisecond
 	p.MaxBuffered = 256
 
-	for r := range tasks {
+	return tasks.PopFrontAll(func(r *Request) error {
 		resp, err := s.handleResponse(r)
 		if err != nil {
 			resp = redis.NewErrorf("ERR handle response, %s", err)
-			if sensitive {
+			if breakOnFailure {
 				s.Conn.Encode(resp, true)
 				return s.incrOpFails(r, err)
 			}
 		}
 		if err := p.Encode(resp); err != nil {
 			return s.incrOpFails(r, err)
 		}
-		if err := p.Flush(len(tasks) == 0); err != nil {
+		fflush := tasks.IsEmpty()
+		if err := p.Flush(fflush); err != nil {
 			return s.incrOpFails(r, err)
 		} else {
 			s.incrOpStats(r, resp.Type)
 		}
-		if len(tasks) == 0 {
+		if fflush {
 			s.flushOpStats(false)
 		}
-	}
-	return nil
+		return nil
+	})
 }
 
 func (s *Session) handleResponse(r *Request) (*redis.Resp, error) {