Fix blocking issue with multiple sessions

When multiple sessions were run serially, the second would block
forever. The Watcher type sends Kubernetes events for a specific
session over a channel to the executor. During clean-up the Watcher
continued to post events, but the executor never read these. Since the
Watcher used an unbuffered channel, it blocked until it was read. This
blocking caused a lock to never be released. This lock was needed to
create a new channel for a new session.

This change switches the Watcher to use a buffered channel which does
not block until it is full. The Watcher is also modified to drop events
when the channel's capacity is exceeded. This prevents it from blocking
during manual tests and allows additional sessions to be scheduled.

In addition, this change adds a missing call to (*queue).Done in the
controller. This call is necessary to instruct the reservation system
that machines are available for scheduling.

Author: Ben Reed

testctrl/go.sum

@@ -52,6 +52,7 @@ github.com/golang/protobuf v1.3.5 h1:F768QJ1E9tib+q5Sc8MkdJi1RxLTbRcTf8LJV56aRls
 github.com/golang/protobuf v1.3.5/go.mod h1:6O5/vntMXwX2lRkT1hjjk0nAC1IDOTvTlVgjlRvqsdk=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
+github.com/google/glog v0.4.0 h1:WV2GdGOpRcDyRt1i9LHUcpATSfmbxDOHL/I5OtjndLI=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0 h1:crn/baboCvb5fXaQ0IJ1SGTsTVrWpDsCWC8EGETZijY=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=

testctrl/svc/orch/controller.go

@@ -228,7 +228,10 @@ func (c *Controller) spawnExecutor(session *types.Session) {
 	c.incExecutors()
 
 	go func() {
-		defer c.decExecutors()
+		defer func() {
+			c.decExecutors()
+			c.waitQueue.Done(session)
+		}()
 
 		if err := executor.Execute(session); err != nil {
 			glog.Infof("%v", err)

testctrl/svc/orch/watcher.go

@@ -71,7 +71,7 @@ func (w *Watcher) Subscribe(sessionName string) (<-chan *PodWatchEvent, error) {
 		return nil, fmt.Errorf("session %v already has a follower", sessionName)
 	}
 
-	eventChan := make(chan *PodWatchEvent)
+	eventChan := make(chan *PodWatchEvent, eventBufferSize)
 	w.eventChans[sessionName] = eventChan
 	return eventChan, nil
 }
@@ -133,7 +133,16 @@ func (w *Watcher) publish(event *PodWatchEvent) {
 	defer w.mux.Unlock()
 
 	eventChan := w.eventChans[event.SessionName]
-	eventChan <- event
+	if eventChan == nil {
+		glog.Warningf("watcher: received event for session without subscriber: %v", event)
+		return
+	}
+
+	if len(eventChan) < cap(eventChan) {
+		eventChan <- event
+	} else {
+		glog.Warningf("watcher: too many events unread in subscriber channel, dropping: %v", event)
+	}
 }
 
 func (w *Watcher) diagnose(pod *corev1.Pod) (Health, error) {
@@ -192,3 +201,5 @@ type PodWatchEvent struct {
 	// Error may provide the error details that led to the failing health.
 	Error error
 }
+
+const eventBufferSize = 32