Merge "Stop NinjaReader from sending new status messages after Close" into main

ui/status/ninja.go

@@ -42,8 +42,9 @@ func NewNinjaReader(ctx logger.Logger, status ToolStatus, fifo string) *NinjaRea
 	n := &NinjaReader{
 		status:     status,
 		fifo:       fifo,
+		forceClose: make(chan bool),
 		done:       make(chan bool),
-		cancel: make(chan bool),
+		cancelOpen: make(chan bool),
 	}
 
 	go n.run()
@@ -54,8 +55,9 @@ func NewNinjaReader(ctx logger.Logger, status ToolStatus, fifo string) *NinjaRea
 type NinjaReader struct {
 	status     ToolStatus
 	fifo       string
+	forceClose chan bool
 	done       chan bool
-	cancel chan bool
+	cancelOpen chan bool
 }
 
 const NINJA_READER_CLOSE_TIMEOUT = 5 * time.Second
@@ -63,18 +65,34 @@ const NINJA_READER_CLOSE_TIMEOUT = 5 * time.Second
 // Close waits for NinjaReader to finish reading from the fifo, or 5 seconds.
 func (n *NinjaReader) Close() {
 	// Signal the goroutine to stop if it is blocking opening the fifo.
-	close(n.cancel)
+	close(n.cancelOpen)
 
+	// Ninja should already have exited or been killed, wait 5 seconds for the FIFO to be closed and any
+	// remaining messages to be processed through the NinjaReader.run goroutine.
 	timeoutCh := time.After(NINJA_READER_CLOSE_TIMEOUT)
-
 	select {
 	case <-n.done:
-		// Nothing
+		return
 	case <-timeoutCh:
-		n.status.Error(fmt.Sprintf("ninja fifo didn't finish after %s", NINJA_READER_CLOSE_TIMEOUT.String()))
+		// Channel is not closed yet
 	}
 
+	n.status.Error(fmt.Sprintf("ninja fifo didn't finish after %s", NINJA_READER_CLOSE_TIMEOUT.String()))
+
+	// Force close the reader even if the FIFO didn't close.
+	close(n.forceClose)
+
+	// Wait again for the reader thread to acknowledge the close before giving up and assuming it isn't going
+	// to send anything else.
+	timeoutCh = time.After(NINJA_READER_CLOSE_TIMEOUT)
+	select {
+	case <-n.done:
 		return
+	case <-timeoutCh:
+		// Channel is not closed yet
+	}
+
+	n.status.Verbose(fmt.Sprintf("ninja fifo didn't finish even after force closing after %s", NINJA_READER_CLOSE_TIMEOUT.String()))
 }
 
 func (n *NinjaReader) run() {
@@ -98,7 +116,7 @@ func (n *NinjaReader) run() {
 	select {
 	case f = <-fileCh:
 		// Nothing
-	case <-n.cancel:
+	case <-n.cancelOpen:
 		return
 	}
 
@@ -108,6 +126,12 @@ func (n *NinjaReader) run() {
 
 	running := map[uint32]*Action{}
 
+	msgChan := make(chan *ninja_frontend.Status)
+
+	// Read from the ninja fifo and decode the protobuf in a goroutine so the main NinjaReader.run goroutine
+	// can listen
+	go func() {
+		defer close(msgChan)
 		for {
 			size, err := readVarInt(r)
 			if err != nil {
@@ -135,6 +159,25 @@ func (n *NinjaReader) run() {
 				continue
 			}
 
+			msgChan <- msg
+		}
+	}()
+
+	for {
+		var msg *ninja_frontend.Status
+		var msgOk bool
+		select {
+		case <-n.forceClose:
+			// Close() has been called, but the reader goroutine didn't get EOF after 5 seconds
+			break
+		case msg, msgOk = <-msgChan:
+			// msg is ready or closed
+		}
+
+		if !msgOk {
+			// msgChan is closed
+			break
+		}
 		// Ignore msg.BuildStarted
 		if msg.TotalEdges != nil {
 			n.status.SetTotalActions(int(msg.TotalEdges.GetTotalEdges()))