From 3cf78316df14425e3811caaf483bf92eebe52cfc Mon Sep 17 00:00:00 2001 From: samosvalishe Date: Mon, 17 Aug 2026 17:14:10 +0300 Subject: [PATCH] =?UTF-8?q?perf:=20=D0=BC=D0=B5=D0=BD=D1=8C=D1=88=D0=B5=20?= =?UTF-8?q?=D1=84=D0=BE=D0=BD=D0=BE=D0=B2=D1=8B=D1=85=20=D1=82=D0=B0=D0=B9?= =?UTF-8?q?=D0=BC=D0=B5=D1=80=D0=BE=D0=B2=20=D0=BF=D1=80=D0=B8=20=D0=BF?= =?UTF-8?q?=D0=BE=D0=B4=D0=BD=D1=8F=D1=82=D0=BE=D0=BC=20=D1=82=D1=83=D0=BD?= =?UTF-8?q?=D0=BD=D0=B5=D0=BB=D0=B5=20=D0=B8=20=D0=BE=D0=BF=D1=80=D0=BE?= =?UTF-8?q?=D1=81=20=D0=BC=D0=B5=D1=82=D1=80=D0=B8=D0=BA=20=D1=82=D0=BE?= =?UTF-8?q?=D0=BB=D1=8C=D0=BA=D0=BE=20=D0=BF=D1=80=D0=B8=20=D0=B2=D0=B8?= =?UTF-8?q?=D0=B4=D0=B8=D0=BC=D0=BE=D0=BC=20=D0=BE=D0=BA=D0=BD=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/proxy/udprelay/run.go | 5 ++- internal/session/session.go | 34 ++++++-------------- internal/session/session_test.go | 26 ++++++++------- internal/session/traffic.go | 55 +++++++++++++++++--------------- 4 files changed, 55 insertions(+), 65 deletions(-) diff --git a/internal/proxy/udprelay/run.go b/internal/proxy/udprelay/run.go index db857d0..6bb486c 100644 --- a/internal/proxy/udprelay/run.go +++ b/internal/proxy/udprelay/run.go @@ -94,7 +94,6 @@ func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, log wg.Go(func() { runListener(runCtx, listenConn, &activeLocalPeer, inboundChan) }) - t := time.Tick(200 * time.Millisecond) // Стрим 1 стартует первым для прогрева кэша учетных данных. okchan := make(chan struct{}, 1) @@ -104,7 +103,7 @@ func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, log DTLSLoop(runCtx, deps, params, peer, listenConn, inboundChan, cchan, okchan, 1) }) wg.Go(func() { - TURNLoop(runCtx, deps, params, peer, cchan, t, 1) + TURNLoop(runCtx, deps, params, peer, cchan, 1) }) } @@ -121,7 +120,7 @@ func Run(ctx context.Context, dtlsDialer *dtlsdial.Dialer, auth AuthHandler, log DTLSLoop(runCtx, deps, params, peer, listenConn, inboundChan, cchan, nil, streamID) }) wg.Go(func() { - TURNLoop(runCtx, deps, params, peer, cchan, t, streamID) + TURNLoop(runCtx, deps, params, peer, cchan, streamID) }) } diff --git a/internal/session/session.go b/internal/session/session.go index 39c9642..7fd08e9 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -40,12 +40,13 @@ var ( ) const ( - defaultStatusInterval = 500 * time.Millisecond - defaultRateInterval = time.Second + defaultStatusInterval = 2 * time.Second defaultUDPHandshakeTimeout = 20 * time.Second defaultHandshakeConcurrency = 3 - wakeTick = 5 * time.Second + // Порог вдвое больше тика: сон короче порога всё равно пропускаем, а тик почаще + // стоил бы пробуждений процесса на всё время сессии. + wakeTick = 30 * time.Second wakeThreshold = 60 * time.Second ) @@ -53,7 +54,6 @@ type Options struct { ConnectTimeout time.Duration StatusInterval time.Duration - RateInterval time.Duration UDPHandshakeTimeout time.Duration HandshakeConcurrency int @@ -64,9 +64,6 @@ func (o Options) withDefaults() Options { if o.StatusInterval <= 0 { o.StatusInterval = defaultStatusInterval } - if o.RateInterval <= 0 { - o.RateInterval = defaultRateInterval - } if o.UDPHandshakeTimeout <= 0 { o.UDPHandshakeTimeout = defaultUDPHandshakeTimeout } @@ -184,28 +181,16 @@ func (s *Session) Run(ctx context.Context) (err error) { defer cancel() var bg sync.WaitGroup - if s.traffic != nil { - bg.Add(1) - go func() { - defer bg.Done() - s.traffic.rateMeter(runCtx, s.opts.RateInterval) - }() - } - - bg.Add(1) - go func() { - defer bg.Done() + bg.Go(func() { s.wake.Watch(runCtx, wakeTick, wakeThreshold, func(gap time.Duration) { log.Warnf("device slept for %s - recycling TURN allocations", gap.Truncate(time.Second)) }) - }() + }) var watchdogErr error - bg.Add(1) - go func() { - defer bg.Done() + bg.Go(func() { watchdogErr = s.watch(runCtx, cancel) - }() + }) relayErr := s.relay(runCtx, prov, peer) stopped := runCtx.Err() != nil @@ -230,8 +215,7 @@ func (s *Session) Snapshot() Snapshot { snap := Snapshot{Phase: st.phase, Streams: st.streams, Total: st.total, Err: st.err} if s.traffic != nil { snap.TxTotal, snap.RxTotal = s.traffic.stats.Counters() - snap.TxRate = s.traffic.txRate.Load() - snap.RxRate = s.traffic.rxRate.Load() + snap.TxRate, snap.RxRate = s.traffic.rates() } return snap } diff --git a/internal/session/session_test.go b/internal/session/session_test.go index 5715c29..dacb271 100644 --- a/internal/session/session_test.go +++ b/internal/session/session_test.go @@ -283,21 +283,25 @@ func TestSnapshotWithoutTrafficStaysZero(t *testing.T) { } } -func TestTrafficRateMeter(t *testing.T) { +func TestTrafficRates(t *testing.T) { tr := newTraffic() - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - go tr.rateMeter(ctx, 10*time.Millisecond) + // Первый вызов только запоминает отметку - скорость считать не с чего. + if tx, rx := tr.rates(); tx != 0 || rx != 0 { + t.Fatalf("first rates = tx %d rx %d, want zero", tx, rx) + } tr.stats.AddTx(1000) tr.stats.AddRx(500) + time.Sleep(50 * time.Millisecond) - deadline := time.Now().Add(2 * time.Second) - for time.Now().Before(deadline) { - if tr.txRate.Load() == 1000 && tr.rxRate.Load() == 500 { - return - } - time.Sleep(5 * time.Millisecond) + tx, rx := tr.rates() + if tx < 1000 || rx < 500 || rx >= tx { + t.Fatalf("rates = tx %d rx %d, want tx >= 1000 > rx >= 500", tx, rx) + } + + // Без нового трафика скорость обязана упасть в ноль. + time.Sleep(10 * time.Millisecond) + if tx, rx := tr.rates(); tx != 0 || rx != 0 { + t.Fatalf("idle rates = tx %d rx %d, want zero", tx, rx) } - t.Fatalf("rates = tx %d rx %d, want tx 1000 rx 500", tr.txRate.Load(), tr.rxRate.Load()) } diff --git a/internal/session/traffic.go b/internal/session/traffic.go index eebdec4..3724380 100644 --- a/internal/session/traffic.go +++ b/internal/session/traffic.go @@ -1,48 +1,51 @@ package session import ( - "context" "math" - "sync/atomic" + "sync" "time" "github.com/samosvalishe/free-turn-proxy/internal/stats" ) -// traffic агрегирует счётчики и мгновенную скорость передачи. +// traffic агрегирует счётчики и считает скорость передачи. type traffic struct { - stats *stats.Stats - txRate atomic.Int64 - rxRate atomic.Int64 + stats *stats.Stats + + mu sync.Mutex + prevTx uint64 + prevRx uint64 + prevAt time.Time } func newTraffic() *traffic { return &traffic{stats: stats.New(true)} } -func (t *traffic) rateMeter(ctx context.Context, interval time.Duration) { - tick := time.NewTicker(interval) - defer tick.Stop() - var prevTx, prevRx uint64 - for { - select { - case <-ctx.Done(): - t.txRate.Store(0) - t.rxRate.Store(0) - return - case <-tick.C: - tx, rx := t.stats.Counters() - t.txRate.Store(clampInt64(tx - prevTx)) - t.rxRate.Store(clampInt64(rx - prevRx)) - prevTx, prevRx = tx, rx - } +// rates - средняя скорость (байт/с) с прошлого вызова. Считается на запрос, а не +// фоновым тикером: спрашивает только открытый экран, и будить процесс ради цифры, +// которую никто не читает, незачем. +func (t *traffic) rates() (txRate, rxRate int64) { + tx, rx := t.stats.Counters() + now := time.Now() + + t.mu.Lock() + defer t.mu.Unlock() + prevTx, prevRx, prevAt := t.prevTx, t.prevRx, t.prevAt + t.prevTx, t.prevRx, t.prevAt = tx, rx, now + + elapsed := now.Sub(prevAt) + if prevAt.IsZero() || elapsed <= 0 { + return 0, 0 } + sec := elapsed.Seconds() + return perSecond(tx-prevTx, sec), perSecond(rx-prevRx, sec) } -// clampInt64 безопасно приводит uint64 к int64 с насыщением при переполнении. -func clampInt64(u uint64) int64 { - if u > math.MaxInt64 { +func perSecond(delta uint64, sec float64) int64 { + rate := float64(delta) / sec + if rate >= math.MaxInt64 { return math.MaxInt64 } - return int64(u) + return int64(rate) }