Fixed some weird behaviour in gRPC Monitor command (#1706)

* Made monitor.jsonDecodeLoop more thread resilient

* Use idiomatic way to check for channel-closed event

* Ensure that monitor processes are killed and collected correctly

* Simplified (and fixed...) monitor stream handling

There is no need to spawn two goroutines, one is enough. Also the extra
context is not needed anymore. The monitor port can be closed when the
goroutine with the grpc recv loop ends.

* Fix cli crash if no configuration is provided in the Monitor gRPC call

* Extended term_example gRPC API test coverage

Now it cycles on the same port two times.

* Reduce timeout for monitor close/quit commands to 250ms

Author: cmaglie

arduino/monitor/monitor.go

@@ -103,38 +103,39 @@ func (mon *PluggableMonitor) String() string {
 	return mon.id
 }
 
-func (mon *PluggableMonitor) jsonDecodeLoop(in io.Reader, outChan chan<- *monitorMessage) {
+func jsonDecodeLoop(in io.Reader, outChan chan<- *monitorMessage, log *logrus.Entry, lastError *error) {
 	decoder := json.NewDecoder(in)
 
 	for {
 		var msg monitorMessage
 		if err := decoder.Decode(&msg); err != nil {
-			mon.incomingMessagesError = err
+			*lastError = err
 			close(outChan)
-			mon.log.Errorf("stopped decode loop: %s", err)
+			log.Errorf("stopped decode loop: %s", err)
 			return
 		}
-		mon.log.
-			WithField("event_type", msg.EventType).
+		log.WithField("event_type", msg.EventType).
 			WithField("message", msg.Message).
 			WithField("error", msg.Error).
 			Infof("received message")
 		if msg.EventType == "port_closed" {
-			mon.log.Infof("monitor port has been closed externally")
+			log.Infof("monitor port has been closed externally")
 		} else {
 			outChan <- &msg
 		}
 	}
 }
 
 func (mon *PluggableMonitor) waitMessage(timeout time.Duration, expectedEvt string) (*monitorMessage, error) {
+	mon.log.WithField("expected", expectedEvt).Debugf("waiting for event")
 	var msg *monitorMessage
 	select {
-	case msg = <-mon.incomingMessagesChan:
-		if msg == nil {
+	case m, ok := <-mon.incomingMessagesChan:
+		if !ok {
 			// channel has been closed
 			return nil, mon.incomingMessagesError
 		}
+		msg = m
 	case <-time.After(timeout):
 		return nil, fmt.Errorf(tr("timeout waiting for message"))
 	}
@@ -192,22 +193,21 @@ func (mon *PluggableMonitor) runProcess() error {
 
 	messageChan := make(chan *monitorMessage)
 	mon.incomingMessagesChan = messageChan
-	go mon.jsonDecodeLoop(stdout, messageChan)
+	go jsonDecodeLoop(stdout, messageChan, mon.log, &mon.incomingMessagesError)
 
 	mon.log.Infof("Monitor process started successfully!")
 	return nil
 }
 
-func (mon *PluggableMonitor) killProcess() error {
+func (mon *PluggableMonitor) killProcess() {
 	mon.log.Infof("Killing monitor process")
 	if err := mon.process.Kill(); err != nil {
-		return err
+		mon.log.WithError(err).Error("Sent kill signal")
 	}
 	if err := mon.process.Wait(); err != nil {
-		return err
+		mon.log.WithError(err).Error("Waiting for process end")
 	}
-	mon.log.Infof("Monitor process killed successfully!")
-	return nil
+	mon.log.Infof("Monitor process killed")
 }
 
 // Run starts the monitor executable process and sends the HELLO command to the monitor to agree on the
@@ -220,15 +220,10 @@ func (mon *PluggableMonitor) Run() (err error) {
 
 	defer func() {
 		// If the monitor process is started successfully but the HELLO handshake
-		// fails the monitor is an unusable state, we kill the process to avoid
+		// fails the monitor is in an unusable state, we kill the process to avoid
 		// further issues down the line.
-		if err == nil {
-			return
-		}
-		if killErr := mon.killProcess(); killErr != nil {
-			// Log failure to kill the process, ideally that should never happen
-			// but it's best to know it if it does
-			mon.log.Errorf("Killing monitor after unsuccessful start: %s", killErr)
+		if err != nil {
+			mon.killProcess()
 		}
 	}()
 
@@ -297,20 +292,19 @@ func (mon *PluggableMonitor) Close() error {
 	if err := mon.sendCommand("CLOSE\n"); err != nil {
 		return err
 	}
-	_, err := mon.waitMessage(time.Second*10, "close")
+	_, err := mon.waitMessage(time.Millisecond*250, "close")
 	return err
 }
 
 // Quit terminates the monitor. No more commands can be accepted by the monitor.
 func (mon *PluggableMonitor) Quit() error {
+	defer mon.killProcess() // ensure that killProcess is called in any case...
+
 	if err := mon.sendCommand("QUIT\n"); err != nil {
 		return err
 	}
-	if _, err := mon.waitMessage(time.Second*10, "quit"); err != nil {
+	if _, err := mon.waitMessage(time.Millisecond*250, "quit"); err != nil {
 		return err
 	}
-	if err := mon.killProcess(); err != nil {
-		mon.log.WithError(err).Info("error killing monitor process")
-	}
 	return nil
 }

commands/daemon/daemon.go

@@ -508,9 +508,11 @@ func (s *ArduinoCoreServerImpl) Monitor(stream rpc.ArduinoCoreService_MonitorSer
 	if err != nil {
 		return err
 	}
-	ctx, cancel := context.WithCancel(stream.Context())
+
 	go func() {
-		defer cancel()
+		// close port on gRPC call EOF or errors
+		defer portProxy.Close()
+
 		for {
 			msg, err := stream.Recv()
 			if errors.Is(err, io.EOF) {
@@ -541,23 +543,20 @@ func (s *ArduinoCoreServerImpl) Monitor(stream rpc.ArduinoCoreService_MonitorSer
 			}
 		}
 	}()
-	go func() {
-		defer cancel()
-		buff := make([]byte, 4096)
-		for {
-			n, err := portProxy.Read(buff)
-			if errors.Is(err, io.EOF) {
-				return
-			}
-			if err != nil {
-				stream.Send(&rpc.MonitorResponse{Error: err.Error()})
-				return
-			}
-			if err := stream.Send(&rpc.MonitorResponse{RxData: buff[:n]}); err != nil {
-				return
-			}
+
+	buff := make([]byte, 4096)
+	for {
+		n, err := portProxy.Read(buff)
+		if errors.Is(err, io.EOF) {
+			break
 		}
-	}()
-	<-ctx.Done()
+		if err != nil {
+			stream.Send(&rpc.MonitorResponse{Error: err.Error()})
+			break
+		}
+		if err := stream.Send(&rpc.MonitorResponse{RxData: buff[:n]}); err != nil {
+			break
+		}
+	}
 	return nil
 }

commands/daemon/term_example/main.go

@@ -74,52 +74,54 @@ func main() {
 	}
 	fmt.Println("Detected port:", port.Label, port.ProtocolLabel)
 
+	connectToPort(cli, instance, port)
+	time.Sleep(5 * time.Second)
+	connectToPort(cli, instance, port)
+	time.Sleep(5 * time.Second)
+}
+
+func connectToPort(cli commands.ArduinoCoreServiceClient, instance *commands.Instance, port *commands.Port) {
 	// Connect to the port monitor
-	fmt.Println("Connecting to monitor")
+	fmt.Println("Connecting to port", port)
+
 	ctx, cancel := context.WithCancel(context.Background())
-	if respStream, err := cli.Monitor(ctx); err != nil {
-		log.Fatal("Monitor:", err)
-	} else {
-		if err := respStream.Send(&commands.MonitorRequest{
-			Instance: instance,
-			Port:     port,
-		}); err != nil {
-			log.Fatal("Monitor send-config:", err)
-		}
-		time.Sleep(1 * time.Second)
-
-		go func() {
-			for {
-				if resp, err := respStream.Recv(); err != nil {
-					fmt.Println("     RECV:", err)
-					break
-				} else {
-					fmt.Println("     RECV:", resp)
-				}
-			}
-		}()
+	monitorClient, err := cli.Monitor(ctx)
+	if err != nil {
+		log.Fatal("Error opening Monitor:", err)
+	}
+	if err := monitorClient.Send(&commands.MonitorRequest{
+		Instance: instance,
+		Port:     port,
+	}); err != nil {
+		log.Fatal("Error sending Monitor config:", err)
+	}
 
-		hello := &commands.MonitorRequest{
-			TxData: []byte("HELLO!"),
-		}
-		fmt.Println("Send:", hello)
-		if err := respStream.Send(hello); err != nil {
-			log.Fatal("Monitor send HELLO:", err)
+	go func() {
+		for {
+			resp, err := monitorClient.Recv()
+			if err != nil {
+				fmt.Println("     RECV-ERR:", err)
+				break
+			}
+			fmt.Println("     RECV:", resp)
 		}
+	}()
 
-		fmt.Println("Send:", hello)
-		if err := respStream.Send(hello); err != nil {
-			log.Fatal("Monitor send HELLO:", err)
-		}
+	hello := &commands.MonitorRequest{
+		TxData: []byte("HELLO!"),
+	}
+	fmt.Println("Send:", hello)
+	if err := monitorClient.Send(hello); err != nil {
+		log.Fatal("Monitor send HELLO:", err)
+	}
 
-		time.Sleep(5 * time.Second)
+	time.Sleep(15 * time.Second)
 
-		fmt.Println("Closing Monitor")
-		if err := respStream.CloseSend(); err != nil {
-			log.Fatal("Monitor close send:", err)
-		}
-		time.Sleep(5 * time.Second)
+	fmt.Println("Closing Monitor")
+	if err := monitorClient.CloseSend(); err != nil {
+		log.Fatal("Monitor close send:", err)
 	}
+	<-monitorClient.Context().Done()
+
 	cancel()
-	time.Sleep(5 * time.Second)
 }

commands/monitor/monitor.go

@@ -86,10 +86,11 @@ func Monitor(ctx context.Context, req *rpc.MonitorRequest) (*PortProxy, *pluggab
 		m.Quit()
 		return nil, nil, &arduino.FailedMonitorError{Cause: err}
 	}
-
-	for _, setting := range req.GetPortConfiguration().Settings {
-		if err := m.Configure(setting.SettingId, setting.Value); err != nil {
-			logrus.Errorf("Could not set configuration %s=%s: %s", setting.SettingId, setting.Value, err)
+	if portConfig := req.GetPortConfiguration(); portConfig != nil {
+		for _, setting := range portConfig.Settings {
+			if err := m.Configure(setting.SettingId, setting.Value); err != nil {
+				logrus.Errorf("Could not set configuration %s=%s: %s", setting.SettingId, setting.Value, err)
+			}
 		}
 	}