Implement async guarding to prevent accidental blocking IO

Author: sveinse

canopen/async_guard.py

@@ -0,0 +1,31 @@
+""" Utils for async """
+
+import functools
+from typing import Optional, Callable
+
+TSentinel = Callable[[], bool]
+
+# NOTE: Global, but needed to be able to use ensure_not_async() in
+#       decorator context.
+_ASYNC_SENTINEL: Optional[TSentinel] = None
+
+
+def set_async_sentinel(fn: TSentinel):
+    """ Register a function to validate if async is running """
+    global _ASYNC_SENTINEL
+    _ASYNC_SENTINEL = fn
+
+
+def ensure_not_async(fn):
+    """ Decorator that will ensure that the function is not called if async
+        is running.
+    """
+
+    @functools.wraps(fn)
+    def async_guard(*args, **kwargs):
+        global _ASYNC_SENTINEL
+        if _ASYNC_SENTINEL:
+            if _ASYNC_SENTINEL():
+                raise RuntimeError("Calling a blocking function while running async")
+        return fn(*args, **kwargs)
+    return async_guard

canopen/emcy.py

@@ -5,6 +5,7 @@
 import asyncio
 import time
 from typing import Callable, List, Optional, TYPE_CHECKING
+from .async_guard import ensure_not_async
 
 if TYPE_CHECKING:
     from .network import Network
@@ -26,6 +27,7 @@ def __init__(self):
         self.emcy_received = threading.Condition()
         self.aemcy_received = asyncio.Condition()
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def on_emcy(self, can_id, data, timestamp):
         # NOTE: Callback. Called from another thread unless async
         code, register, data = EMCY_STRUCT.unpack(data)
@@ -76,6 +78,8 @@ def reset(self):
         self.log = []
         self.active = []
 
+    # FIXME: Make async implementation
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def wait(
         self, emcy_code: Optional[int] = None, timeout: float = 10
     ) -> "EmcyError":

canopen/lss.py

@@ -8,6 +8,7 @@
     import queue
 except ImportError:
     import Queue as queue
+from .async_guard import ensure_not_async
 
 if TYPE_CHECKING:
     from .network import Network
@@ -248,6 +249,8 @@ def send_identify_non_configured_remote_slave(self):
         message[0] = CS_IDENTIFY_NON_CONFIGURED_REMOTE_SLAVE
         self.__send_command(message)
 
+    # FIXME: Make async implementation
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def fast_scan(self):
         """This command sends a series of fastscan message
         to find unconfigured slave with lowest number of LSS idenities
@@ -279,7 +282,7 @@ def fast_scan(self):
                 if not self.__send_fast_scan_message(lss_id[lss_sub], lss_bit_check, lss_sub, lss_next):
                     return False, None
 
-                time.sleep(0.01)  # NOTE: Blocking
+                time.sleep(0.01)  # NOTE: Blocking call
 
                 # Now the next 32 bits will be scanned
                 lss_sub += 1
@@ -303,6 +306,8 @@ def __send_fast_scan_message(self, id_number, bit_checker, lss_sub, lss_next):
 
         return False
 
+    # FIXME: Make async implementation
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def __send_lss_address(self, req_cs, number):
         message = bytearray(8)
 
@@ -366,6 +371,8 @@ def __send_configure(self, req_cs, value1=0, value2=0):
             error_msg = "LSS Error: %d" % error_code
             raise LssError(error_msg)
 
+    # FIXME: Make async implementation
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def __send_command(self, message):
         """Send a LSS operation code to the network
 
@@ -385,7 +392,7 @@ def __send_command(self, message):
         response = None
         if not self.responses.empty():
             logger.info("There were unexpected messages in the queue")
-            self.responses = queue.Queue()  # FIXME: Recreating the queue. Async too?
+            self.responses = queue.Queue()  # FIXME: Recreating the queue
 
         self.network.send_message(self.LSS_TX_COBID, message)
 
@@ -402,6 +409,7 @@ def __send_command(self, message):
 
         return response
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def on_message_received(self, can_id, data, timestamp):
         # NOTE: Callback. Called from another thread
         self.responses.put(bytes(data))  # NOTE: Blocking call

canopen/network.py

@@ -29,6 +29,7 @@
 from .lss import LssMaster
 from .objectdictionary.eds import import_from_node
 from .objectdictionary import ObjectDictionary
+from .async_guard import set_async_sentinel
 
 logger = logging.getLogger(__name__)
 
@@ -132,6 +133,10 @@ def connect(self, *args, **kwargs) -> "Network":
             kwargs_notifier["loop"] = kwargs["loop"]
             self.loop = kwargs["loop"]
             del kwargs["loop"]
+            # Register this function as the means to check if canopen is run in
+            # async mode. This enables the @ensure_not_async() decorator to
+            # work. See async_guard.py
+            set_async_sentinel(self.is_async)
         self.bus = can.Bus(*args, **kwargs)
         logger.info("Connected to '%s'", self.bus.channel_info)
         self.notifier = can.Notifier(self.bus, self.listeners, 1, **kwargs_notifier)
