Fix race condition

This commit is contained in:
Ingo Oppermann
2023-07-12 10:47:29 +02:00
parent 62fdf8e370
commit 0f08480366

View File

@@ -1,6 +1,7 @@
package session package session
import ( import (
"context"
"sync" "sync"
"time" "time"
@@ -16,8 +17,7 @@ type session struct {
logger log.Logger logger log.Logger
sessionActivate sync.Mutex active bool
active bool
location string location string
peer string peer string
@@ -33,8 +33,8 @@ type session struct {
txBitrate *average.SlidingWindow txBitrate *average.SlidingWindow
txBytes uint64 txBytes uint64
tickerStop chan struct{} tickerStop context.CancelFunc
sessionClose sync.Once running bool
lock sync.Mutex lock sync.Mutex
topRxBitrate float64 topRxBitrate float64
@@ -44,6 +44,9 @@ type session struct {
} }
func (s *session) Init(id, reference string, closeCallback func(*session), inactive, timeout time.Duration, logger log.Logger) { func (s *session) Init(id, reference string, closeCallback func(*session), inactive, timeout time.Duration, logger log.Logger) {
s.lock.Lock()
defer s.lock.Unlock()
s.id = id s.id = id
s.reference = reference s.reference = reference
s.createdAt = time.Now() s.createdAt = time.Now()
@@ -71,35 +74,40 @@ func (s *session) Init(id, reference string, closeCallback func(*session), inact
s.callback = func(s *session) {} s.callback = func(s *session) {}
} }
s.tickerStop = make(chan struct{}) s.running = true
s.sessionClose = sync.Once{}
pendingTimeout := inactive pendingTimeout := inactive
if timeout < pendingTimeout { if timeout < pendingTimeout {
pendingTimeout = timeout pendingTimeout = timeout
} }
s.lock.Lock()
defer s.lock.Unlock()
s.stale = time.AfterFunc(pendingTimeout, func() { s.stale = time.AfterFunc(pendingTimeout, func() {
s.close() s.close()
}) })
} }
func (s *session) close() { func (s *session) close() {
s.sessionClose.Do(func() { s.lock.Lock()
s.lock.Lock() defer s.lock.Unlock()
s.stale.Stop()
s.lock.Unlock()
s.closedAt = time.Now() if !s.running {
return
}
close(s.tickerStop) s.running = false
s.rxBitrate.Stop()
s.txBitrate.Stop() s.stale.Stop()
go s.callback(s)
}) s.closedAt = time.Now()
if s.tickerStop != nil {
s.tickerStop()
s.tickerStop = nil
}
s.rxBitrate.Stop()
s.txBitrate.Stop()
go s.callback(s)
} }
func (s *session) Register(location, peer string) { func (s *session) Register(location, peer string) {
@@ -116,15 +124,18 @@ func (s *session) Register(location, peer string) {
} }
func (s *session) Activate() bool { func (s *session) Activate() bool {
s.sessionActivate.Lock() s.lock.Lock()
defer s.sessionActivate.Unlock() defer s.lock.Unlock()
if s.active { if s.active {
return false return false
} }
s.active = true s.active = true
go s.ticker() ctx, cancel := context.WithCancel(context.Background())
s.tickerStop = cancel
go s.ticker(ctx)
return true return true
} }
@@ -237,13 +248,13 @@ func (s *session) SetTopTxBitrate(bitrate float64) {
s.topTxBitrate = bitrate s.topTxBitrate = bitrate
} }
func (s *session) ticker() { func (s *session) ticker(ctx context.Context) {
ticker := time.NewTicker(time.Second) ticker := time.NewTicker(time.Second)
defer ticker.Stop() defer ticker.Stop()
for { for {
select { select {
case <-s.tickerStop: case <-ctx.Done():
return return
case <-ticker.C: case <-ticker.C:
s.lock.Lock() s.lock.Lock()