From d3c6b38823ef0f6ff59be47da069e4b1c5ac556e Mon Sep 17 00:00:00 2001 From: Samuel Sadok Date: Thu, 3 Dec 2020 21:20:53 +0100 Subject: [PATCH] fix coroutine lifetime --- tools/odrive/pyfibre/fibre/libfibre.py | 36 ++++++++++++++------------ tools/odrive/pyfibre/fibre/shell.py | 2 +- 2 files changed, 21 insertions(+), 17 deletions(-) diff --git a/tools/odrive/pyfibre/fibre/libfibre.py b/tools/odrive/pyfibre/fibre/libfibre.py index 348bdc35..6162789f 100644 --- a/tools/odrive/pyfibre/fibre/libfibre.py +++ b/tools/odrive/pyfibre/fibre/libfibre.py @@ -470,17 +470,14 @@ class RemoteFunction(object): rx_length = sum(arg[2].get_length() for arg in self._outputs) - # Create task which will start the TX operation soon. This allows us to - # start an RX operation before the TX operation is actually started. - tx_future = asyncio.create_task(tx_stream.write_all(tx_buf)) - - try: - try: - rx_buf = await rx_stream.read_all(rx_length) - finally: - await tx_future - finally: - await call_future + # The order here is important: The read_all coroutine must get a chance + # to run before the write_all coroutine. Otherwise it's possible that + # RX data is lost. + rx_buf, _, _ = await asyncio.gather( + rx_stream.read_all(rx_length), + tx_stream.write_all(tx_buf), + call_future, + ) outputs = [] for arg in self._outputs: @@ -510,8 +507,7 @@ class RemoteFunction(object): if (len(self._inputs) != len(args)): raise TypeError("expected {} arguments but have {}".format(len(self._inputs), len(args))) - coro = self.async_call(instance, args, cancellation_token) - return asyncio.ensure_future(coro, loop=instance._libfibre.loop) + return self.async_call(instance, args, cancellation_token) def __get__(self, instance, owner): return MethodType(self, instance) if instance else self @@ -548,13 +544,20 @@ class RemoteAttribute(object): return self if self._magic_getter: + if threading.current_thread() == libfibre_thread: + # read() behaves asynchronously when run on the fibre thread + # which means it returns an awaitable which _must_ be awaited + # (otherwise it's a bug). However hasattr(...) internally calls + # __get__ and does not await the result. Thus the safest thing + # is to just disallow __get__ from run as an async method. + raise Exception("Cannot use magic getter on Fibre thread. Use _[prop_name]_propery.read() instead.") return self._get_obj(instance).read() else: return self._get_obj(instance) def __set__(self, instance, val): if self._magic_setter: - self._get_obj(instance).exchange(val) + return self._get_obj(instance).exchange(val) else: raise Exception("this attribute cannot be written to") @@ -731,7 +734,7 @@ class LibFibre(): def _on_found_object(self, ctx, obj): py_obj = self._objects[obj] # notify the subscriber - asyncio.ensure_future(self.discovery_processes[ctx]['callback'](py_obj)) + self.discovery_processes[ctx]['callback'](py_obj) def _on_discovery_stopped(self, ctx, result): print("discovery stopped") @@ -788,6 +791,7 @@ def run_event_loop(): libfibre = LibFibre() libfibre.loop.run_until_complete(terminate_libfibre) + libfibre.loop.set_debug(True) libfibre_close(libfibre.ctx) @@ -870,7 +874,7 @@ def start_discovery(path, obj_filter, libfibre.loop.call_soon_threadsafe(lambda: libfibre.start_discovery( path, - on_object_discovered_filter, + lambda x: asyncio.ensure_future(on_object_discovered_filter(x), loop=libfibre.loop), search_cancellation_token)) diff --git a/tools/odrive/pyfibre/fibre/shell.py b/tools/odrive/pyfibre/fibre/shell.py index e8ba71a7..ebcf423a 100644 --- a/tools/odrive/pyfibre/fibre/shell.py +++ b/tools/odrive/pyfibre/fibre/shell.py @@ -13,7 +13,7 @@ async def discovered_device(device, message and making the device available to the interactive console """ - serial_number = '{:012X}'.format(await device.serial_number) if hasattr(device, 'serial_number') else "[unknown serial number]" + serial_number = '{:012X}'.format(await device._serial_number_property.read()) if hasattr(device, '_serial_number_property') else "[unknown serial number]" if serial_number in discovered_devices: verb = "Reconnected" index = discovered_devices.index(serial_number)