Correct various race conditions

dharmab · dharmab · commit 425097b8aeb4 · 2024-10-16T21:11:46.000-06:00
diff --git a/Makefile b/Makefile
@@ -127,6 +127,10 @@ $(SKYEYE_BIN): generate $(SKYEYE_SOURCES) $(LIBWHISPER_PATH) $(WHISPER_H_PATH)
 $(SKYEYE_SCALER_BIN): generate $(SKYEYE_SOURCES)
 	$(BUILD_VARS) $(GO) build $(BUILD_FLAGS) ./cmd/skyeye-scaler/
 
+.PHONY: run
+run:
+	$(BUILD_VARS) $(GO) run -race $(BUILD_FLAGS) ./cmd/skyeye/ $(ARGS)
+
 .PHONY: test
 test: generate
 	$(BUILD_VARS) $(GO) run gotest.tools/gotestsum -- $(BUILD_FLAGS) ./...
diff --git a/internal/application/app.go b/internal/application/app.go
@@ -190,7 +190,7 @@ func (a *app) Run(ctx context.Context, cancel context.CancelFunc, wg *sync.WaitG
 	go func() {
 		defer wg.Done()
 		log.Info().Msg("streaming telemetry data to radar")
-		a.tacviewClient.Stream(ctx, a.starts, a.updates, a.fades)
+		a.tacviewClient.Stream(ctx, wg, a.starts, a.updates, a.fades)
 	}()
 
 	wg.Add(1)
diff --git a/pkg/controller/callbacks.go b/pkg/controller/callbacks.go
@@ -48,6 +48,6 @@ func (c *controller) handleFaded(location orb.Point, group brevity.Group, coalit
 	}
 }
 
-func (c *controller) handleRemoved(trackfile trackfiles.Trackfile) {
+func (c *controller) handleRemoved(trackfile *trackfiles.Trackfile) {
 	c.remove(trackfile.Contact.ID)
 }
diff --git a/pkg/radar/callbacks.go b/pkg/radar/callbacks.go
@@ -10,6 +10,8 @@ import (
 type StartedCallback func()
 
 func (s *scope) SetStartedCallback(callback StartedCallback) {
+	s.callbackLock.Lock()
+	defer s.callbackLock.Unlock()
 	s.startedCallback = callback
 }
 
@@ -18,13 +20,17 @@ func (s *scope) SetStartedCallback(callback StartedCallback) {
 type FadedCallback func(location orb.Point, group brevity.Group, coalition coalitions.Coalition)
 
 func (s *scope) SetFadedCallback(callback FadedCallback) {
+	s.callbackLock.Lock()
+	defer s.callbackLock.Unlock()
 	s.fadedCallback = callback
 }
 
 // RemovedCallback is a callback function that is called when a trackfile is aged out and removed.
 // A copy of the trackfile is provided.
-type RemovedCallback func(trackfile trackfiles.Trackfile)
+type RemovedCallback func(trackfile *trackfiles.Trackfile)
 
 func (s *scope) SetRemovedCallback(callback RemovedCallback) {
+	s.callbackLock.Lock()
+	defer s.callbackLock.Unlock()
 	s.removalCallback = callback
 }
diff --git a/pkg/radar/faded.go b/pkg/radar/faded.go
@@ -89,6 +89,8 @@ func (s *scope) handleFaded(fades []sim.Faded) {
 	}
 
 	// call the faded callback for each group
+	s.callbackLock.RLock()
+	defer s.callbackLock.RUnlock()
 	for _, grp := range groups {
 		if s.fadedCallback != nil {
 			s.fadedCallback(grp.point(), &grp, grp.contacts[0].Contact.Coalition)
diff --git a/pkg/radar/radar.go b/pkg/radar/radar.go
@@ -133,6 +133,7 @@ type scope struct {
 	startedCallback       StartedCallback
 	fadedCallback         FadedCallback
 	removalCallback       RemovedCallback
+	callbackLock          sync.RWMutex
 	center                orb.Point
 	mandatoryThreatRadius unit.Length
 }
@@ -239,7 +240,7 @@ func (s *scope) handleGarbageCollection() {
 				Stringer("age", s.missionTime.Sub(lastSeen)).
 				Msg("expired trackfile")
 			if s.removalCallback != nil {
-				s.removalCallback(*trackfile)
+				s.removalCallback(trackfile)
 			}
 		}
 	}
diff --git a/pkg/radar/started.go b/pkg/radar/started.go
@@ -5,6 +5,8 @@ import "github.com/rs/zerolog/log"
 func (s *scope) handleStarted() {
 	log.Info().Msg("clearing all trackfiles due to mission (re)start")
 	s.contacts.reset()
+	s.callbackLock.RLock()
+	defer s.callbackLock.RUnlock()
 	if s.startedCallback != nil {
 		s.startedCallback()
 	}
diff --git a/pkg/sim/sim.go b/pkg/sim/sim.go
@@ -1,29 +1,9 @@
-// package sim provides an inteface for receiving telemetry data from DCS World
 package sim
 
 import (
-	"context"
-	"time"
-
-	"github.com/dharmab/skyeye/pkg/coalitions"
 	"github.com/dharmab/skyeye/pkg/trackfiles"
-	"github.com/paulmach/orb"
 )
 
-// Sim is the interface for receiving telemetry data from the flight simulator.
-type Sim interface {
-	// Stream aircraft updates from the sim to the provided channels.
-	// The first channel receives updates for active aircraft.
-	// The second channel receives messages when an aircraft disappears.
-	// This function blocks until the context is cancelled.
-	Stream(context.Context, chan<- Started, chan<- Updated, chan<- Faded)
-	// Bullseye returns the coalition's bullseye center.
-	Bullseye(coalitions.Coalition) (orb.Point, error)
-	// Time returns the starting time of the mission.
-	// This is useful for looking up magnetic variation.
-	Time() time.Time
-}
-
 // Started is a message sent when a new mission starts.
 type Started struct {
 }
diff --git a/pkg/simpleradio/client.go b/pkg/simpleradio/client.go
@@ -86,7 +86,8 @@ type client struct {
 
 	// lastPing tracks the last time a ping was received. If no pings are received for a period of time, the client will
 	// attempt to reconnect.
-	lastPing time.Time
+	lastPing     time.Time
+	lastPingLock sync.RWMutex
 }
 
 func NewClient(config types.ClientConfiguration) (Client, error) {
@@ -165,20 +166,24 @@ func (c *client) autoheal(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			if time.Since(c.lastPing) > pingInterval*3 {
-				log.Warn().Msg("stopped receiving traffic from SRS server")
-
-				log.Warn().Msg("attempting to reconnect to SRS server")
-				if reconnectErr := c.reconnect(ctx); reconnectErr != nil {
-					log.Err(reconnectErr).Msg("failed to reconnect to SRS server")
-					continue
-				}
-				if initErr := c.initialize(); initErr != nil {
-					log.Err(initErr).Msg("failed to reinitialize SRS client")
-					continue
+			func() {
+				c.lastPingLock.Lock()
+				defer c.lastPingLock.Unlock()
+				if time.Since(c.lastPing) > pingInterval*3 {
+					log.Warn().Msg("stopped receiving traffic from SRS server")
+
+					log.Warn().Msg("attempting to reconnect to SRS server")
+					if reconnectErr := c.reconnect(ctx); reconnectErr != nil {
+						log.Err(reconnectErr).Msg("failed to reconnect to SRS server")
+						return
+					}
+					if initErr := c.initialize(); initErr != nil {
+						log.Err(initErr).Msg("failed to reinitialize SRS client")
+						return
+					}
+					c.lastPing = time.Now()
 				}
-				c.lastPing = time.Now()
-			}
+			}()
 		}
 	}
 }
diff --git a/pkg/simpleradio/ping.go b/pkg/simpleradio/ping.go
@@ -58,7 +58,12 @@ func (c *client) receivePings(ctx context.Context, in <-chan []byte) {
 				log.Debug().Int("bytes", n).Msg("received UDP ping larger than expected")
 			} else {
 				log.Trace().Str("GUID", string(b[0:types.GUIDLength])).Msg("received UDP ping")
-				c.lastPing = time.Now()
+				t := time.Now()
+				func() {
+					c.lastPingLock.Lock()
+					defer c.lastPingLock.Unlock()
+					c.lastPing = t
+				}()
 			}
 		case <-ctx.Done():
 			log.Info().Msg("stopping SRS ping receiver due to context cancellation")
diff --git a/pkg/tacview/client/client.go b/pkg/tacview/client/client.go
@@ -23,10 +23,8 @@ import (
 )
 
 type client struct {
-	// starts should be sent to when the real-time telemetry server changes missions.
-	starts chan struct{}
-	// removals should be sent to when an object is removed in the ACMI data.
-	removals chan *types.Object
+	starts chan sim.Started
+	fades  chan sim.Faded
 
 	// updateInterval is how often to send updates to the channels passed to Stream().
 	updateInterval time.Duration
@@ -44,45 +42,70 @@ type client struct {
 	state map[uint64]*types.Object
 	// bullseyesIdx maps coalitions to bullseye object IDs.
 	bullseyesIdx map[coalitions.Coalition]uint64
-	// stateLock protects state and bullseyesIdx.
-	stateLock sync.RWMutex
+	// lock protects state and bullseyesIdx.
+	lock sync.RWMutex
 }
 
-var _ sim.Sim = &client{}
-
 func NewClient(
 	updateInterval time.Duration,
 ) *client {
 	c := &client{
-		starts:         make(chan struct{}),
-		removals:       make(chan *types.Object),
+		starts:         make(chan sim.Started),
+		fades:          make(chan sim.Faded),
 		updateInterval: updateInterval,
 	}
 	c.reset()
 	return c
 }
 
-func (c *client) Stream(ctx context.Context, starts chan<- sim.Started, updates chan<- sim.Updated, fades chan<- sim.Faded) {
+func (c *client) Stream(ctx context.Context, wg *sync.WaitGroup, starts chan<- sim.Started, updates chan<- sim.Updated, fades chan<- sim.Faded) {
 	ticker := time.NewTicker(c.updateInterval)
 	defer ticker.Stop()
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-c.starts:
-			log.Info().Msg("dispatching mission start event")
-			starts <- sim.Started{}
-		case removed := <-c.removals:
-			fades <- sim.Faded{ID: removed.ID}
-		case <-ticker.C:
-			c.sendUpdates(updates)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case start := <-c.starts:
+				starts <- start
+			}
 		}
-	}
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case fade := <-c.fades:
+				fades <- fade
+			}
+		}
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				c.sendUpdates(updates)
+			}
+		}
+	}()
+
+	<-ctx.Done()
 }
 
 func (c *client) Bullseye(coalition coalitions.Coalition) (orb.Point, error) {
-	c.stateLock.RLock()
-	defer c.stateLock.RUnlock()
+	c.lock.RLock()
+	defer c.lock.RUnlock()
 
 	if id, ok := c.bullseyesIdx[coalition]; ok {
 		if bullseye, ok := c.state[id]; ok {
@@ -95,14 +118,14 @@ func (c *client) Bullseye(coalition coalitions.Coalition) (orb.Point, error) {
 }
 
 func (c *client) Time() time.Time {
-	c.stateLock.RLock()
-	defer c.stateLock.RUnlock()
+	c.lock.RLock()
+	defer c.lock.RUnlock()
 	return c.cursorTime
 }
 
 func (c *client) sendUpdates(updates chan<- sim.Updated) {
-	c.stateLock.RLock()
-	defer c.stateLock.RUnlock()
+	c.lock.RLock()
+	defer c.lock.RUnlock()
 
 	for _, object := range c.state {
 		logger := log.With().Uint64("id", object.ID).Logger()
@@ -171,7 +194,7 @@ func (c *client) handleLines(ctx context.Context, reader *bufio.Reader) error {
 	log.Info().Msg("resetting ACMI client state")
 	c.reset()
 	log.Info().Msg("sending mission start message")
-	c.starts <- struct{}{}
+	c.starts <- sim.Started{}
 
 	ticker := time.NewTicker(1 * time.Minute)
 	for {
@@ -255,6 +278,8 @@ func (c *client) handleUpdate(reader *bufio.Reader) error {
 }
 
 func (c *client) handleTimeFrame(line string) error {
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	if !strings.HasPrefix(line, "#") {
 		return nil
 	}
@@ -270,6 +295,8 @@ func (c *client) handleTimeFrame(line string) error {
 }
 
 func (c *client) updateGlobalObject(update *types.ObjectUpdate) error {
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	if update.ID != types.GlobalObjectID {
 		return nil
 	}
@@ -304,8 +331,8 @@ func (c *client) updateGlobalObject(update *types.ObjectUpdate) error {
 }
 
 func (c *client) updateObject(update *types.ObjectUpdate) error {
-	c.stateLock.Lock()
-	defer c.stateLock.Unlock()
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	var isNewObject bool
 	logger := log.With().Uint64("id", update.ID).Logger()
 
@@ -352,7 +379,7 @@ func (c *client) updateObject(update *types.ObjectUpdate) error {
 
 	if update.IsRemoval {
 		delete(c.state, object.ID)
-		c.removals <- object
+		c.fades <- sim.Faded{ID: object.ID}
 		if IsRelevantObject(taglist) {
 			logger.Info().Msg("recording object removal")
 		}
@@ -362,8 +389,8 @@ func (c *client) updateObject(update *types.ObjectUpdate) error {
 }
 
 func (c *client) reset() {
-	c.stateLock.Lock()
-	defer c.stateLock.Unlock()
+	c.lock.Lock()
+	defer c.lock.Unlock()
 
 	c.referenceTime = time.Time{}
 	c.referencePoint = orb.Point{}
diff --git a/pkg/tacview/client/interface.go b/pkg/tacview/client/interface.go
@@ -3,11 +3,16 @@ package client
 import (
 	"context"
 	"sync"
+	"time"
 
+	"github.com/dharmab/skyeye/pkg/coalitions"
 	"github.com/dharmab/skyeye/pkg/sim"
+	"github.com/paulmach/orb"
 )
 
 type Client interface {
-	sim.Sim
 	Run(context.Context, *sync.WaitGroup) error
+	Stream(context.Context, *sync.WaitGroup, chan<- sim.Started, chan<- sim.Updated, chan<- sim.Faded)
+	Bullseye(coalitions.Coalition) (orb.Point, error)
+	Time() time.Time
 }
diff --git a/pkg/trackfiles/trackfile.go b/pkg/trackfiles/trackfile.go

Original file line number	Diff line number	Diff line change
`@@ -48,6 +48,6 @@ func (c *controller) handleFaded(location orb.Point, group brevity.Group, coalit`
`48`	`48`	`}`
`49`	`49`	`}`
`50`	`50`
`51`		`-func (c *controller) handleRemoved(trackfile trackfiles.Trackfile) {`
	`51`	`+func (c controller) handleRemoved(trackfile trackfiles.Trackfile) {`
`52`	`52`	`c.remove(trackfile.Contact.ID)`
`53`	`53`	`}`
Original file line number	Diff line number	Diff line change
`@@ -89,6 +89,8 @@ func (s *scope) handleFaded(fades []sim.Faded) {`
`89`	`89`	`}`
`90`	`90`
`91`	`91`	`// call the faded callback for each group`
	`92`	`+ s.callbackLock.RLock()`
	`93`	`+ defer s.callbackLock.RUnlock()`
`92`	`94`	`for _, grp := range groups {`
`93`	`95`	`if s.fadedCallback != nil {`
`94`	`96`	`s.fadedCallback(grp.point(), &grp, grp.contacts[0].Contact.Coalition)`
Original file line number	Diff line number	Diff line change
`@@ -133,6 +133,7 @@ type scope struct {`
`133`	`133`	`startedCallback StartedCallback`
`134`	`134`	`fadedCallback FadedCallback`
`135`	`135`	`removalCallback RemovedCallback`
	`136`	`+ callbackLock sync.RWMutex`
`136`	`137`	`center orb.Point`
`137`	`138`	`mandatoryThreatRadius unit.Length`
`138`	`139`	`}`
`@@ -239,7 +240,7 @@ func (s *scope) handleGarbageCollection() {`
`239`	`240`	`Stringer("age", s.missionTime.Sub(lastSeen)).`
`240`	`241`	`Msg("expired trackfile")`
`241`	`242`	`if s.removalCallback != nil {`
`242`		`- s.removalCallback(*trackfile)`
	`243`	`+ s.removalCallback(trackfile)`
`243`	`244`	`}`
`244`	`245`	`}`
`245`	`246`	`}`
Original file line number	Diff line number	Diff line change
`@@ -5,6 +5,8 @@ import "github.com/rs/zerolog/log"`
`5`	`5`	`func (s *scope) handleStarted() {`
`6`	`6`	`log.Info().Msg("clearing all trackfiles due to mission (re)start")`
`7`	`7`	`s.contacts.reset()`
	`8`	`+ s.callbackLock.RLock()`
	`9`	`+ defer s.callbackLock.RUnlock()`
`8`	`10`	`if s.startedCallback != nil {`
`9`	`11`	`s.startedCallback()`
`10`	`12`	`}`