fix(api): Synchronize access to the smoothie and rpc (#3528)

In smoothie: Only allow one thing at a time to send a command.

This is tricky because there's no clean interfaces anywhere in apiv1; it's hard
to determine where the right place for a lock is. This commit puts the lock
around the lowest level primitive, _send_command. That means that two threads
racing to execute e.g. pipette.home() and pipette.move() (from rpc
calibration.move_to_front()) will interleave the subcommands around setting
speeds, setting currents, and actually moving. However, they will no longer put
the smoothie driver or serial link into a bad state. Is this an improvement?
Maybe, but it's not everything - we should probably have similar locks around
rpc actions (aside from cancel of course) and http.

In RPC: Synchronize access to motion in rpc

In addition to the low-level of synchronization in the smoothie driver to
prevent reentrant access to the serial port, we also need to synchronize
higher-level compound actions. For instance, if someone starts a tip probe via
rpc and then commands a home via http, one action should complete before the
next one starts. This still may lead to bad behavior, but there's a limited
amount we can do about interrupting flows high level enough that the api doesn't
even consider them - that's on a client.

This synchronization is applied to both certain endpoints in the control
endpoints via http and any motion-related rpc endpoints (as well as simulate(),
unfortunately, since at least on apiv1 that would conflict with other motions
since it shares the same robot singleton. We can remove this when we switch to
apiv2).

Closes #3527

api/src/opentrons/api/calibration.py

@@ -1,3 +1,4 @@
+import functools
 import logging
 from copy import copy
 
@@ -25,18 +26,33 @@ def _well0(cont):
         return cont.wells(0)
 
 
+def _require_lock(func):
+    """ Decorator to make a function require a lock. Only works for instance
+    methods of CalibrationManager """
+    @functools.wraps(func)
+    def decorated(*args, **kwargs):
+        self = args[0]
+        if self._lock:
+            with self._lock:
+                return func(*args, **kwargs)
+        else:
+            return func(*args, **kwargs)
+    return decorated
+
+
 class CalibrationManager:
     """
     Serves endpoints that are primarily used in
     opentrons/app/ui/robot/api-client/client.js
     """
     TOPIC = 'calibration'
 
-    def __init__(self, hardware, loop=None, broker=None):
+    def __init__(self, hardware, loop=None, broker=None, lock=None):
         self._broker = broker or Broker()
         self._hardware = hardware
         self._loop = loop
         self.state = None
+        self._lock = lock
 
     def _set_state(self, state):
         if state not in VALID_STATES:
@@ -45,6 +61,7 @@ def _set_state(self, state):
         self.state = state
         self._on_state_changed()
 
+    @_require_lock
     def tip_probe(self, instrument):
         inst = instrument._instrument
         log.info('Probing tip with {}'.format(instrument.name))
@@ -82,6 +99,7 @@ def tip_probe(self, instrument):
         self.move_to_front(instrument)
         self._set_state('ready')
 
+    @_require_lock
     def pick_up_tip(self, instrument, container):
         if not isinstance(container, Container):
             raise ValueError(
@@ -100,6 +118,7 @@ def pick_up_tip(self, instrument, container):
             inst.pick_up_tip(_well0(container._container))
         self._set_state('ready')
 
+    @_require_lock
     def drop_tip(self, instrument, container):
         if not isinstance(container, Container):
             raise ValueError(
@@ -118,6 +137,7 @@ def drop_tip(self, instrument, container):
             inst.drop_tip(_well0(container._container))
         self._set_state('ready')
 
+    @_require_lock
     def return_tip(self, instrument):
         inst = instrument._instrument
         log.info('Returning tip from {}'.format(instrument.name))
@@ -130,6 +150,7 @@ def return_tip(self, instrument):
             inst.return_tip()
         self._set_state('ready')
 
+    @_require_lock
     def move_to_front(self, instrument):
         inst = instrument._instrument
         log.info('Moving {}'.format(instrument.name))
@@ -153,6 +174,7 @@ def move_to_front(self, instrument):
                 inst, inst.robot)
         self._set_state('ready')
 
+    @_require_lock
     def move_to(self, instrument, container):
         if not isinstance(container, Container):
             raise ValueError(
@@ -176,6 +198,7 @@ def move_to(self, instrument, container):
 
         self._set_state('ready')
 
+    @_require_lock
     def jog(self, instrument, distance, axis):
         inst = instrument._instrument
         log.info('Jogging {} by {} in {}'.format(
@@ -192,6 +215,7 @@ def jog(self, instrument, distance, axis):
                 robot=inst.robot)
         self._set_state('ready')
 
+    @_require_lock
     def home(self, instrument):
         inst = instrument._instrument
         log.info('Homing {}'.format(instrument.name))
@@ -204,6 +228,7 @@ def home(self, instrument):
             inst.home()
         self._set_state('ready')
 
+    @_require_lock
     def update_container_offset(self, container, instrument):
         inst = instrument._instrument
         log.info('Updating {} in {}'.format(container.name, container.slot))

api/src/opentrons/api/routers.py

@@ -6,7 +6,7 @@
 
 
 class MainRouter:
-    def __init__(self, hardware=None, loop=None):
+    def __init__(self, hardware=None, loop=None, lock=None):
         topics = [Session.TOPIC, CalibrationManager.TOPIC]
         self._broker = Broker()
         self._notifications = Notifications(topics, self._broker, loop=loop)
@@ -16,10 +16,12 @@ def __init__(self, hardware=None, loop=None):
         self.session_manager = SessionManager(
             hardware=hardware,
             loop=loop,
-            broker=self._broker)
+            broker=self._broker,
+            lock=lock)
         self.calibration_manager = CalibrationManager(hardware=hardware,
                                                       loop=loop,
-                                                      broker=self._broker)
+                                                      broker=self._broker,
+                                                      lock=lock)
 
     @property
     def notifications(self):

api/src/opentrons/api/session.py

@@ -1,7 +1,7 @@
 import ast
 import asyncio
 from copy import copy
-from functools import reduce
+from functools import reduce, wraps
 import json
 import logging
 from time import time
@@ -28,8 +28,22 @@
 VALID_STATES = {'loaded', 'running', 'finished', 'stopped', 'paused', 'error'}
 
 
+def _motion_lock(func):
+    """ Decorator to make a function require a lock. Only works for instance
+    methods of Session (or SessionManager) """
+    @wraps(func)
+    def decorated(*args, **kwargs):
+        self = args[0]
+        if self._motion_lock:
+            with self._motion_lock:
+                return func(*args, **kwargs)
+        else:
+            return func(*args, **kwargs)
+    return decorated
+
+
 class SessionManager(object):
-    def __init__(self, hardware, loop=None, broker=None):
+    def __init__(self, hardware, loop=None, broker=None, lock=None):
         self._broker = broker or Broker()
         self._loop = loop or asyncio.get_event_loop()
         self.session = None
@@ -38,6 +52,7 @@ def __init__(self, hardware, loop=None, broker=None):
         self._command_logger = logging.getLogger(
             'opentrons.server.command_logger')
         self._broker.set_logger(self._command_logger)
+        self._motion_lock = lock
 
     def __del__(self):
         if isinstance(getattr(self, '_hardware', None),
@@ -59,7 +74,8 @@ def create(self, name, text):
                 text=text,
                 hardware=self._hardware,
                 loop=self._loop,
-                broker=self._broker)
+                broker=self._broker,
+                motion_lock=self._motion_lock)
         finally:
             self._session_lock = False
 
@@ -83,12 +99,12 @@ class Session(object):
     TOPIC = 'session'
 
     @classmethod
-    def build_and_prep(cls, name, text, hardware, loop, broker):
-        sess = cls(name, text, hardware, loop, broker)
+    def build_and_prep(cls, name, text, hardware, loop, broker, motion_lock):
+        sess = cls(name, text, hardware, loop, broker, motion_lock)
         sess.prepare()
         return sess
 
-    def __init__(self, name, text, hardware, loop, broker):
+    def __init__(self, name, text, hardware, loop, broker, motion_lock):
         self._broker = broker
         self._default_logger = self._broker.logger
         self._sim_logger = self._broker.logger.getChild('sim')
@@ -116,6 +132,7 @@ def __init__(self, name, text, hardware, loop, broker):
         self.metadata = {}
 
         self.startTime = None
+        self._motion_lock = motion_lock
 
     def prepare(self):
         self._hardware.discover_modules()
@@ -159,6 +176,7 @@ def clear_logs(self):
         self.command_log.clear()
         self.errors.clear()
 
+    @_motion_lock
     def _simulate(self):
         self._reset()
 
@@ -299,7 +317,8 @@ def resume(self):
         self.set_state('running')
         return self
 
-    def _run(self):  # noqa(C901)
+    @_motion_lock  # noqa(C901)
+    def _run(self):
         def on_command(message):
             if message['$'] == 'before':
                 self.log_append()

api/src/opentrons/drivers/smoothie_drivers/driver_3_0.py

@@ -2,7 +2,7 @@
 from os import environ
 import logging
 from time import sleep
-from threading import Event
+from threading import Event, RLock
 from typing import Any, Dict, Optional
 
 from serial.serialutil import SerialException
@@ -109,7 +109,16 @@
 
 
 class SmoothieError(Exception):
-    pass
+    def __init__(self, ret_code: str = None, command: str = None) -> None:
+        self.ret_code = ret_code
+        self.command = command
+        super().__init__()
+
+    def __repr__(self):
+        return f'<SmoothieError: {self.ret_code} from {self.command}>'
+
+    def __str__(self):
+        return f'SmoothieError: {self.command} returned {self.ret_code}'
 
 
 class ParseError(Exception):
@@ -261,7 +270,7 @@ def _parse_homing_status_values(raw_homing_status_values):
 
 
 class SmoothieDriver_3_0_0:
-    def __init__(self, config):
+    def __init__(self, config, handle_locks=True):
         self.run_flag = Event()
         self.run_flag.set()
         self.dist_from_eeprom = EEPROM_DEFAULT.copy()
@@ -329,6 +338,18 @@ def __init__(self, config):
             'C': False
         })
 
