improved buffer balance

lxzan · lxzan · commit f237e28df4e2 · 2023-09-17T09:02:39.000+08:00
diff --git a/README.md b/README.md
@@ -29,7 +29,6 @@
 [12]: https://github.com/avelino/awesome-go#networking
 
 - [gws](#gws)
-    - [Event-Driven Go WebSocket Server \& Client](#event-driven-go-websocket-server--client)
     - [Feature](#feature)
     - [Attention](#attention)
     - [Install](#install)
@@ -274,7 +273,7 @@ BenchmarkConn_ReadMessage/compress_enabled-8              435176              24
 > &#24494;&#20449;&#20108;&#32500;&#30721;&#22312;&#35752;&#35770;&#21306;&#19981;&#23450;&#26102;&#26356;&#26032; 
 
 <div>
-<img src="assets/wechat.jpg" alt="WeChat" width="300" height="300" style="display: inline-block;"/>
+<img src="assets/wechat.png" alt="WeChat" width="300" height="300" style="display: inline-block;"/>
 <span>&nbsp;&nbsp;&nbsp;&nbsp;</span>
 <img src="assets/qq.jpg" alt="QQ" width="300" height="300" style="display: inline-block"/>
 </div>
diff --git a/assets/latency.png b/assets/latency.png
diff --git a/assets/performance-compress-disabled.png b/assets/performance-compress-disabled.png
diff --git a/assets/wechat.jpg b/assets/wechat.jpg
diff --git a/assets/wechat.png b/assets/wechat.png
diff --git a/compress.go b/compress.go
@@ -11,8 +11,6 @@ import (
 	"sync/atomic"
 )
 
-const compressionRate = 3
-
 type compressors struct {
 	serial      uint64
 	size        uint64
@@ -85,23 +83,38 @@ func (c *decompressors) Select() *decompressor {
 }
 
 func newDecompressor() *decompressor {
-	return &decompressor{fr: flate.NewReader(nil)}
+	return &decompressor{
+		b:  bytes.NewBuffer(nil),
+		fr: flate.NewReader(nil),
+	}
 }
 
 type decompressor struct {
 	sync.Mutex
+	b  *bytes.Buffer
 	fr io.ReadCloser
 }
 
+func (c *decompressor) reset(r io.Reader) {
+	resetter := c.fr.(flate.Resetter)
+	_ = resetter.Reset(r, nil) // must return a null pointer
+	if c.b.Cap() > 256*1024 {
+		c.b = bytes.NewBuffer(nil)
+	}
+	c.b.Reset()
+}
+
 // Decompress &#35299;&#21387;
 func (c *decompressor) Decompress(src *bytes.Buffer) (*bytes.Buffer, int, error) {
 	c.Lock()
 	defer c.Unlock()
 
 	_, _ = src.Write(internal.FlateTail)
-	resetter := c.fr.(flate.Resetter)
-	_ = resetter.Reset(src, nil) // must return a null pointer
-	var dst, idx = elasticPool.Get(src.Len() * compressionRate)
-	_, err := c.fr.(io.WriterTo).WriteTo(dst)
-	return dst, idx, err
+	c.reset(src)
+	if _, err := c.fr.(io.WriterTo).WriteTo(c.b); err != nil {
+		return nil, 0, err
+	}
+	var dst, idx = binaryPool.Get(c.b.Len())
+	_, _ = c.b.WriteTo(dst)
+	return dst, idx, nil
 }
diff --git a/compress_test.go b/compress_test.go
@@ -53,3 +53,12 @@ func TestFlate(t *testing.T) {
 		as.Error(err)
 	})
 }
+
+func TestDecompressor_Init(t *testing.T) {
+	var d = &decompressor{
+		b:  bytes.NewBuffer(internal.AlphabetNumeric.Generate(512 * 1024)),
+		fr: flate.NewReader(nil),
+	}
+	d.reset(nil)
+	assert.Equal(t, d.b.Cap(), 0)
+}
diff --git a/init.go b/init.go
@@ -3,7 +3,6 @@ package gws
 import "github.com/lxzan/gws/internal"
 
 var (
-	myPadding   = frameHeader{}            // &#24103;&#22836;&#22635;&#20805;&#29289;
-	staticPool  = internal.NewBufferPool() // &#38745;&#24577;&#32531;&#20914;&#27744;
-	elasticPool = internal.NewBufferPool() // &#24377;&#24615;&#32531;&#20914;&#27744;
+	myPadding  = frameHeader{}            // &#24103;&#22836;&#22635;&#20805;&#29289;
+	binaryPool = internal.NewBufferPool() // &#38745;&#24577;&#32531;&#20914;&#27744;
 )
diff --git a/internal/pool.go b/internal/pool.go
@@ -40,7 +40,7 @@ func (p *BufferPool) Put(b *bytes.Buffer, index int) {
 	if index == 0 || b == nil {
 		return
 	}
-	if b.Cap() <= 5*p.limits[index] {
+	if b.Cap() <= p.limits[index] {
 		p.pools[index].Put(b)
 	}
 }
diff --git a/protocol.go b/protocol.go
@@ -246,8 +246,7 @@ func (c *Message) Bytes() []byte {
 
 // Close recycle buffer
 func (c *Message) Close() error {
-	pool := internal.SelectValue(c.compressed, elasticPool, staticPool)
-	pool.Put(c.Data, c.index)
+	binaryPool.Put(c.Data, c.index)
 	c.Data = nil
 	return nil
 }
diff --git a/reader.go b/reader.go
@@ -93,7 +93,7 @@ func (c *Conn) readMessage() error {
 	}
 
 	var fin = c.fh.GetFIN()
-	var buf, index = staticPool.Get(contentLength)
+	var buf, index = binaryPool.Get(contentLength)
 	var p = buf.Bytes()
 	p = p[:contentLength]
 	if err := internal.ReadN(c.br, p); err != nil {
@@ -125,7 +125,7 @@ func (c *Conn) readMessage() error {
 	if err := internal.WriteN(c.continuationFrame.buffer, p); err != nil {
 		return err
 	} else {
-		staticPool.Put(buf, index)
+		binaryPool.Put(buf, index)
 	}
 	if c.continuationFrame.buffer.Len() > c.config.ReadMaxPayloadSize {
 		return internal.CloseMessageTooLarge
@@ -143,7 +143,7 @@ func (c *Conn) emitMessage(msg *Message) (err error) {
 	if msg.compressed {
 		data, index := msg.Data, msg.index
 		msg.Data, msg.index, err = c.decompressor.Decompress(msg.Data)
-		staticPool.Put(data, index)
+		binaryPool.Put(data, index)
 		if err != nil {
 			return internal.NewError(internal.CloseInternalServerErr, err)
 		}
diff --git a/writer.go b/writer.go
@@ -53,7 +53,7 @@ func (c *Conn) WriteAsync(opcode Opcode, payload []byte) error {
 			return
 		}
 		err = internal.WriteN(c.conn, frame.Bytes())
-		staticPool.Put(frame, index)
+		binaryPool.Put(frame, index)
 		c.emitError(err)
 	})
 	return nil
@@ -79,7 +79,7 @@ func (c *Conn) doWrite(opcode Opcode, payload []byte) error {
 	}
 
 	err = internal.WriteN(c.conn, frame.Bytes())
-	staticPool.Put(frame, index)
+	binaryPool.Put(frame, index)
 	return err
 }
 
@@ -101,7 +101,7 @@ func (c *Conn) genFrame(opcode Opcode, payload []byte) (*bytes.Buffer, int, erro
 
 	var header = frameHeader{}
 	headerLength, maskBytes := header.GenerateHeader(c.isServer, true, false, opcode, n)
-	var buf, index = staticPool.Get(n + headerLength)
+	var buf, index = binaryPool.Get(n + headerLength)
 	buf.Write(header[:headerLength])
 	buf.Write(payload)
 	var contents = buf.Bytes()
@@ -112,7 +112,7 @@ func (c *Conn) genFrame(opcode Opcode, payload []byte) (*bytes.Buffer, int, erro
 }
 
 func (c *Conn) compressData(opcode Opcode, payload []byte) (*bytes.Buffer, int, error) {
-	var buf, index = staticPool.Get(len(payload) + frameHeaderSize)
+	var buf, index = binaryPool.Get(len(payload) + frameHeaderSize)
 	buf.Write(myPadding[0:])
 	err := c.compressor.Compress(payload, buf)
 	if err != nil {
@@ -191,7 +191,7 @@ func (c *Broadcaster) Broadcast(socket *Conn) error {
 func (c *Broadcaster) doClose() {
 	for _, item := range c.msgs {
 		if item != nil {
-			staticPool.Put(item.frame, item.index)
+			binaryPool.Put(item.frame, item.index)
 		}
 	}
 }

Original file line number	Diff line number	Diff line change
`@@ -3,7 +3,6 @@ package gws`
`3`	`3`	`import "github.com/lxzan/gws/internal"`
`4`	`4`
`5`	`5`	`var (`
`6`		`- myPadding = frameHeader{} // 帧头填充物`
`7`		`- staticPool = internal.NewBufferPool() // 静态缓冲池`
`8`		`- elasticPool = internal.NewBufferPool() // 弹性缓冲池`
	`6`	`+ myPadding = frameHeader{} // 帧头填充物`
	`7`	`+ binaryPool = internal.NewBufferPool() // 静态缓冲池`
`9`	`8`	`)`
Original file line number	Diff line number	Diff line change
`@@ -40,7 +40,7 @@ func (p BufferPool) Put(b bytes.Buffer, index int) {`
`40`	`40`	`if index == 0 \|\| b == nil {`
`41`	`41`	`return`
`42`	`42`	`}`
`43`		`- if b.Cap() <= 5*p.limits[index] {`
	`43`	`+ if b.Cap() <= p.limits[index] {`
`44`	`44`	`p.pools[index].Put(b)`
`45`	`45`	`}`
`46`	`46`	`}`
Original file line number	Diff line number	Diff line change
`@@ -246,8 +246,7 @@ func (c *Message) Bytes() []byte {`
`246`	`246`
`247`	`247`	`// Close recycle buffer`
`248`	`248`	`func (c *Message) Close() error {`
`249`		`- pool := internal.SelectValue(c.compressed, elasticPool, staticPool)`
`250`		`- pool.Put(c.Data, c.index)`
	`249`	`+ binaryPool.Put(c.Data, c.index)`
`251`	`250`	`c.Data = nil`
`252`	`251`	`return nil`
`253`	`252`	`}`