opt: get as much data read from socket per loop as possible

Author: panjf2000

connection_unix.go

@@ -176,7 +176,7 @@ func (c *conn) ReadN(n int) (int, []byte) {
 		return inBufferLen, buf
 	}
 	buf, _ := c.inboundBuffer.Peek(n)
-	return n, buf
+	return len(buf), buf
 }
 
 func (c *conn) ShiftN(n int) int {

gnet.go

@@ -283,10 +283,13 @@ func Serve(eventHandler EventHandler, protoAddr string, opts ...Option) (err err
 			"while you are trying to set up %d\n", options.NumEventLoop)
 		return errors.ErrTooManyEventLoopThreads
 	}
-
-	if rbc := options.ReadBufferCap; rbc <= 0 {
+	rbc := options.ReadBufferCap
+	switch {
+	case rbc <= 0:
 		options.ReadBufferCap = ringbuffer.TCPReadBufferSize
-	} else {
+	case rbc <= ringbuffer.DefaultBufferSize:
+		options.ReadBufferCap = ringbuffer.DefaultBufferSize
+	default:
 		options.ReadBufferCap = toolkit.CeilToPowerOfTwo(rbc)
 		ringbuffer.TCPReadBufferSize = options.ReadBufferCap
 	}

pkg/ringbuffer/ring_buffer.go

@@ -27,7 +27,8 @@ import (
 )
 
 const (
-	defaultBufferSize   = 1024     // 1KB for the first-time allocation on ring-buffer.
+	// DefaultBufferSize is the first-time allocation on a ring-buffer.
+	DefaultBufferSize   = 1024     // 1KB
 	bufferGrowThreshold = 4 * 1024 // 4KB
 )
 
@@ -380,8 +381,8 @@ func (rb *RingBuffer) Reset() {
 
 func (rb *RingBuffer) grow(newCap int) {
 	if n := rb.size; n == 0 {
-		if newCap <= defaultBufferSize {
-			newCap = defaultBufferSize
+		if newCap <= DefaultBufferSize {
+			newCap = DefaultBufferSize
 		} else {
 			newCap = toolkit.CeilToPowerOfTwo(newCap)
 		}

pkg/ringbuffer/ring_buffer_test.go

@@ -127,9 +127,9 @@ func TestZeroRingBuffer(t *testing.T) {
 	assert.EqualValues(t, 0, rb.Free(), "expect free is 0")
 	buf := []byte(strings.Repeat("1234", 12))
 	_, _ = rb.Write(buf)
-	assert.EqualValuesf(t, defaultBufferSize, rb.Len(), "expect rb.Len()=%d, but got rb.Len()=%d", defaultBufferSize, rb.Len())
-	assert.EqualValuesf(t, defaultBufferSize, rb.Cap(), "expect rb.Cap()=%d, but got rb.Cap()=%d", defaultBufferSize, rb.Cap())
-	assert.Truef(t, rb.r == 0 && rb.w == 48 && rb.size == defaultBufferSize, "expect rb.r=0, rb.w=48, rb.size=64, rb.mask=63, but got rb.r=%d, rb.w=%d, rb.size=%d", rb.r, rb.w, rb.size)
+	assert.EqualValuesf(t, DefaultBufferSize, rb.Len(), "expect rb.Len()=%d, but got rb.Len()=%d", DefaultBufferSize, rb.Len())
+	assert.EqualValuesf(t, DefaultBufferSize, rb.Cap(), "expect rb.Cap()=%d, but got rb.Cap()=%d", DefaultBufferSize, rb.Cap())
+	assert.Truef(t, rb.r == 0 && rb.w == 48 && rb.size == DefaultBufferSize, "expect rb.r=0, rb.w=48, rb.size=64, rb.mask=63, but got rb.r=%d, rb.w=%d, rb.size=%d", rb.r, rb.w, rb.size)
 	assert.EqualValues(t, buf, rb.ByteBuffer().Bytes(), "expect it is equal")
 	rb.Discard(48)
 	assert.Truef(t, rb.IsEmpty() && rb.r == 0 && rb.w == 0, "expect rb is empty and rb.r=rb.w=0, but got rb.r=%d and rb.w=%d", rb.r, rb.w)
@@ -140,29 +140,29 @@ func TestRingBufferGrow(t *testing.T) {
 	head, tail := rb.Peek(2)
 	assert.Empty(t, head, "head should be empty")
 	assert.Empty(t, tail, "tail should be empty")
-	data := make([]byte, defaultBufferSize+1)
+	data := make([]byte, DefaultBufferSize+1)
 	n, err := rand.Read(data)
 	assert.NoError(t, err, "failed to generate random data")
-	assert.EqualValuesf(t, defaultBufferSize+1, n, "expect random data length is %d but got %d", defaultBufferSize+1, n)
+	assert.EqualValuesf(t, DefaultBufferSize+1, n, "expect random data length is %d but got %d", DefaultBufferSize+1, n)
 	n, err = rb.Write(data)
 	assert.NoError(t, err)
-	assert.EqualValues(t, defaultBufferSize+1, n)
-	assert.EqualValues(t, 2*defaultBufferSize, rb.Cap())
-	assert.EqualValues(t, 2*defaultBufferSize, rb.Len())
-	assert.EqualValues(t, defaultBufferSize+1, rb.Length())
-	assert.EqualValues(t, defaultBufferSize-1, rb.Free())
+	assert.EqualValues(t, DefaultBufferSize+1, n)
+	assert.EqualValues(t, 2*DefaultBufferSize, rb.Cap())
+	assert.EqualValues(t, 2*DefaultBufferSize, rb.Len())
+	assert.EqualValues(t, DefaultBufferSize+1, rb.Length())
+	assert.EqualValues(t, DefaultBufferSize-1, rb.Free())
 	assert.EqualValues(t, data, rb.ByteBuffer().Bytes())
 
-	rb = New(defaultBufferSize)
+	rb = New(DefaultBufferSize)
 	newData := make([]byte, 3*512)
 	n, err = rand.Read(newData)
 	assert.NoError(t, err, "failed to generate random data")
 	assert.EqualValuesf(t, 3*512, n, "expect random data length is %d but got %d", 3*512, n)
 	n, err = rb.Write(newData)
 	assert.NoError(t, err)
 	assert.EqualValues(t, 3*512, n)
-	assert.EqualValues(t, 2*defaultBufferSize, rb.Cap())
-	assert.EqualValues(t, 2*defaultBufferSize, rb.Len())
+	assert.EqualValues(t, 2*DefaultBufferSize, rb.Cap())
+	assert.EqualValues(t, 2*DefaultBufferSize, rb.Len())
 	assert.EqualValues(t, 3*512, rb.Length())
 	assert.EqualValues(t, 512, rb.Free())
 	assert.EqualValues(t, newData, rb.ByteBuffer().Bytes())

pkg/ringbuffer/ring_buffer_unix.go

@@ -35,20 +35,27 @@ func (rb *RingBuffer) CopyFromSocket(fd int) (n int, err error) {
 }
 
 // Rewind moves the data from its tail to head and rewind its pointers of read and write.
-func (rb *RingBuffer) Rewind() int {
+func (rb *RingBuffer) Rewind() (n int) {
 	if rb.IsEmpty() {
 		rb.Reset()
-		return 0
+		return
 	}
-	if rb.w != 0 {
-		return 0
-	}
-	if rb.r < rb.size-rb.r {
-		rb.grow(rb.size + rb.size/2)
-		return rb.size - rb.r
+	if rb.w == 0 {
+		if rb.r < rb.size-rb.r {
+			rb.grow(rb.size + rb.size - rb.r)
+			return rb.size - rb.r
+		}
+		n = copy(rb.buf, rb.buf[rb.r:])
+		rb.r = 0
+		rb.w = n
+	} else if rb.size-rb.w < DefaultBufferSize {
+		if rb.r < rb.w-rb.r {
+			rb.grow(rb.size + rb.w - rb.r)
+			return rb.w - rb.r
+		}
+		n = copy(rb.buf, rb.buf[rb.r:rb.w])
+		rb.r = 0
+		rb.w = n
 	}
-	n := copy(rb.buf, rb.buf[rb.r:])
-	rb.r = 0
-	rb.w = n
-	return n
+	return
 }