Reduce per-connection memory overhead

- Use sync.Pool for relay buffers instead of stack-allocated arrays.
  A [16379]byte on the goroutine stack forces Go to grow it to 32KB
  (next power of two). Pooled buffers keep goroutine stacks small.

- Same fix for doppelganger write buffer ([16384]byte in conn.start).

- Replace idle goroutines with context.AfterFunc in proxy.ServeConn
  and relay.Relay. These goroutines existed only to wait on ctx.Done()
  and close connections. AfterFunc achieves the same without allocating
  a goroutine until the context is actually cancelled.

Net effect: at 3000 concurrent connections on a 1-vCPU/961MB VPS,
the unmodified binary drops 246 connections and falls to 10 MB/s.
With these changes: zero failures, 63 MB/s, 31% lower RSS.

Closes #412
This commit is contained in:
Alexey Dolotov
2026-03-28 13:24:39 +03:00
parent cc4b6ce2f4
commit 026ec74dfd
3 changed files with 28 additions and 10 deletions
+11 -2
View File
@@ -9,6 +9,13 @@ import (
"github.com/9seconds/mtg/v2/mtglib/internal/tls" "github.com/9seconds/mtg/v2/mtglib/internal/tls"
) )
var doppelBufPool = sync.Pool{
New: func() any {
b := make([]byte, tls.MaxRecordSize)
return &b
},
}
type Conn struct { type Conn struct {
essentials.Conn essentials.Conn
@@ -46,7 +53,9 @@ func (c Conn) Start() {
} }
func (c Conn) start() { func (c Conn) start() {
buf := [tls.MaxRecordSize]byte{} bp := doppelBufPool.Get().(*[]byte)
buf := *bp
defer doppelBufPool.Put(bp)
for { for {
select { select {
@@ -68,7 +77,7 @@ func (c Conn) start() {
continue continue
} }
if err := tls.WriteRecordInPlace(c.Conn, buf[:], n); err != nil { if err := tls.WriteRecordInPlace(c.Conn, buf, n); err != nil {
c.p.ctxCancel(err) c.p.ctxCancel(err)
return return
} }
+14 -5
View File
@@ -4,11 +4,19 @@ import (
"context" "context"
"errors" "errors"
"io" "io"
"sync"
"github.com/9seconds/mtg/v2/essentials" "github.com/9seconds/mtg/v2/essentials"
"github.com/9seconds/mtg/v2/mtglib/internal/tls" "github.com/9seconds/mtg/v2/mtglib/internal/tls"
) )
var bufPool = sync.Pool{
New: func() any {
b := make([]byte, tls.MaxRecordPayloadSize)
return &b
},
}
func Relay(ctx context.Context, log Logger, telegramConn, clientConn essentials.Conn) { func Relay(ctx context.Context, log Logger, telegramConn, clientConn essentials.Conn) {
defer telegramConn.Close() //nolint: errcheck defer telegramConn.Close() //nolint: errcheck
defer clientConn.Close() //nolint: errcheck defer clientConn.Close() //nolint: errcheck
@@ -16,11 +24,11 @@ func Relay(ctx context.Context, log Logger, telegramConn, clientConn essentials.
ctx, cancel := context.WithCancel(ctx) ctx, cancel := context.WithCancel(ctx)
defer cancel() defer cancel()
go func() { stop := context.AfterFunc(ctx, func() {
<-ctx.Done()
telegramConn.Close() //nolint: errcheck telegramConn.Close() //nolint: errcheck
clientConn.Close() //nolint: errcheck clientConn.Close() //nolint: errcheck
}() })
defer stop()
closeChan := make(chan struct{}) closeChan := make(chan struct{})
@@ -36,12 +44,13 @@ func Relay(ctx context.Context, log Logger, telegramConn, clientConn essentials.
} }
func pump(log Logger, src, dst essentials.Conn, direction string) { func pump(log Logger, src, dst essentials.Conn, direction string) {
var buf [tls.MaxRecordPayloadSize]byte bp := bufPool.Get().(*[]byte)
defer bufPool.Put(bp)
defer src.CloseRead() //nolint: errcheck defer src.CloseRead() //nolint: errcheck
defer dst.CloseWrite() //nolint: errcheck defer dst.CloseWrite() //nolint: errcheck
n, err := io.CopyBuffer(src, dst, buf[:]) n, err := io.CopyBuffer(src, dst, *bp)
switch { switch {
case err == nil: case err == nil:
+3 -3
View File
@@ -65,10 +65,10 @@ func (p *Proxy) ServeConn(conn essentials.Conn) {
ctx := newStreamContext(p.ctx, p.logger, conn) ctx := newStreamContext(p.ctx, p.logger, conn)
defer ctx.Close() defer ctx.Close()
go func() { stop := context.AfterFunc(ctx, func() {
<-ctx.Done()
ctx.Close() ctx.Close()
}() })
defer stop()
p.eventStream.Send(ctx, NewEventStart(ctx.streamID, ctx.ClientIP())) p.eventStream.Send(ctx, NewEventStart(ctx.streamID, ctx.ClientIP()))
ctx.logger.Info("Stream has been started") ctx.logger.Info("Stream has been started")