ec3po: Add optional shutdown_pipe StartLoop() arguments.

When ec3po.console.StartLoop() or ec3po.interpreter.StartLoop() is given a
shutdown_pipe file object, it will exit the loop when that file becomes
readable (unblocked), and will close the file upon loop exit (as they do
with the other files they poll).

This will be used by the servod-side of ec3po (in hdctools repo) to replace
use of multiprocessing.Process.terminate(), because that has no threading
equivalent, and I am migrating ec3po to use threads instead of
subprocesses.

BRANCH=none
BUG=b:79684405
TEST=With a corresponding servod-side change in hdctools to use
shutdown_pipe args instead of terminate(), servod shutdown via either
ctrl+c or SIGTERM still happens correctly, without any delay, leftover
processes, or tracebacks.  Testing performed with a servo_micro attached to
an octopus_ite.

Change-Id: I82db2fd60620ac2a05a4d09afe263a57c141c615
Signed-off-by: Matthew Blecker <matthewb@chromium.org>
Reviewed-on: https://chromium-review.googlesource.com/1277615
Reviewed-by: Ruben Rodriguez Buchillon <coconutruben@chromium.org>

2 files changed, 75 insertions, 40 deletions

diff --git a/util/ec3po/console.py b/util/ec3po/console.py
index 3657ceb20e..29fc705d61 100755
--- a/util/ec3po/console.py
+++ b/util/ec3po/console.py
@@ -817,13 +817,16 @@ def IsPrintable(byte):
   """
   return byte >= ord(' ') and byte <= ord('~')
 
-def StartLoop(console, command_active):
+def StartLoop(console, command_active, shutdown_pipe=None):
   """Starts the infinite loop of console processing.
 
   Args:
     console: A Console object that has been properly initialzed.
     command_active: multiprocessing.Value indicating if servod owns
         the console, or user owns the console. This prevents input collisions.
+    shutdown_pipe: A file object for a pipe or equivalent that becomes readable
+      (not blocked) to indicate that the loop should exit.  Can be None to never
+      exit the loop.
   """
   try:
     console.logger.debug('Console is being served on %s.', console.user_pty)
@@ -836,7 +839,11 @@ def StartLoop(console, command_active):
     ep = select.epoll()
     ep.register(console.master_pty, select.EPOLLHUP)
 
-    while True:
+    # This is used instead of "break" to avoid exiting the loop in the middle of
+    # an iteration.
+    continue_looping = True
+
+    while continue_looping:
       # Check to see if pts is connected to anything
       events = ep.poll(0)
       master_connected = not events
@@ -846,6 +853,8 @@ def StartLoop(console, command_active):
                    console.cmd_pipe, console.dbg_pipe]
       if master_connected:
         read_list.append(console.master_pty)
+      if shutdown_pipe is not None:
+        read_list.append(shutdown_pipe)
 
       # Check if any input is ready, or wait for .1 sec and re-poll if
       # a user has connected to the pts.
@@ -855,28 +864,30 @@ def StartLoop(console, command_active):
       ready_for_reading = select_output[0]
 
       for obj in ready_for_reading:
-        if obj is console.master_pty and not command_active.value:
-          # Convert to bytes so we can look for non-printable chars such as
-          # Ctrl+A, Ctrl+E, etc.
-          try:
-            line = bytearray(os.read(console.master_pty, CONSOLE_MAX_READ))
-            console.logger.debug('Input from user: %s, locked:%s',
+        if obj is console.master_pty:
+          if not command_active.value:
+            # Convert to bytes so we can look for non-printable chars such as
+            # Ctrl+A, Ctrl+E, etc.
+            try:
+              line = bytearray(os.read(console.master_pty, CONSOLE_MAX_READ))
+              console.logger.debug('Input from user: %s, locked:%s',
+                  str(line).strip(), command_active.value)
+              for i in line:
+                # Handle each character as it arrives.
+                console.HandleChar(i)
+            except OSError:
+              console.logger.debug('Ptm read failed, probably user disconnect.')
+
+        elif obj is console.interface_pty:
+          if command_active.value:
+            # Convert to bytes so we can look for non-printable chars such as
+            # Ctrl+A, Ctrl+E, etc.
+            line = bytearray(os.read(console.interface_pty, CONSOLE_MAX_READ))
+            console.logger.debug('Input from interface: %s, locked:%s',
                 str(line).strip(), command_active.value)
             for i in line:
               # Handle each character as it arrives.
               console.HandleChar(i)
-          except OSError:
-            console.logger.debug('Ptm read failed, probably user disconnect.')
-
-        if obj is console.interface_pty and command_active.value:
-          # Convert to bytes so we can look for non-printable chars such as
-          # Ctrl+A, Ctrl+E, etc.
-          line = bytearray(os.read(console.interface_pty, CONSOLE_MAX_READ))
-          console.logger.debug('Input from interface: %s, locked:%s',
-              str(line).strip(), command_active.value)
-          for i in line:
-            # Handle each character as it arrives.
-            console.HandleChar(i)
 
         elif obj is console.cmd_pipe:
           data = console.cmd_pipe.recv()
@@ -904,6 +915,11 @@ def StartLoop(console, command_active):
           if command_active.value:
             os.write(console.interface_pty, data)
 
+        elif obj is shutdown_pipe:
+          console.logger.debug(
+              'ec3po console received shutdown pipe unblocked notification')
+          continue_looping = False
+
       while not console.oobm_queue.empty():
         console.logger.debug('OOBM queue ready for reading.')
         console.ProcessOOBMQueue()
@@ -916,14 +932,13 @@ def StartLoop(console, command_active):
     traceback.print_exc()
 
   finally:
-    # Unregister poll.
     ep.unregister(console.master_pty)
-    # Close pipes.
     console.dbg_pipe.close()
     console.cmd_pipe.close()
-    # Close file descriptor.
     os.close(console.master_pty)
     os.close(console.interface_pty)
+    if shutdown_pipe is not None:
+      shutdown_pipe.close()
     console.logger.debug('Exit ec3po console loop for %s', console.user_pty)
 
 
diff --git a/util/ec3po/interpreter.py b/util/ec3po/interpreter.py
index 9af5a67dd9..3957466dbf 100644
--- a/util/ec3po/interpreter.py
+++ b/util/ec3po/interpreter.py
@@ -194,15 +194,15 @@ class Interpreter(object):
           self.cmd_retries = COMMAND_RETRIES
           self.last_cmd = ''
         # Get the UART that the interpreter is attached to.
-        fd = self.ec_uart_pty
-        self.logger.debug('fd: %r', fd)
+        fileobj = self.ec_uart_pty
+        self.logger.debug('fileobj: %r', fileobj)
         # Remove the descriptor from the inputs and outputs.
-        self.inputs.remove(fd)
-        if fd in self.outputs:
-          self.outputs.remove(fd)
-        self.logger.debug('Removed fd. Remaining inputs: %r', self.inputs)
+        self.inputs.remove(fileobj)
+        if fileobj in self.outputs:
+          self.outputs.remove(fileobj)
+        self.logger.debug('Removed fileobj. Remaining inputs: %r', self.inputs)
         # Close the file.
-        fd.close()
+        fileobj.close()
         # Mark the interpreter as disconnected now.
         self.connected = False
         self.logger.debug('Disconnected from %s.', self.ec_uart_pty_name)
@@ -212,12 +212,12 @@ class Interpreter(object):
       if not self.connected:
         self.logger.debug('UART reconnect request.')
         # Reopen the PTY.
-        fd = open(self.ec_uart_pty_name, 'a+')
-        self.logger.debug('fd: %r', fd)
-        self.ec_uart_pty = fd
+        fileobj = open(self.ec_uart_pty_name, 'a+')
+        self.logger.debug('fileobj: %r', fileobj)
+        self.ec_uart_pty = fileobj
         # Add the descriptor to the inputs.
-        self.inputs.append(fd)
-        self.logger.debug('fd added. curr inputs: %r', self.inputs)
+        self.inputs.append(fileobj)
+        self.logger.debug('fileobj added. curr inputs: %r', self.inputs)
         # Mark the interpreter as connected now.
         self.connected = True
         self.logger.debug('Connected to %s.', self.ec_uart_pty_name)
@@ -373,7 +373,7 @@ def Crc8(data):
   return crc >> 8
 
 
-def StartLoop(interp):
+def StartLoop(interp, shutdown_pipe=None):
   """Starts an infinite loop of servicing the user and the EC.
 
   StartLoop checks to see if there are any commands to process, processing them
