mirror of
https://github.com/ScuroNeko/mtg.git
synced 2026-08-31 09:54:01 +03:00
Remove clock goroutine
This is a followup for https://github.com/9seconds/mtg/issues/412 it makes sense to manage timers inplace instead of creating for new goroutines: saves memory
This commit is contained in:
@@ -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())
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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{})
|
||||
}
|
||||
@@ -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()
|
||||
})
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user