Merge remote-tracking branch 'upstream/master'

# Conflicts:
#	mtglib/proxy.go

mtglib/internal/doppel/clock.go

@@ -1,35 +0,0 @@
-package doppel
-
-import (
-	"context"
-	"time"
-)
-
-type Clock struct {
-	stats *Stats
-	tick  chan struct{}
-}
-
-func (c Clock) Start(ctx context.Context) {
-	tickTock := time.NewTimer(c.stats.Delay())
-	defer func() {
-		tickTock.Stop()
-		select {
-		case <-tickTock.C:
-		default:
-		}
-	}()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-tickTock.C:
-			select {
-			case <-ctx.Done():
-			case c.tick <- struct{}{}:
-			}
-			tickTock.Reset(c.stats.Delay())
-		}
-	}
-}

mtglib/internal/doppel/clock_test.go

@@ -1,80 +0,0 @@
-package doppel
-
-import (
-	"context"
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/suite"
-)
-
-type ClockTestSuite struct {
-	suite.Suite
-
-	clock     Clock
-	wg        sync.WaitGroup
-	ctx       context.Context
-	ctxCancel context.CancelFunc
-}
-
-func (suite *ClockTestSuite) SetupTest() {
-	ctx, cancel := context.WithCancel(context.Background())
-
-	suite.ctx = ctx
-	suite.ctxCancel = cancel
-	suite.clock = Clock{
-		stats: &Stats{
-			k:      StatsDefaultK,
-			lambda: StatsDefaultLambda,
-		},
-		tick: make(chan struct{}),
-	}
-
-	suite.wg.Go(func() {
-		suite.clock.Start(suite.ctx)
-	})
-}
-
-func (suite *ClockTestSuite) TearDownTest() {
-	suite.ctxCancel()
-	suite.wg.Wait()
-}
-
-func (suite *ClockTestSuite) TestTicks() {
-	received := 0
-
-	for range 3 {
-		select {
-		case <-suite.clock.tick:
-			received++
-		case <-time.After(2 * time.Second):
-			suite.Fail("timed out waiting for tick")
-		}
-	}
-
-	suite.Equal(3, received)
-}
-
-func (suite *ClockTestSuite) TestStopsOnCancel() {
-	select {
-	case <-suite.clock.tick:
-	case <-time.After(2 * time.Second):
-		suite.Fail("timed out waiting for first tick")
-	}
-
-	suite.ctxCancel()
-
-	time.Sleep(50 * time.Millisecond)
-
-	select {
-	case <-suite.clock.tick:
-		suite.Fail("received tick after cancel")
-	default:
-	}
-}
-
-func TestClock(t *testing.T) {
-	t.Parallel()
-	suite.Run(t, &ClockTestSuite{})
-}

mtglib/internal/doppel/conn.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"sync"
+	"time"
 
 	"github.com/9seconds/mtg/v2/essentials"
 	"github.com/9seconds/mtg/v2/mtglib/internal/tls"