@@ -359,8 +364,7 @@ def update(self, data: bytes) -> None:
         :param data:
             New data to transmit
         """
-        # NOTE: Called from callback, which is another thread on non-async use.
-        #       Make sure this is thread-safe.
+        # NOTE: Callback. Called from another thread unless async
         new_data = bytearray(data)
         old_data = self.msg.data
         self.msg.data = new_data
@@ -436,6 +440,7 @@ def search(self, limit: int = 127) -> None:
         """Search for nodes by sending SDO requests to all node IDs."""
         if self.network is None:
             raise RuntimeError("A Network is required to do active scanning")
+        # SDO upload request, parameter 0x1000:0x00
         sdo_req = b"\x40\x00\x10\x00\x00\x00\x00\x00"
         for node_id in range(1, limit + 1):
             self.network.send_message(0x600 + node_id, sdo_req)

canopen/nmt.py

@@ -7,6 +7,7 @@
 from typing import Callable, Optional, TYPE_CHECKING
 
 from .network import CanError
+from .async_guard import ensure_not_async
 
 if TYPE_CHECKING:
     from .network import Network
@@ -96,15 +97,14 @@ def state(self) -> str:
         - 'RESET'
         - 'RESET COMMUNICATION'
         """
-        logger.warning("Accessing NmtBase.state attribute is deprecated")
         if self._state in NMT_STATES:
             return NMT_STATES[self._state]
         else:
             return self._state
 
     @state.setter
     def state(self, new_state: str):
-        logger.warning("Accessing NmtBase.state setter is deprecated")
+        logger.warning("Accessing NmtBase.state setter is deprecated, use set_state()")
         self.set_state(new_state)
 
     def set_state(self, new_state: str):
@@ -129,6 +129,7 @@ def __init__(self, node_id: int):
         self.astate_update = asyncio.Condition()
         self._callbacks = []
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def on_heartbeat(self, can_id, data, timestamp):
         # NOTE: Callback. Called from another thread unless async
         with self.state_update:  # NOTE: Blocking call
@@ -177,6 +178,7 @@ def send_command(self, code: int):
             "Sending NMT command 0x%X to node %d", code, self.id)
         self.network.send_message(0, [code, self.id])
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def wait_for_heartbeat(self, timeout: float = 10):
         """Wait until a heartbeat message is received."""
         with self.state_update:  # NOTE: Blocking call
@@ -186,6 +188,17 @@ def wait_for_heartbeat(self, timeout: float = 10):
             raise NmtError("No boot-up or heartbeat received")
         return self.state
 
+    async def await_for_heartbeat(self, timeout: float = 10):
+        """Wait until a heartbeat message is received."""
+        async with self.astate_update:
+            self._state_received = None
+            try:
+                await asyncio.wait_for(self.astate_update.wait(), timeout=timeout)
+            except asyncio.TimeoutError:
+                raise NmtError("No boot-up or heartbeat received")
+        return self.state
+
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def wait_for_bootup(self, timeout: float = 10) -> None:
         """Wait until a boot-up message is received."""
         end_time = time.time() + timeout
@@ -199,6 +212,20 @@ def wait_for_bootup(self, timeout: float = 10) -> None:
             if self._state_received == 0:
                 break
 
