feat: bytebuferpoll use mcache replace

pkg/app/server/hertz_test.go

@@ -26,6 +26,7 @@ import (
 	"io/ioutil"
 	"net"
 	"net/http"
+	"runtime"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -248,6 +249,56 @@ func TestServer_Run(t *testing.T) {
 	_ = hertz.Shutdown(ctx)
 }
 
+func TestBigBodyBug(t *testing.T) {
+	runtime.GOMAXPROCS(3)
+	hertz := New(WithHostPorts("127.0.0.1:8888"))
+	hertz.GET("/test1", func(c context.Context, ctx *app.RequestContext) {
+		body := make([]byte, 1024*1024*9)
+		ctx.SetBodyString(string(body))
+	})
+	hertz.GET("/test2", func(c context.Context, ctx *app.RequestContext) {
+		body := make([]byte, 1024)
+		ctx.SetBodyString(string(body))
+	})
+	hertz.GET("/test3", func(c context.Context, ctx *app.RequestContext) {
+		body := make([]byte, 1024*2)
+		ctx.SetBodyString(string(body))
+	})
+	go hertz.Run()
+
+	go func() {
+		for i := 0; i < 5; i++ {
+			go func() {
+				for true {
+					http.Get("http://127.0.0.1:8888/test1")
+				}
+			}()
+		}
+	}()
+
+	go func() {
+		for i := 0; i < 10; i++ {
+			go func() {
+				for true {
+					http.Get("http://127.0.0.1:8888/test2")
+				}
+			}()
+		}
+	}()
+
+	go func() {
+		for i := 0; i < 10; i++ {
+			go func() {
+				for true {
+					http.Get("http://127.0.0.1:8888/test3")
+				}
+			}()
+		}
+	}()
+
+	<-make(chan struct{})
+}
+
 func TestNotAbsolutePath(t *testing.T) {
 	engine := New(WithHostPorts("127.0.0.1:9990"))
 	engine.POST("/", func(c context.Context, ctx *app.RequestContext) {

pkg/common/bytebufferpool/pool.go

@@ -45,6 +45,8 @@ import (
 	"sort"
 	"sync"
 	"sync/atomic"
+
+	"github.com/bytedance/gopkg/lang/mcache"
 )
 
 const (
@@ -54,7 +56,7 @@ const (
 	minSize = 1 << minBitSize
 	maxSize = 1 << (minBitSize + steps - 1)
 
-	calibrateCallsThreshold = 42000
+	calibrateCallsThreshold = 10
 	maxPercentile           = 0.95
 )
 
@@ -96,6 +98,16 @@ func (p *Pool) Get() *ByteBuffer {
 	}
 }
 
+func (p *Pool) GetWithSize(size int) *ByteBuffer {
+	return &ByteBuffer{
+		B: mcache.Malloc(size)[:0],
+	}
+}
+
+func (p *Pool) PutWithByte(b []byte) {
+	mcache.Free(b)
+}
+
 // Put returns byte buffer to the pool.
 //
 // ByteBuffer.B mustn't be touched after returning it to the pool.

pkg/common/config/option.go

@@ -37,7 +37,7 @@ const (
 	defaultAddr               = ":8888"
 	defaultNetwork            = "tcp"
 	defaultBasePath           = "/"
-	defaultMaxRequestBodySize = 4 * 1024 * 1024
+	defaultMaxRequestBodySize = 0
 	defaultWaitExitTimeout    = time.Second * 5
 	defaultReadBufferSize     = 4 * 1024
 )

pkg/protocol/response.go

@@ -42,6 +42,7 @@
 package protocol
 
 import (
+	"fmt"
 	"io"
 	"net"
 	"sync"
@@ -139,8 +140,9 @@ func (resp *Response) SetConnectionClose() {
 
 // SetBodyString sets response body.
 func (resp *Response) SetBodyString(body string) {
-	resp.CloseBodyStream()            //nolint:errcheck
-	resp.BodyBuffer().SetString(body) //nolint:errcheck
+	resp.CloseBodyStream() //nolint:errcheck
+	// resp.BodyBuffer().SetString(body) //nolint:errcheck
+	resp.BodyBufferWithSize(len(body)).SetString(body) //nolint:errcheck
 }
 
 func (resp *Response) ConstructBodyStream(body *bytebufferpool.ByteBuffer, bodyStream io.Reader) {
@@ -299,7 +301,8 @@ func (resp *Response) ResetBody() {
 			resp.body.Reset()
 			return
 		}
-		responseBodyPool.Put(resp.body)
+		// responseBodyPool.Put(resp.body)
+		responseBodyPool.PutWithByte(resp.body.B)
 		resp.body = nil
 	}
 }
@@ -385,11 +388,21 @@ func (resp *Response) CloseBodyStream() error {
 	return err
 }
 
+func (resp *Response) BodyBufferWithSize(size int) *bytebufferpool.ByteBuffer {
+	if resp.body == nil {
+		resp.body = responseBodyPool.GetWithSize(size)
+	}
+	resp.bodyRaw = nil
+	fmt.Printf("flipped resp Body len=%d cap=%d\n", resp.body.Len(), cap(resp.body.B))
+	return resp.body
+}
+
 func (resp *Response) BodyBuffer() *bytebufferpool.ByteBuffer {
 	if resp.body == nil {
 		resp.body = responseBodyPool.Get()
 	}
 	resp.bodyRaw = nil
+	// fmt.Printf("flipped resp Body len=%d cap=%d\n", resp.body.Len(), cap(resp.body.B))
 	return resp.body
 }