MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2024-12-26 15:15:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

Keep larger merge buffers for RPC (#20654)

Browse Source 
Keep larger merge buffers

When sending large messages >1K, the merge buffer would continuously be reallocated.

This could happen on listings, where blocks are typically 4->8K.

Keep merge buffer of up to 256KB.

Benchmark with 4096b messages:

```
benchmark                                          old ns/op     new ns/op     delta
BenchmarkRequests/servers=2/bytes/par=32-32        8271          6360          -23.10%
BenchmarkRequests/servers=2/bytes/par=64-32        7840          4731          -39.66%
BenchmarkRequests/servers=2/bytes/par=128-32       7291          4740          -34.99%
BenchmarkRequests/servers=2/bytes/par=256-32       7095          4580          -35.45%
BenchmarkRequests/servers=2/bytes/par=512-32       6757          4584          -32.16%
BenchmarkRequests/servers=2/bytes/par=1024-32      6429          4453          -30.74%

benchmark                                          old bytes     new bytes     delta
BenchmarkRequests/servers=2/bytes/par=32-32        12090         821           -93.21%
BenchmarkRequests/servers=2/bytes/par=64-32        17423         820           -95.29%
BenchmarkRequests/servers=2/bytes/par=128-32       18493         822           -95.56%
BenchmarkRequests/servers=2/bytes/par=256-32       18892         821           -95.65%
BenchmarkRequests/servers=2/bytes/par=512-32       19064         826           -95.67%
BenchmarkRequests/servers=2/bytes/par=1024-32      19038         842           -95.58%
```
This commit is contained in:
Klaus Post 2024-11-16 09:18:48 -08:00 committed by GitHub
parent 267f0ecea2
commit e8a476ef5a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 11 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
internal/grid/connection.go
View File

@ -1104,7 +1104,6 @@ func (c *Connection) writeStream(ctx context.Context, conn net.Conn, cancel cont
defer ping.Stop() defer ping.Stop()
queue := make([][]byte, 0, maxMergeMessages) queue := make([][]byte, 0, maxMergeMessages)
merged := make([]byte, 0, writeBufferSize)
var queueSize int var queueSize int
var buf bytes.Buffer var buf bytes.Buffer
var wsw wsWriter var wsw wsWriter
@ -1132,7 +1131,7 @@ func (c *Connection) writeStream(ctx context.Context, conn net.Conn, cancel cont
} }
return false return false
} }
if buf.Cap() > writeBufferSize*4 { if buf.Cap() > writeBufferSize*8 {
// Reset buffer if it gets too big, so we don't keep it around. // Reset buffer if it gets too big, so we don't keep it around.
buf = bytes.Buffer{} buf = bytes.Buffer{}
} }
@ -1140,6 +1139,8 @@ func (c *Connection) writeStream(ctx context.Context, conn net.Conn, cancel cont
return true return true
} }
// Merge buffer to keep between calls
merged := make([]byte, 0, writeBufferSize)
for { for {
var toSend []byte var toSend []byte
select { select {
@ -1238,17 +1239,17 @@ func (c *Connection) writeStream(ctx context.Context, conn net.Conn, cancel cont
fmt.Println("Merging", len(queue), "messages") fmt.Println("Merging", len(queue), "messages")
} }
toSend = merged[:0] merged = merged[:0]
m := message{Op: OpMerged, Seq: uint32(len(queue))} m := message{Op: OpMerged, Seq: uint32(len(queue))}
var err error var err error
toSend, err = m.MarshalMsg(toSend) merged, err = m.MarshalMsg(merged)
if err != nil { if err != nil {
gridLogIf(ctx, fmt.Errorf("msg.MarshalMsg: %w", err)) gridLogIf(ctx, fmt.Errorf("msg.MarshalMsg: %w", err))
return return
} }
// Append as byte slices. // Append as byte slices.
for _, q := range queue { for _, q := range queue {
toSend = msgp.AppendBytes(toSend, q) merged = msgp.AppendBytes(merged, q)
PutByteBuffer(q) PutByteBuffer(q)
} }
queue = queue[:0] queue = queue[:0]
@ -1256,14 +1257,17 @@ func (c *Connection) writeStream(ctx context.Context, conn net.Conn, cancel cont
// Combine writes. // Combine writes.
// Consider avoiding buffer copy. // Consider avoiding buffer copy.
err = wsw.writeMessage(&buf, c.side, ws.OpBinary, toSend) err = wsw.writeMessage(&buf, c.side, ws.OpBinary, merged)
if err != nil { if err != nil {
if !xnet.IsNetworkOrHostDown(err, true) { if !xnet.IsNetworkOrHostDown(err, true) {
gridLogIf(ctx, fmt.Errorf("ws writeMessage: %w", err)) gridLogIf(ctx, fmt.Errorf("ws writeMessage: %w", err))
} }
return return
} }
if cap(merged) > writeBufferSize*8 {
// If we had to send an excessively large package, reset size.
merged = make([]byte, 0, writeBufferSize)
}
if !writeBuffer() { if !writeBuffer() {
return return
} }