+    async def await_for_bootup(self, timeout: float = 10) -> None:
+        """Wait until a boot-up message is received."""
+        async def wait_for_bootup():
+            while True:
+                async with self.astate_update:
+                    self._state_received = None
+                    await self.astate_update.wait()
+                    if self._state_received == 0:
+                        return
+        try:
+            await asyncio.wait_for(wait_for_bootup(), timeout=timeout)
+        except asyncio.TimeoutError:
+            raise NmtError("Timeout waiting for boot-up message")
+
     def add_hearbeat_callback(self, callback: Callable[[int], None]):
         """Add function to be called on heartbeat reception.
 
@@ -255,7 +282,7 @@ def send_command(self, code: int) -> None:
         # The heartbeat service should start on the transition
         # between INITIALIZING and PRE-OPERATIONAL state
         if old_state == 0 and self._state == 127:
-            heartbeat_time_ms = self._local_node.sdo[0x1017].get_raw()
+            heartbeat_time_ms = self._local_node.sdo[0x1017].get_raw()  # FIXME: Blocking?
             self.start_heartbeat(heartbeat_time_ms)
         else:
             self.update_heartbeat()

canopen/node/remote.py

@@ -151,9 +151,9 @@ def __load_configuration_helper(self, index, subindex, name, value):
                     subindex=subindex,
                     name=name,
                     value=value)))
-                self.sdo[index][subindex].set_raw(value)
+                self.sdo[index][subindex].set_raw(value)  # FIXME: Blocking?
             else:
-                self.sdo[index].set_raw(value)
+                self.sdo[index].set_raw(value)  # FIXME: Blocking?
                 logger.info(str('SDO [{index:#06x}]: {name}: {value:#06x}'.format(
                     index=index,
                     name=name,

canopen/objectdictionary/eds.py

@@ -8,6 +8,7 @@
     from ConfigParser import RawConfigParser, NoOptionError, NoSectionError
 from canopen import objectdictionary
 from canopen.sdo import SdoClient
+from canopen.async_guard import ensure_not_async
 
 logger = logging.getLogger(__name__)
 
@@ -167,12 +168,13 @@ def import_eds(source, node_id):
     return od
 
 
+# FIXME: Make async variant
+@ensure_not_async  # NOTE: Safeguard for accidental async use
 def import_from_node(node_id, network):
     """ Download the configuration from the remote node
     :param int node_id: Identifier of the node
     :param network: network object
     """
-    # FIXME: Implement async variant
     # Create temporary SDO client
     sdo_client = SdoClient(0x600 + node_id, 0x580 + node_id, objectdictionary.ObjectDictionary())
     sdo_client.network = network

canopen/pdo/base.py

@@ -16,6 +16,7 @@
 from ..sdo import SdoAbortedError
 from .. import objectdictionary
 from .. import variable
+from ..async_guard import ensure_not_async
 
 PDO_NOT_VALID = 1 << 31
 RTR_NOT_ALLOWED = 1 << 30
@@ -205,7 +206,7 @@ def __init__(self, pdo_node: PdoBase, com_record, map_array):
         #: Set explicitly or using the :meth:`start()` method.
         self.period: Optional[float] = None
         self.callbacks = []
-        self.receive_condition = threading.Condition()  # FIXME Async
+        self.receive_condition = threading.Condition()
         self.areceive_condition = asyncio.Condition()
         self.is_received: bool = False
         self._task = None
@@ -307,11 +308,12 @@ def is_periodic(self) -> bool:
         # Unknown transmission type, assume non-periodic
         return False
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def on_message(self, can_id, data, timestamp):
         # NOTE: Callback. Called from another thread unless async
         is_transmitting = self._task is not None
         if can_id == self.cob_id and not is_transmitting:
-            with self.receive_condition:  # FIXME: Blocking
+            with self.receive_condition:  # NOTE: Blocking call
                 self.is_received = True
                 self.data = data
                 if self.timestamp is not None:
@@ -394,6 +396,7 @@ def read_generator(self):
 
         self.subscribe()
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def read(self, from_od=False) -> None:
         """Read PDO configuration for this map using SDO or from OD."""
         gen = self.read_generator()
@@ -404,7 +407,7 @@ def read(self, from_od=False) -> None:
                 value = var.od.default
             else:
                 # Get value from SDO
-                value = var.get_raw()
+                value = var.get_raw()  # FIXME: Blocking?
             try:
                 # Deliver value into read_generator and wait for next object
                 var = gen.send(value)
@@ -454,8 +457,7 @@ def save_generator(self):
                 # mappings for an invalid object 0x0000:00 to overwrite any
                 # excess entries with all-zeros.
 
-                # FIXME: This is a blocking call which might be called from async
-                self._fill_map(self.map_array[0].get_raw())
+                self._fill_map(self.map_array[0].get_raw())  # FIXME: Blocking?
             subindex = 1
             for var in self.map:
                 logger.info("Writing %s (0x%X:%d, %d bits) to PDO map",
@@ -485,10 +487,11 @@ def save_generator(self):
             yield self.com_record[1], self.cob_id | (RTR_NOT_ALLOWED if not self.rtr_allowed else 0x0)
             self.subscribe()
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def save(self) -> None:
         """Read PDO configuration for this map using SDO."""
         for sdo, value in self.save_generator():
-            sdo.set_raw(value)
+            sdo.set_raw(value)  # FIXME: Blocking?
 
     async def asave(self) -> None:
         """Read PDO configuration for this map using SDO, async variant."""
@@ -596,15 +599,16 @@ def remote_request(self) -> None:
         if self.enabled and self.rtr_allowed:
             self.pdo_node.network.send_message(self.cob_id, None, remote=True)
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def wait_for_reception(self, timeout: float = 10) -> float:
         """Wait for the next transmit PDO.
 
         :param float timeout: Max time to wait in seconds.
         :return: Timestamp of message received or None if timeout.
         """
-        with self.receive_condition:  # FIXME: Blocking
+        with self.receive_condition:  # NOTE: Blocking call
             self.is_received = False
-            self.receive_condition.wait(timeout)  # FIXME: Blocking
+            self.receive_condition.wait(timeout)  # NOTE: Blocking call
         return self.timestamp if self.is_received else None
 
     async def await_for_reception(self, timeout: float = 10) -> float: