Refurbish async guard system

* Thread dependent sentinel guard

Author: sveinse

canopen/async_guard.py

@@ -1,31 +1,24 @@
 """ Utils for async """
-
 import functools
-from typing import Optional, Callable
-
-TSentinel = Callable[[], bool]
+import threading
 
 # NOTE: Global, but needed to be able to use ensure_not_async() in
 #       decorator context.
-_ASYNC_SENTINEL: Optional[TSentinel] = None
+_ASYNC_SENTINELS: dict[int, bool] = {}
 
 
-def set_async_sentinel(fn: TSentinel):
+def set_async_sentinel(enable: bool):
     """ Register a function to validate if async is running """
-    global _ASYNC_SENTINEL
-    _ASYNC_SENTINEL = fn
+    _ASYNC_SENTINELS[threading.get_ident()] = enable
 
 
 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")
+    def async_guard_wrap(*args, **kwargs):
+        if _ASYNC_SENTINELS.get(threading.get_ident(), False):
+            raise RuntimeError(f"Calling a blocking function in async. {fn.__qualname__}() in {fn.__code__.co_filename}:{fn.__code__.co_firstlineno}, while running async")
         return fn(*args, **kwargs)
-    return async_guard
+    return async_guard_wrap

canopen/emcy.py

@@ -1,3 +1,4 @@
+from __future__ import annotations
 import asyncio
 import logging
 import struct
@@ -83,7 +84,7 @@ def reset(self):
     @ensure_not_async  # NOTE: Safeguard for accidental async use
     def wait(
         self, emcy_code: Optional[int] = None, timeout: float = 10
-    ) -> "EmcyError":
+    ) -> EmcyError:
         """Wait for a new EMCY to arrive.
 
         :param emcy_code: EMCY code to wait for
@@ -111,6 +112,14 @@ def wait(
                     # This is the one we're interested in
                     return emcy
 
+    def async_wait(
+        self, emcy_code: Optional[int] = None, timeout: float = 10
+    ) -> EmcyError:
+        # FIXME: Implement this function
+        raise NotImplementedError(
+            "async_wait is not implemented."
+        )
+
 
 class EmcyProducer:
 

canopen/network.py

@@ -60,7 +60,7 @@ def __init__(self, bus: Optional[can.BusABC] = None, notifier: Optional[can.Noti
         # 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)
+        set_async_sentinel(self.is_async())
 
         if self.is_async():
             self.subscribe(self.lss.LSS_RX_COBID, self.lss.aon_message_received)
@@ -142,6 +142,9 @@ def disconnect(self) -> None:
         self.bus = None
         self.check()
 
+        # Remove the async sentinel
+        set_async_sentinel(False)
+
     def __enter__(self):
         return self
 

canopen/objectdictionary/eds.py

@@ -4,7 +4,6 @@
 from configparser import NoOptionError, NoSectionError, RawConfigParser
 
 from canopen import objectdictionary
-from canopen.async_guard import ensure_not_async
 from canopen.objectdictionary import ObjectDictionary, datatypes
 from canopen.sdo import SdoClient
 
@@ -175,7 +174,6 @@ def import_eds(source, node_id):
 
 
 # FIXME: Make async variant "aimport_from_node"
-@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

canopen/pdo/base.py

@@ -58,6 +58,7 @@ def __getitem__(self, key) -> PdoBase:
     def __len__(self):
         return len(self.map)
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def read(self, from_od=False):
         """Read PDO configuration from node using SDO."""
         for pdo_map in self.map.values():
@@ -68,6 +69,7 @@ async def aread(self, from_od=False):
         for pdo_map in self.map.values():
             await pdo_map.aread(from_od=from_od)
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def save(self):
         """Save PDO configuration to node using SDO."""
         for pdo_map in self.map.values():

canopen/sdo/base.py

@@ -8,6 +8,7 @@
 from canopen import objectdictionary
 from canopen import variable
 from canopen.utils import pretty_index
+from canopen.async_guard import ensure_not_async
 
 
 class CrcXmodem:
@@ -183,12 +184,14 @@ def __init__(self, sdo_node: SdoBase, od: objectdictionary.ODVariable):
     def __await__(self):
         return self.aget_raw().__await__()
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def get_data(self) -> bytes:
         return self.sdo_node.upload(self.od.index, self.od.subindex)
 
     async def aget_data(self) -> bytes:
         return await self.sdo_node.aupload(self.od.index, self.od.subindex)
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def set_data(self, data: bytes):
         force_segment = self.od.data_type == objectdictionary.DOMAIN
         self.sdo_node.download(self.od.index, self.od.subindex, data, force_segment)
@@ -205,6 +208,7 @@ def writable(self) -> bool:
     def readable(self) -> bool:
         return self.od.readable
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def open(self, mode="rb", encoding="ascii", buffering=1024, size=None,
              block_transfer=False, request_crc_support=True):
         """Open the data stream as a file like object.