+        if handle_locks:
+            self._serial_lock = RLock()
+        else:
+            class DummyLock:
+                def __enter__(self):
+                    pass
+
+                def __exit__(self, *args, **kwargs):
+                    pass
+
+            self._serial_lock = DummyLock()
+
     @property
     def homed_position(self):
         return self._homed_position.copy()
@@ -894,6 +915,25 @@ def _send_command(self, command, timeout=DEFAULT_SMOOTHIE_TIMEOUT):
         """
         if self.simulating:
             return
+        try:
+            with self._serial_lock:
+                return self._send_command_unsynchronized(command, timeout)
+        except SmoothieError as se:
+            # XXX: This is a reentrancy error because another command could
+            # swoop in here. We're already resetting though and errors (should
+            # be) rare so it's probably fine, but the actual solution to this
+            # is locking at a higher level like in APIv2.
+            self._reset_from_error()
+            error_axis = se.ret_code.strip()[-1]
+            if GCODES['HOME'] not in command and error_axis in 'XYZABC':
+                log.warning(
+                    f"alarm/error in {se.ret_code}, homing {error_axis}")
+                self.home(error_axis)
+                raise SmoothieError(se.ret_code, command)
+
+    def _send_command_unsynchronized(self,
+                                     command,
+                                     timeout=DEFAULT_SMOOTHIE_TIMEOUT):
         cmd_ret = self._write_with_retries(
             command + SMOOTHIE_COMMAND_TERMINATOR,
             5.0, DEFAULT_COMMAND_RETRIES)
@@ -911,11 +951,6 @@ def _handle_return(self, ret_code: str, was_home: bool):
         # Smoothieware returns error state if a switch was hit while moving
         if (ERROR_KEYWORD in ret_code.lower()) or \
                 (ALARM_KEYWORD in ret_code.lower()):
-            self._reset_from_error()
-            error_axis = ret_code.strip()[-1]
-            if not was_home and error_axis in 'XYZABC':
-                log.warning(f"alarm/error in {ret_code}, homing {error_axis}")
-                self.home(error_axis)
             raise SmoothieError(ret_code)
 
     def _remove_unwanted_characters(self, command, response):

api/src/opentrons/hardware_control/controller.py

@@ -81,8 +81,9 @@ def __init__(self, config, loop, force=False):
                 raise
 
         self.config = config or opentrons.config.robot_configs.load()
+        # We handle our own locks in the hardware controller thank you
         self._smoothie_driver = driver_3_0.SmoothieDriver_3_0_0(
-            config=self.config)
+            config=self.config, handle_locks=False)
         self._cached_fw_version: Optional[str] = None
 
     def move(self, target_position: Dict[str, float],

api/src/opentrons/legacy_api/robot/robot.py

@@ -2,6 +2,7 @@
 import os
 import logging
 from functools import lru_cache
+from threading import Lock
 from typing import Dict, Any
 
 from numpy import add, subtract
@@ -141,6 +142,7 @@ def __init__(self, config=None, broker=None):
 
         self._commands = []
         self._unsubscribe_commands = None
+        self._smoothie_lock = Lock()
         self.reset()
 
     def __del__(self):