Bugfix: Deadlock in CI mode

Author: g3force

internal/app/api/server.go

@@ -70,10 +70,10 @@ func (a *Server) WsHandler(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *ServerConnection) publish() {
-	hook := make(chan engine.HookOut)
-	s.gcEngine.RegisterHook(hook)
+	hook := make(chan engine.HookOut, 10)
+	s.gcEngine.RegisterHook("apiServer", hook)
 	defer func() {
-		s.gcEngine.UnregisterHook(hook)
+		s.gcEngine.UnregisterHook("apiServer")
 		close(hook)
 	}()
 

internal/app/ci/ci.go

@@ -83,8 +83,9 @@ func (s *Server) serve(conn net.Conn) {
 			s.latestTime = time.Unix(sec, nSec)
 			s.mutex.Unlock()
 			select {
-			case s.tickChan <- time.Now():
-			default:
+			case s.tickChan <- s.latestTime:
+			case <-time.After(1 * time.Second):
+				log.Printf("tickChan unresponsive! Failed to sent %v", s.latestTime)
 			}
 		}
 	}

internal/app/engine/engine.go

@@ -23,7 +23,7 @@ type Engine struct {
 	currentState             *state.State
 	stateMachine             *statemachine.StateMachine
 	queue                    chan *statemachine.Change
-	hooks                    []chan HookOut
+	hooks                    map[string]chan HookOut
 	timeProvider             timer.TimeProvider
 	lastTimeUpdate           time.Time
 	gcState                  *GcState
@@ -40,7 +40,7 @@ func NewEngine(gameConfig config.Game) (e *Engine) {
 	e.stateStore = store.NewStore(gameConfig.StateStoreFile)
 	e.stateMachine = statemachine.NewStateMachine(gameConfig)
 	e.queue = make(chan *statemachine.Change, 100)
-	e.hooks = []chan HookOut{}
+	e.hooks = map[string]chan HookOut{}
 	e.SetTimeProvider(func() time.Time { return time.Now() })
 	e.gcState = new(GcState)
 	e.gcState.TeamState = map[string]*GcStateTeam{
@@ -241,10 +241,12 @@ func (e *Engine) processChange(change *statemachine.Change) {
 		log.Println("Could not add new state to store: ", err)
 	}
 	stateCopy := e.currentState.Clone()
-	for _, hook := range e.hooks {
+	hookOut := HookOut{Change: entry.Change, State: stateCopy}
+	for name, hook := range e.hooks {
 		select {
-		case hook <- HookOut{Change: entry.Change, State: stateCopy}:
-		default:
+		case hook <- hookOut:
+		case <-time.After(1 * time.Second):
+			log.Printf("Hook %v unresponsive! Failed to sent %v", name, hookOut)
 		}
 	}
 

internal/app/engine/engine_test.go

@@ -26,7 +26,7 @@ func Test_Engine(t *testing.T) {
 	gameConfig.StateStoreFile = tmpDir + "/store.json.stream"
 	engine := NewEngine(gameConfig)
 	hook := make(chan HookOut)
-	engine.RegisterHook(hook)
+	engine.RegisterHook("engineTest", hook)
 	if err := engine.Start(); err != nil {
 		t.Fatal("Could not start engine")
 	}
@@ -41,7 +41,7 @@ func Test_Engine(t *testing.T) {
 	})
 	// wait for the changes to be processed
 	<-hook
-	engine.UnregisterHook(hook)
+	engine.UnregisterHook("engineTest")
 	engine.Stop()
 
 	wantNewState.Command = state.NewCommandNeutral(state.Command_HALT)

internal/app/engine/hook.go

@@ -3,6 +3,7 @@ package engine
 import (
 	"github.com/RoboCup-SSL/ssl-game-controller/internal/app/state"
 	"github.com/RoboCup-SSL/ssl-game-controller/internal/app/statemachine"
+	"log"
 )
 
 type HookOut struct {
@@ -11,25 +12,24 @@ type HookOut struct {
 }
 
 // RegisterHook registers given hook for post processing after each change
-func (e *Engine) RegisterHook(hook chan HookOut) {
+func (e *Engine) RegisterHook(name string, hook chan HookOut) {
 	e.mutex.Lock()
 	defer e.mutex.Unlock()
-	e.hooks = append(e.hooks, hook)
+	if _, ok := e.hooks[name]; ok {
+		log.Printf("Hook %v already registered!", name)
+	}
+	e.hooks[name] = hook
 }
 
 // UnregisterHook unregisters hooks that were registered before
-func (e *Engine) UnregisterHook(hook chan HookOut) bool {
+func (e *Engine) UnregisterHook(name string) {
 	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:]...)
-			select {
-			case <-hook:
-			default:
-			}
-			return true
+	if hook, ok := e.hooks[name]; ok {
+		select {
+		case <-hook:
+		default:
 		}
 	}
-	return false
+	delete(e.hooks, name)
 }

internal/app/engine/process_tick.go

@@ -4,6 +4,7 @@ import (
 	"github.com/RoboCup-SSL/ssl-game-controller/internal/app/state"
 	"github.com/RoboCup-SSL/ssl-game-controller/internal/app/statemachine"
 	"github.com/golang/protobuf/ptypes"
+	"log"
 	"time"
 )
 
@@ -63,10 +64,12 @@ func (e *Engine) processTick() {
 	e.processPenalty()
 
 	stateCopy := e.currentState.Clone()
+	hookOut := HookOut{State: stateCopy}
 	for _, hook := range e.hooks {
 		select {
-		case hook <- HookOut{State: stateCopy}:
-		default:
+		case hook <- hookOut:
+		case <-time.After(1 * time.Second):
+			log.Printf("processTick: Hook unresponsive! Failed to sent %v", hookOut)
 		}
 	}
 }

internal/app/gc/gc.go

@@ -71,7 +71,7 @@ func (c *GameController) Start() {
 		c.ciServer.Start()
 	}
 
-	c.gcEngine.RegisterHook(c.messageGenerator.EngineHook)
+	c.gcEngine.RegisterHook("messageGen", c.messageGenerator.EngineHook)
 	c.messageGenerator.Start()
 	c.autoRefServer.Server.Start()
 	c.autoRefServerTls.Server.Start()
@@ -86,7 +86,7 @@ func (c *GameController) Start() {
 // Stop stops all go routines
 func (c *GameController) Stop() {
 	// Note: Stopping is not (yet) implemented correctly by all servers.
-	c.gcEngine.UnregisterHook(c.messageGenerator.EngineHook)
+	c.gcEngine.UnregisterHook("messageGen")
 	c.messageGenerator.Stop()
 	c.ciServer.Stop()
 	c.trackerReceiver.Stop()