canopen/sdo/client.py

@@ -12,6 +12,7 @@
 from canopen.sdo.constants import *
 from canopen.sdo.exceptions import *
 from canopen.utils import pretty_index
+from canopen.async_guard import ensure_not_async
 
 
 logger = logging.getLogger(__name__)
@@ -49,9 +50,11 @@ def __init__(self, rx_cobid, tx_cobid, od):
     def on_response(self, can_id, data, timestamp):
         self.responses.put_nowait(bytes(data))
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def send_request(self, request):
         retries_left = self.MAX_RETRIES
         if self.PAUSE_BEFORE_SEND:
+            # NOTE: Blocking
             time.sleep(self.PAUSE_BEFORE_SEND)
         while True:
             try:
@@ -63,6 +66,7 @@ def send_request(self, request):
                     raise
                 logger.info(str(e))
                 if self.RETRY_DELAY:
+                    # NOTE: Blocking
                     time.sleep(self.RETRY_DELAY)
             else:
                 break
@@ -108,6 +112,7 @@ def abort(self, abort_code=0x08000000):
         self.send_request(request)
         logger.error("Transfer aborted by client with code 0x%08X", abort_code)
 
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def upload(self, index: int, subindex: int) -> bytes:
         """May be called to make a read operation without an Object Dictionary.
 
@@ -126,7 +131,9 @@ def upload(self, index: int, subindex: int) -> bytes:
         with self.open(index, subindex, buffering=0) as fp:
             response_size = fp.size
             data = fp.read()
+        return self.truncate_data(index, subindex, data, response_size)
 
+    def truncate_data(self, index: int, subindex: int, data: bytes, size: int) -> bytes:
         # If size is available through variable in OD, then use the smaller of the two sizes.
         # Some devices send U32/I32 even if variable is smaller in OD
         var = self.od.get_variable(index, subindex)
@@ -137,7 +144,7 @@ def upload(self, index: int, subindex: int) -> bytes:
             if var.data_type not in objectdictionary.DATA_TYPES:
                 # Get the size in bytes for this variable
                 var_size = len(var) // 8
-                if response_size is None or var_size < response_size:
+                if size is None or var_size < size:
                     # Truncate the data to specified size
                     data = data[0:var_size]
         return data
@@ -152,8 +159,16 @@ async def aupload(self, index: int, subindex: int) -> bytes:
             #    upload -> open -> ReadableStream -> request_reponse -> send_request -> network.send_message
             #    recv -> on_reponse -> queue.put
             #                                        request_reponse -> read_response -> queue.get
-            return await asyncio.to_thread(self.upload, index, subindex)
+            def _upload():
+                with self._open(index, subindex, buffering=0) as fp:
+                    response_size = fp.size
+                    data = fp.read()
+                return data, response_size
 
+            data, response_size = await asyncio.to_thread(_upload)
+            return self.truncate_data(index, subindex, data, response_size)
+
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def download(
         self,
         index: int,
@@ -193,10 +208,22 @@ async def adownload(
         """
         async with self.lock:  # Ensure only one active SDO request per channel
             # Deferring to thread because there are sleeps in the call chain
-            return await asyncio.to_thread(self.download, index, subindex, data, force_segment)
 
+            def _download():
+                with self._open(index, subindex, "wb", buffering=7, size=len(data),
+                                force_segment=force_segment) as fp:
+                    fp.write(data)
+
+            return await asyncio.to_thread(_download)
+
+    @ensure_not_async  # NOTE: Safeguard for accidental async use
     def open(self, index, subindex=0, mode="rb", encoding="ascii",
              buffering=1024, size=None, block_transfer=False, force_segment=False, request_crc_support=True):
+        return self._open(index, subindex, mode, encoding, buffering,
+                          size, block_transfer, force_segment, request_crc_support)
+
+    def _open(self, index, subindex=0, mode="rb", encoding="ascii",
+             buffering=1024, size=None, block_transfer=False, force_segment=False, request_crc_support=True):
         """Open the data stream as a file like object.
 
         :param int index: