Remove data races

Author: g3force

internal/app/ci/ci.go

@@ -6,6 +6,7 @@ import (
 	"github.com/RoboCup-SSL/ssl-go-tools/pkg/sslconn"
 	"log"
 	"net"
+	"sync"
 	"time"
 )
 
@@ -17,6 +18,7 @@ type Server struct {
 	conn            net.Conn
 	latestTime      time.Time
 	tickChan        chan time.Time
+	mutex           sync.Mutex
 	TrackerConsumer func(*tracker.TrackerWrapperPacket)
 }
 
@@ -77,7 +79,9 @@ func (s *Server) serve(conn net.Conn) {
 		if input.Timestamp != nil {
 			sec := *input.Timestamp / 1e9
 			nSec := *input.Timestamp - sec*1e9
+			s.mutex.Lock()
 			s.latestTime = time.Unix(sec, nSec)
+			s.mutex.Unlock()
 			select {
 			case s.tickChan <- time.Now():
 			default:
@@ -99,6 +103,8 @@ func (s *Server) SendMessage(refMsg *state.Referee) {
 }
 
 func (s *Server) Time() time.Time {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
 	return s.latestTime
 }
 

internal/app/engine/consume_geometry.go

@@ -7,8 +7,10 @@ import (
 )
 
 func (e *Engine) ProcessGeometry(data *vision.SSL_GeometryData) {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 
-	currentGeometry := e.GetGeometry()
+	currentGeometry := e.stateMachine.Geometry
 	newGeometry := currentGeometry
 
 	newGeometry.FieldWidth = float64(*data.Field.FieldWidth) / 1000.0
@@ -22,7 +24,7 @@ func (e *Engine) ProcessGeometry(data *vision.SSL_GeometryData) {
 		}
 	}
 
-	e.SetGeometry(newGeometry)
+	e.stateMachine.Geometry = newGeometry
 
 	if currentGeometry != newGeometry {
 		log.Printf("Geometry changed from %v to %v", currentGeometry, newGeometry)

internal/app/engine/consume_tracker.go

@@ -5,6 +5,9 @@ import (
 )
 
 func (e *Engine) ProcessTrackerFrame(wrapperFrame *tracker.TrackerWrapperPacket) {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
+
 	if wrapperFrame.TrackedFrame == nil {
 		return
 	}
@@ -17,17 +20,15 @@ func (e *Engine) ProcessTrackerFrame(wrapperFrame *tracker.TrackerWrapperPacket)
 
 	readyToContinue := e.readyToContinue()
 
-	e.UpdateGcState(func(gcState *GcState) {
-		gcState.TrackerState[*wrapperFrame.Uuid] = &state
+	e.gcState.TrackerState[*wrapperFrame.Uuid] = &state
 
-		// for now, all tracker sources update the GC state
-		e.gcState.TrackerStateGc = &state
+	// for now, all tracker sources update the GC state
+	e.gcState.TrackerStateGc = &state
 
-		if e.gcState.ReadyToContinue == nil {
-			e.gcState.ReadyToContinue = new(bool)
-		}
-		*e.gcState.ReadyToContinue = readyToContinue
-	})
+	if e.gcState.ReadyToContinue == nil {
+		e.gcState.ReadyToContinue = new(bool)
+	}
+	*e.gcState.ReadyToContinue = readyToContinue
 }
 
 func convertRobots(robots []*tracker.TrackedRobot) (rs []*Robot) {

internal/app/engine/engine.go

@@ -27,7 +27,7 @@ type Engine struct {
 	timeProvider             timer.TimeProvider
 	lastTimeUpdate           time.Time
 	gcState                  *GcState
-	gcStateMutex             sync.Mutex
+	mutex                    sync.Mutex
 	noProgressDetector       NoProgressDetector
 	ballPlacementCoordinator BallPlacementCoordinator
 	tickChanProvider         func() <-chan time.Time
@@ -68,29 +68,30 @@ func (e *Engine) Enqueue(change *statemachine.Change) {
 	e.queue <- change
 }
 
-// SetGeometry sets a new geometry
-func (e *Engine) SetGeometry(geometry config.Geometry) {
-	e.stateMachine.Geometry = geometry
-}
-
-// GetGeometry returns the current geometry
-func (e *Engine) GetGeometry() config.Geometry {
+// getGeometry returns the current geometry
+func (e *Engine) getGeometry() config.Geometry {
 	return e.stateMachine.Geometry
 }
 
 // SetTimeProvider sets a new time provider for this engine
 func (e *Engine) SetTimeProvider(provider timer.TimeProvider) {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 	e.timeProvider = provider
 	e.lastTimeUpdate = e.timeProvider()
 }
 
 // SetTickChanProvider sets an alternative provider for the tick channel
 func (e *Engine) SetTickChanProvider(provider func() <-chan time.Time) {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 	e.tickChanProvider = provider
 }
 
 // Start loads the state store and runs a go routine that consumes the change queue
 func (e *Engine) Start() error {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 	if err := e.stateStore.Open(); err != nil {
 		return errors.Wrap(err, "Could not open state store")
 	}
@@ -105,6 +106,8 @@ func (e *Engine) Start() error {
 
 // Stop stops the go routine that processes the change queue
 func (e *Engine) Stop() {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 	close(e.queue)
 	if err := e.stateStore.Close(); err != nil {
 		log.Printf("Could not close store: %v", err)
@@ -113,28 +116,30 @@ func (e *Engine) Stop() {
 
 // CurrentState returns a deep copy of the current state
 func (e *Engine) CurrentState() (s *state.State) {
-	s = new(state.State)
-	proto.Merge(s, e.currentState)
-	return
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
+	return e.currentState.Clone()
 }
 
 // CurrentGcState returns a deep copy of the current GC state
 func (e *Engine) CurrentGcState() (s *GcState) {
-	e.gcStateMutex.Lock()
-	defer e.gcStateMutex.Unlock()
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 	s = new(GcState)
 	proto.Merge(s, e.gcState)
 	return
 }
 
 func (e *Engine) UpdateGcState(fn func(gcState *GcState)) {
-	e.gcStateMutex.Lock()
-	defer e.gcStateMutex.Unlock()
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 	fn(e.gcState)
 }
 
 // LatestChangesUntil returns all changes with a id larger than the given id
 func (e *Engine) LatestChangesUntil(id int32) (changes []*statemachine.StateChange) {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 	for _, change := range e.stateStore.Entries() {
 		if *change.Id > id {
 			changes = append(changes, change)
@@ -145,6 +150,8 @@ func (e *Engine) LatestChangesUntil(id int32) (changes []*statemachine.StateChan
 
 // LatestChangeId returns the latest change id or -1, if there is no change
 func (e *Engine) LatestChangeId() int32 {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 	entries := e.stateStore.Entries()
 	if len(entries) > 0 {
 		return *entries[len(entries)-1].Id
@@ -169,8 +176,8 @@ func (e *Engine) processChanges() {
 
 // ResetMatch creates a backup of the current state store, removes it and starts with a fresh state
 func (e *Engine) ResetMatch() {
-	e.gcStateMutex.Lock()
-	defer e.gcStateMutex.Unlock()
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 	if err := e.stateStore.Reset(); err != nil {
 		log.Printf("Could not reset store: %v", err)
 	} else {
@@ -179,6 +186,8 @@ func (e *Engine) ResetMatch() {
 }
 
 func (e *Engine) processChange(change *statemachine.Change) {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 
 	var newChanges []*statemachine.Change
 	entry := statemachine.StateChange{}
@@ -218,7 +227,7 @@ func (e *Engine) processChange(change *statemachine.Change) {
 		entry.State, newChanges = e.stateMachine.Process(e.currentState, change)
 	}
 
-	e.currentState = entry.State
+	e.currentState = entry.State.Clone()
 
 	e.postProcessChange(entry)
 
@@ -229,8 +238,12 @@ func (e *Engine) processChange(change *statemachine.Change) {
 	if err := e.stateStore.Add(&entry); err != nil {
 		log.Println("Could not add new state to store: ", err)
 	}
+	stateCopy := e.currentState.Clone()
 	for _, hook := range e.hooks {
-		hook <- HookOut{Change: entry.Change, State: e.CurrentState()}
+		select {
+		case hook <- HookOut{Change: entry.Change, State: stateCopy}:
+		default:
+		}
 	}
 }
 

internal/app/engine/hook.go

@@ -12,11 +12,15 @@ type HookOut struct {
 
 // RegisterHook registers given hook for post processing after each change
 func (e *Engine) RegisterHook(hook chan HookOut) {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 	e.hooks = append(e.hooks, hook)
 }
 
 // UnregisterHook unregisters hooks that were registered before
 func (e *Engine) UnregisterHook(hook chan HookOut) bool {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
 	for i, h := range e.hooks {
 		if h == hook {
 			e.hooks = append(e.hooks[:i], e.hooks[i+1:]...)

internal/app/engine/process_noprogress.go

@@ -46,7 +46,7 @@ func (d *NoProgressDetector) process() {
 		duration := float32(timeSinceLastProgress.Seconds())
 		location := d.gcEngine.gcState.TrackerStateGc.Ball.Pos.ToVector2()
 		for _, team := range state.BothTeams() {
-			defenseArea := geom.NewDefenseArea(d.gcEngine.GetGeometry(), *d.gcEngine.currentState.TeamState[team.String()].OnPositiveHalf)
+			defenseArea := geom.NewDefenseArea(d.gcEngine.getGeometry(), *d.gcEngine.currentState.TeamState[team.String()].OnPositiveHalf)
 			if defenseArea.IsPointInside(d.gcEngine.gcState.TrackerStateGc.Ball.Pos.ToVector2()) {
 				d.gcEngine.Enqueue(createGameEventChange(state.GameEvent_KEEPER_HELD_BALL, state.GameEvent{
 					Event: &state.GameEvent_KeeperHeldBall_{

internal/app/engine/process_prepare.go

@@ -105,7 +105,7 @@ func (e *Engine) robotPos(robotId *state.RobotId) *geom.Vector2 {
 func (e *Engine) posInsideGoal(pos *geom.Vector2) bool {
 	forTeam := *e.currentState.Command.ForTeam
 	teamInfo := e.currentState.TeamState[forTeam.String()]
-	goalCenter := geom.GoalCenter(e.GetGeometry(), *teamInfo.OnPositiveHalf)
-	goalArea := geom.NewRectangleFromCenter(goalCenter, robotRadius*2, e.GetGeometry().GoalWidth)
+	goalCenter := geom.GoalCenter(e.getGeometry(), *teamInfo.OnPositiveHalf)
+	goalArea := geom.NewRectangleFromCenter(goalCenter, robotRadius*2, e.getGeometry().GoalWidth)
 	return goalArea.IsPointInside(pos)
 }

internal/app/engine/process_runningtostop.go

@@ -44,14 +44,14 @@ func (e *Engine) ballPlacementRequired() bool {
 	}
 
 	// The ball is inside the field.
-	field := geom.NewRectangleFromCenter(geom.NewVector2(0, 0), e.GetGeometry().FieldLength, e.GetGeometry().FieldWidth)
+	field := geom.NewRectangleFromCenter(geom.NewVector2(0, 0), e.getGeometry().FieldLength, e.getGeometry().FieldWidth)
 	if !field.IsPointInside(ballPos) {
 		return true
 	}
 
 	// The ball is at least 0.7m away from any defense area.
 	for _, sign := range []float64{1, -1} {
-		defenseArea := geom.NewDefenseAreaBySign(e.GetGeometry(), sign)
+		defenseArea := geom.NewDefenseAreaBySign(e.getGeometry(), sign)
 		forbiddenArea := defenseArea.WithMargin(e.gameConfig.BallPlacementMinDistanceToDefenseArea)
 		if forbiddenArea.IsPointInside(ballPos) {
 			return true

internal/app/engine/process_tick.go

@@ -9,6 +9,9 @@ import (
 
 // processTick updates the timers of the state and triggers changes if required
 func (e *Engine) processTick() {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
+
 	currentTime := e.timeProvider()
 	delta := currentTime.Sub(e.lastTimeUpdate)
 	e.lastTimeUpdate = currentTime
@@ -40,8 +43,12 @@ func (e *Engine) processTick() {
 	e.processPrepare()
 	e.processBotNumber()
 
+	stateCopy := e.currentState.Clone()
 	for _, hook := range e.hooks {
-		hook <- HookOut{State: e.CurrentState()}
+		select {
+		case hook <- HookOut{State: stateCopy}:
+		default:
+		}
 	}
 }
 

internal/app/gc/gc.go

@@ -51,9 +51,6 @@ func NewGameController(cfg config.Controller) (c *GameController) {
 
 // Start starts all go routines
 func (c *GameController) Start() {
-	if err := c.gcEngine.Start(); err != nil {
-		panic(err)
-	}
 
 	switch c.config.TimeAcquisitionMode {
 	case config.TimeAcquisitionModeSystem:
@@ -80,6 +77,10 @@ func (c *GameController) Start() {
 	c.autoRefServerTls.Server.Start()
 	c.teamServer.Server.Start()
 	c.teamServerTls.Server.Start()
+
+	if err := c.gcEngine.Start(); err != nil {
+		panic(err)
+	}
 }
 
 // Stop stops all go routines