diff --git a/tools/odrive/pyfibre/fibre/libfibre.py b/tools/odrive/pyfibre/fibre/libfibre.py index b623fa56..5dece883 100644 --- a/tools/odrive/pyfibre/fibre/libfibre.py +++ b/tools/odrive/pyfibre/fibre/libfibre.py @@ -591,27 +591,39 @@ class RemoteAttribute(object): self._magic_setter = magic_setter def _get_obj(self, instance): - py_intf = self._libfibre._load_py_intf(self._intf_name, self._intf_handle) - obj_handle = c_void_p(0) status = libfibre_get_attribute(instance._obj_handle, self._attr_handle, byref(obj_handle)) if status != kFibreOk: raise _get_exception(status) - return self._libfibre._objects[obj_handle.value] + obj = self._libfibre._load_py_obj(obj_handle.value, self._intf_handle) + if obj in instance._children: + self._libfibre._release_py_obj(obj_handle.value) + else: + # the object will be released when the parent is released + instance._children.add(obj) + + return obj def __get__(self, instance, owner): if not instance: 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") @@ -624,6 +636,8 @@ class RemoteObject(object): def __init__(self, libfibre, obj_handle): self.__class__._refcount += 1 + self._refcount = 0 + self._children = set() self._libfibre = libfibre self._obj_handle = obj_handle @@ -682,10 +696,15 @@ class RemoteObject(object): def _destroy(self): libfibre = self._libfibre on_lost = self._on_lost + children = self._children self._libfibre = None self._obj_handle = None self._on_lost = None + self._children = set() + + for child in children: + libfibre._release_py_obj(child._obj_handle) self.__class__._refcount -= 1 if self.__class__._refcount == 0: @@ -729,6 +748,7 @@ class LibFibre(): event_loop.cancel_timer = self.c_cancel_timer self.ctx = c_void_p(libfibre_open(event_loop)) + assert(self.ctx) def _post(self, callback, ctx): self.loop.call_soon_threadsafe(callback, ctx) @@ -786,26 +806,19 @@ class LibFibre(): self._objects[obj_handle] = py_obj else: py_obj = self._objects[obj_handle] + + # Note: this refcount does not count the python references to the object + # but rather mirrors the libfibre-internal refcount of the object. This + # is so that we can destroy the Python object when libfibre releases it. py_obj._refcount += 1 return py_obj def _release_py_obj(self, obj_handle): py_obj = self._objects[obj_handle] py_obj._refcount -= 1 - if py_obj.refcount <= 0: + if py_obj._refcount <= 0: self._objects.pop(obj_handle) - - #def _construct_object(self, ctx, obj, intf, name, name_length): - # #increment_lib_refcount() - # name = None if name is None else string_at(name, name_length).decode('utf-8') - # py_intf = self._load_py_intf(name, intf) - # assert(not obj in self._objects) - # self._objects[obj] = py_intf(self, obj) - - def _free_py_obj(self, ctx, obj): - py_obj = self._objects.pop(obj) - py_obj._destroy() - #decrement_lib_refcount() + py_obj._destroy() def _on_found_object(self, ctx, obj, intf): py_obj = self._load_py_obj(obj, intf) @@ -816,7 +829,7 @@ class LibFibre(): old_future.set_result(None) def _on_lost_object(self, ctx, obj): - self._free_py_obj(obj) + self._release_py_obj(obj) def _on_discovery_stopped(self, ctx, result): print("discovery stopped") @@ -853,31 +866,11 @@ class LibFibre(): return kFibreBusy -# def start_discovery(self, path, on_obj_discovered, cancellation_token): -# buf = path.encode('ascii') -# -# discovery = { -# 'domain_handle': c_void_p(0), -# 'handle': c_void_p(0), -# 'callback': on_obj_discovered -# } -# discovery_id = insert_with_new_id(self.discovery_processes, discovery) -# -# def stop_discovery(): -# libfibre_stop_discovery(discovery['handle']) -# print("closing domain") -# -# -# print("ok ") -# cancellation_token.subscribe(lambda: stop_discovery) -# -# discovery['domain_handle'] = libfibre_open_domain(self.ctx, buf, len(buf)) -# assert(discovery['domain_handle']) -# libfibre_start_discovery(discovery['domain_handle'], byref(discovery['handle']), self.c_on_found_object, self.c_on_lost_object, self.c_on_discovery_stopped, discovery_id) -# print("disc handle ", hex(discovery['handle'].value)) -# - class Discovery(): + """ + All public members of this class are thread-safe. + """ + def __init__(self, domain): self._domain = domain self._id = 0 @@ -893,8 +886,19 @@ class Discovery(): def _stop(self): self._domain._libfibre.discovery_processes.pop(self._id) libfibre_stop_discovery(self._discovery_handle) + self._future.set_exception(asyncio.CancelledError()) + + def stop(self): + if threading.current_thread() == libfibre_thread: + self._stop() + else: + run_coroutine_threadsafe(self._domain._libfibre.loop, self._stop) class _Domain(): + """ + All public members of this class are thread-safe. + """ + def __init__(self, libfibre, handle): self._libfibre = libfibre self._domain_handle = handle @@ -917,8 +921,25 @@ class _Domain(): return obj def discover_one(self): + """ + Blocks until exactly one object is discovered. + """ return run_coroutine_threadsafe(self._libfibre.loop, self._discover_one) + def run_discovery(self, callback): + """ + Invokes `callback` for every object that is discovered. The callback is + invoked on the libfibre thread and can be an asynchronous function. + Returns a `Discovery` object on which `stop()` can be called to + terminate the discovery. + """ + discovery = run_coroutine_threadsafe(self._libfibre.loop, self._start_discovery) + async def loop(): + while True: + obj = await discovery._next() + await callback(obj) + self._libfibre.loop.call_soon_threadsafe(lambda: asyncio.ensure_future(loop())) + return discovery class Domain(): @@ -1006,38 +1027,6 @@ def decrement_lib_refcount(): libfibre_thread.join() libfibre_thread = None - -#def start_discovery(path, obj_filter, -# on_object_discovered, -# search_cancellation_token, -# channel_termination_token): -# """ -# Starts scanning for Fibre objects that match the specified path spec and calls -# the callback for each Fibre object that is found. -# -# This function is non-blocking and thread-safe. -# """ -# -# async def on_object_discovered_filter(obj): -# increment_lib_refcount() -# if not channel_termination_token is None: -# channel_termination_token.subscribe(lambda: decrement_lib_refcount()) -# if await obj_filter(obj): -# result = on_object_discovered(obj) -# if not result is None: -# await result -# elif not channel_termination_token is None: -# channel_termination_token.set() -# -# increment_lib_refcount() -# search_cancellation_token.subscribe(lambda: decrement_lib_refcount()) -# -# libfibre.loop.call_soon_threadsafe(lambda: libfibre.start_discovery( -# path, -# on_object_discovered_filter, -# search_cancellation_token)) - - def get_user_name(obj): """ Can be overridden by the application to return the user-facing name of an diff --git a/tools/odrive/pyfibre/fibre/shell.py b/tools/odrive/pyfibre/fibre/shell.py index 31ba99a5..27421567 100644 --- a/tools/odrive/pyfibre/fibre/shell.py +++ b/tools/odrive/pyfibre/fibre/shell.py @@ -4,42 +4,47 @@ import platform import threading import fibre -async def did_discover_device(device, +async def discovered_device(device, interactive_variables, discovered_devices, - branding_short, branding_long, - logger, app_shutdown_token): + mount, shutdown_token, logger): """ Handles the discovery of new devices by displaying a 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]" - if serial_number in discovered_devices: + mount_result = await mount(device) + if mount_result is None: + logger.debug("ignoring device") + return + + display_name, var_name = mount_result + + if display_name in discovered_devices: verb = "Reconnected" - index = discovered_devices.index(serial_number) + index = discovered_devices.index(display_name) else: verb = "Connected" - discovered_devices.append(serial_number) + discovered_devices.append(display_name) index = len(discovered_devices) - 1 - interactive_name = branding_short + str(index) + + var_name = var_name + str(index) # Publish new device to interactive console - interactive_variables[interactive_name] = device - globals()[interactive_name] = device # Add to globals so tab complete works - logger.notify("{} to {} {} as {}".format(verb, branding_long, serial_number, interactive_name)) + interactive_variables[var_name] = device + globals()[var_name] = device # Add to globals so tab complete works + logger.notify("{} to {} as {}".format(verb, display_name, var_name)) # Subscribe to disappearance of the device - device._on_lost.add_done_callback(lambda x: did_lose_device(interactive_name, logger, app_shutdown_token)) + device._on_lost.add_done_callback(lambda x: lost_device(var_name, shutdown_token, logger)) -def did_lose_device(interactive_name, logger, app_shutdown_token): +def lost_device(interactive_name, shutdown_token, logger): """ Handles the disappearance of a device by displaying a message. """ - if not app_shutdown_token.is_set(): + if not shutdown_token[0]: logger.warn("Oh no {} disappeared".format(interactive_name)) - def get_user_name(interactive_variables, obj): queue = [(k, v) for k, v in interactive_variables.items() if isinstance(v, fibre.libfibre.RemoteObject)] @@ -57,12 +62,10 @@ def get_user_name(interactive_variables, obj): return "anonymous_remote_object_" + str(self._obj_handle) -def launch_shell(args, - object_filter, +def launch_shell(args, mount, interactive_variables, print_banner, print_help, - logger, app_shutdown_token, - branding_short="dev", branding_long="device"): + logger): """ Launches an interactive python or IPython command line interface. @@ -72,78 +75,76 @@ def launch_shell(args, """ discovered_devices = [] + shutdown_token = [False] globals().update(interactive_variables) fibre.libfibre.get_user_name = lambda obj: get_user_name(interactive_variables, obj) # Connect to device - logger.debug("Waiting for {}...".format(branding_long)) - fibre.start_discovery(args.path, object_filter, - lambda dev: did_discover_device(dev, interactive_variables, discovered_devices, branding_short, branding_long, logger, app_shutdown_token), - app_shutdown_token, - app_shutdown_token, - logger=logger) + with fibre.Domain(args.path) as domain: + on_discovery = lambda dev: discovered_device(dev, interactive_variables, discovered_devices, mount, shutdown_token, logger) + discovery = domain.run_discovery(on_discovery) - # Check if IPython is installed - if args.no_ipython: - use_ipython = False - else: - try: - import IPython - use_ipython = True - except: - print("Warning: you don't have IPython installed.") - print("If you want to have an improved interactive console with pretty colors,") - print("you should install IPython\n") + # Check if IPython is installed + if args.no_ipython: use_ipython = False + else: + try: + import IPython + use_ipython = True + except: + print("Warning: you don't have IPython installed.") + print("If you want to have an improved interactive console with pretty colors,") + print("you should install IPython\n") + use_ipython = False - interactive_variables["help"] = lambda: print_help(args, len(discovered_devices) > 0) + interactive_variables["help"] = lambda: print_help(args, len(discovered_devices) > 0) - # If IPython is installed, embed IPython shell, otherwise embed regular shell - if use_ipython: - # Override help function # pylint: disable=W0612 - help = lambda: print_help(args, len(discovered_devices) > 0) - # to fix broken "%run -i script.py" - locals()['__name__'] = globals()['__name__'] - console = IPython.terminal.embed.InteractiveShellEmbed(banner1='') + # If IPython is installed, embed IPython shell, otherwise embed regular shell + if use_ipython: + # Override help function # pylint: disable=W0612 + help = lambda: print_help(args, len(discovered_devices) > 0) + # to fix broken "%run -i script.py" + locals()['__name__'] = globals()['__name__'] + console = IPython.terminal.embed.InteractiveShellEmbed(banner1='') - # hack to make IPython look like the regular console - console.runcode = console.run_cell - interact = console + # hack to make IPython look like the regular console + console.runcode = console.run_cell + interact = console - # Catch ObjectLostError (since disconnect is not always an error) - default_exception_hook = console._showtraceback - def filtered_exception_hook(ex_class, ex, trace): - if(ex_class.__module__+'.'+ex_class.__name__ != 'fibre.libfibre.ObjectLostError'): - default_exception_hook(ex_class,ex,trace) - - console._showtraceback = filtered_exception_hook - else: - # Enable tab complete if possible - try: - import readline # Works only on Unix - readline.parse_and_bind("tab: complete") - except: - sudo_prefix = "" if platform.system() == "Windows" else "sudo " - print("Warning: could not enable tab-complete. User experience will suffer.\n" - "Run `{}pip install readline` and then restart this script to fix this." - .format(sudo_prefix)) + # Catch ObjectLostError (since disconnect is not always an error) + default_exception_hook = console._showtraceback + def filtered_exception_hook(ex_class, ex, trace): + if(ex_class.__module__+'.'+ex_class.__name__ != 'fibre.libfibre.ObjectLostError'): + default_exception_hook(ex_class,ex,trace) + + console._showtraceback = filtered_exception_hook + else: + # Enable tab complete if possible + try: + import readline # Works only on Unix + readline.parse_and_bind("tab: complete") + except: + sudo_prefix = "" if platform.system() == "Windows" else "sudo " + print("Warning: could not enable tab-complete. User experience will suffer.\n" + "Run `{}pip install readline` and then restart this script to fix this." + .format(sudo_prefix)) - import code - console = code.InteractiveConsole(locals=interactive_variables) - interact = lambda: console.interact(banner='') + import code + console = code.InteractiveConsole(locals=interactive_variables) + interact = lambda: console.interact(banner='') - # Catch ObjectLostError (since disconnect is not alway an error) - console.runcode("import sys") - console.runcode("default_exception_hook = sys.excepthook") - console.runcode("def filtered_exception_hook(ex_class, ex, trace):\n" - " if ex_class.__module__ + '.' + ex_class.__name__ != 'fibre.libfibre.ObjectLostError':\n" - " default_exception_hook(ex_class,ex,trace)") - console.runcode("sys.excepthook=filtered_exception_hook") + # Catch ObjectLostError (since disconnect is not alway an error) + console.runcode("import sys") + console.runcode("default_exception_hook = sys.excepthook") + console.runcode("def filtered_exception_hook(ex_class, ex, trace):\n" + " if ex_class.__module__ + '.' + ex_class.__name__ != 'fibre.libfibre.ObjectLostError':\n" + " default_exception_hook(ex_class,ex,trace)") + console.runcode("sys.excepthook=filtered_exception_hook") + # Launch shell + print_banner() + logger._skip_bottom_line = True + interact() - # Launch shell - print_banner() - logger._skip_bottom_line = True - interact() - app_shutdown_token.set() + shutdown_token[0] = True diff --git a/tools/odrive/pyfibre/fibre/utils.py b/tools/odrive/pyfibre/fibre/utils.py index 2893bd10..4e38c238 100644 --- a/tools/odrive/pyfibre/fibre/utils.py +++ b/tools/odrive/pyfibre/fibre/utils.py @@ -46,15 +46,20 @@ class Event(): Sets the event and invokes all subscribers if the event was not already set """ + subscribers = [] self._mutex.acquire() try: if not self._evt.is_set(): self._evt.set() for s in self._subscribers: - s() + subscribers.append(s) finally: self._mutex.release() + # Invoke subscribes with the mutex released to prevent deadlocks + for s in subscribers: + s() + def subscribe(self, handler): """ Invokes the specified handler exactly once as soon as the