@@ -25,7 +26,7 @@ type Conn struct {
 type connPayload struct {
 	ctx         context.Context
 	ctxCancel   context.CancelCauseFunc
-	clock       Clock
+	stats       Stats
 	wg          sync.WaitGroup
 	writeStream bytes.Buffer
 	writtenCond sync.Cond
@@ -46,25 +47,23 @@ func (c Conn) Write(p []byte) (int, error) {
 	return len(p), context.Cause(c.p.ctx)
 }
 
-func (c Conn) Start() {
-	c.p.wg.Go(func() {
-		c.start()
-	})
-}
-
 func (c Conn) start() {
 	bp := doppelBufPool.Get().(*[]byte)
 	buf := *bp
 	defer doppelBufPool.Put(bp)
 
+	timer := time.NewTimer(c.p.stats.Delay())
+	defer timer.Stop()
+
 	for {
 		select {
 		case <-c.p.ctx.Done():
 			return
-		case <-c.p.clock.tick:
+		case <-timer.C:
+			timer.Reset(c.p.stats.Delay())
 		}
 
-		size := c.p.clock.stats.Size()
+		size := c.p.stats.Size()
 
 		c.p.writtenCond.L.Lock()
 		for c.p.writeStream.Len() == 0 && !c.p.done {
@@ -95,28 +94,22 @@ func (c Conn) Stop() {
 	c.p.wg.Wait()
 }
 
-func NewConn(ctx context.Context, conn essentials.Conn, stats *Stats) Conn {
+func NewConn(ctx context.Context, conn essentials.Conn, stats Stats) Conn {
 	ctx, cancel := context.WithCancelCause(ctx)
 	rv := Conn{
 		Conn: conn,
 		p: &connPayload{
 			ctx:       ctx,
 			ctxCancel: cancel,
+			stats:     stats,
 			writtenCond: sync.Cond{
 				L: &sync.Mutex{},
 			},
-			clock: Clock{
-				stats: stats,
-				tick:  make(chan struct{}),
-			},
 		},
 	}
 
 	rv.p.writeStream.Grow(tls.DefaultBufferSize)
 
-	rv.p.wg.Go(func() {
-		rv.p.clock.Start(ctx)
-	})
 	rv.p.wg.Go(func() {
 		rv.start()
 	})

mtglib/internal/doppel/conn_test.go

@@ -63,7 +63,7 @@ func (suite *ConnTestSuite) TearDownTest() {
 }
 
 func (suite *ConnTestSuite) makeConn() Conn {
-	return NewConn(suite.ctx, suite.connMock, &Stats{
+	return NewConn(suite.ctx, suite.connMock, Stats{
 		k:      2.0,
 		lambda: 0.01,
 	})
@@ -152,7 +152,7 @@ func (suite *ConnTestSuite) TestStopDoesNotDeadlockWhenStartIsWaiting() {
 			ctx, cancel := context.WithCancel(suite.ctx)
 			defer cancel()
 
-			c := NewConn(ctx, suite.connMock, &Stats{
+			c := NewConn(ctx, suite.connMock, Stats{
 				k:      2.0,
 				lambda: 0.01,
 			})

mtglib/internal/doppel/ganger.go

@@ -47,7 +47,7 @@ type Ganger struct {
 
 	drs bool
 
-	stats     *Stats
+	stats     Stats
 	durations []time.Duration
 	certSizes []int
 
@@ -113,7 +113,7 @@ func (g *Ganger) run() {
 	scoutCollectedChan := make(chan scoutRaidResult)
 	currentScoutCollectedChan := scoutCollectedChan
 
-	updatedStatsChan := make(chan *Stats)
+	updatedStatsChan := make(chan Stats)
 
 	g.wg.Go(func() {
 		g.runScoutRaid(scoutCollectedChan)
@@ -256,7 +256,7 @@ func NewGanger(
 		scoutRaidEach:    scoutEach,
 		scoutRaidRepeats: scoutRepeats,
 		drs:              drs,
-		stats: &Stats{
+		stats: Stats{
 			k:      StatsDefaultK,
 			lambda: StatsDefaultLambda,
 			drs:    drs,

mtglib/internal/doppel/stats.go

@@ -112,7 +112,7 @@ func (d *Stats) Size() int {
 	return TLSRecordSizeMax
 }
 
-func NewStats(durations []time.Duration, drs bool) *Stats {
+func NewStats(durations []time.Duration, drs bool) Stats {
 	n := float64(len(durations))
 
 	// in milliseconds
@@ -162,7 +162,7 @@ func NewStats(durations []time.Duration, drs bool) *Stats {
 	// λ = (Σxᵢᵏ / n)^(1/k)
 	lambda := math.Pow(sumXK/n, 1.0/k)
 
-	return &Stats{
+	return Stats{
 		k:      k,
 		lambda: lambda,
 		drs:    drs,

mtglib/internal/relay/pool_settings_mips.go

@@ -0,0 +1,13 @@
+//go:build mips || mipsle
+
+package relay
+
+import "github.com/9seconds/mtg/v2/mtglib/internal/tls"
+
+const (
+	// MIPS is quite short in resources, and usually it means that it will run
+	// on Microtiks, OpenWRT-based routers or similar hardware. I think it worth
+	// to sacrifice a number of read syscalls (read, CPU load) to shrink
+	// limited RAM resources.
+	bufPoolSize = tls.MaxRecordPayloadSize / 2
+)

mtglib/internal/relay/pool_settings_other.go

@@ -0,0 +1,9 @@
+//go:build !mips && !mipsle
+
+package relay
+
+import "github.com/9seconds/mtg/v2/mtglib/internal/tls"
+
+const (
+	bufPoolSize = tls.MaxRecordPayloadSize
+)

mtglib/internal/relay/pools.go

@@ -0,0 +1,18 @@
+package relay
+
+import "sync"
+
+var bufPool = sync.Pool{
+	New: func() any {
+		b := make([]byte, bufPoolSize)
+		return &b
+	},
+}
+
+func acquireBuffer() *[]byte {
+	return bufPool.Get().(*[]byte)
+}
+
+func releaseBuffer(p *[]byte) {
+	bufPool.Put(p)
+}

mtglib/internal/relay/relay.go

@@ -4,19 +4,10 @@ import (
 	"context"
 	"errors"
 	"io"
-	"sync"
 
 	"github.com/9seconds/mtg/v2/essentials"
-	"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) {
 	defer telegramConn.Close() //nolint: errcheck
 	defer clientConn.Close()   //nolint: errcheck
@@ -44,13 +35,13 @@ func Relay(ctx context.Context, log Logger, telegramConn, clientConn essentials.
 }
 
 func pump(log Logger, src, dst essentials.Conn, direction string) {
-	bp := bufPool.Get().(*[]byte)
-	defer bufPool.Put(bp)
+	buf := acquireBuffer()
+	defer releaseBuffer(buf)
 
 	defer src.CloseRead()  //nolint: errcheck
 	defer dst.CloseWrite() //nolint: errcheck
 
-	n, err := io.CopyBuffer(src, dst, *bp)
+	n, err := io.CopyBuffer(src, dst, *buf)
 
 	switch {
 	case err == nil:

mtglib/proxy.go

@@ -111,13 +111,11 @@ func (p *Proxy) ServeConn(conn essentials.Conn) {
 		return
 	}
 
-	countedClientConn := newCountingConn(ctx.clientConn, p.stats, ctx.secretName)
-
 	relay.Relay(
 		ctx,
 		ctx.logger.Named("relay"),
-		ctx.telegramConn,
-		countedClientConn,
+		connIdleTimeout{Conn: ctx.telegramConn, timeout: p.idleTimeout},
+		newCountingConn(connIdleTimeout{Conn: ctx.clientConn, timeout: p.idleTimeout}, p.stats, ctx.secretName),
 	)
 }