@@ -388,10 +388,25 @@ def StartLoop(interp):
 
   Args:
     interp: An Interpreter object that has been properly initialised.
+    shutdown_pipe: A file object for a pipe or equivalent that becomes readable
+      (not blocked) to indicate that the loop should exit.  Can be None to never
+      exit the loop.
   """
   try:
-    while True:
-      readable, writeable, _ = select.select(interp.inputs, interp.outputs, [])
+    # This is used instead of "break" to avoid exiting the loop in the middle of
+    # an iteration.
+    continue_looping = True
+
+    while continue_looping:
+      # The inputs list is created anew in each loop iteration because the
+      # Interpreter class sometimes modifies the interp.inputs list.
+      if shutdown_pipe is None:
+        inputs = interp.inputs
+      else:
+        inputs = list(interp.inputs)
+        inputs.append(shutdown_pipe)
+
+      readable, writeable, _ = select.select(inputs, interp.outputs, [])
 
       for obj in readable:
         # Handle any debug prints from the EC.
@@ -402,6 +417,11 @@ def StartLoop(interp):
         elif obj is interp.cmd_pipe:
           interp.HandleUserData()
 
+        elif obj is shutdown_pipe:
+          interp.logger.debug(
+              'ec3po console received shutdown pipe unblocked notification')
+          continue_looping = False
+
       for obj in writeable:
         # Send a command to the EC.
         if obj is interp.ec_uart_pty:
@@ -415,10 +435,10 @@ def StartLoop(interp):
     traceback.print_exc()
 
   finally:
-    # Close pipes.
     interp.cmd_pipe.close()
     interp.dbg_pipe.close()
-    # Close file descriptor.
     interp.ec_uart_pty.close()
+    if shutdown_pipe is not None:
+      shutdown_pipe.close()
     interp.logger.debug('Exit ec3po interpreter loop for %s',
                         interp.ec_uart_pty_name)