diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 00000000..4a382938 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,3 @@ +libfibre*.so filter=lfs diff=lfs merge=lfs -text +libfibre*.dll filter=lfs diff=lfs merge=lfs -text +libfibre*.dylib filter=lfs diff=lfs merge=lfs -text diff --git a/README.md b/README.md new file mode 100644 index 00000000..cfaec288 --- /dev/null +++ b/README.md @@ -0,0 +1,17 @@ +# PyFibre + +This directory provides Python bindings for [Fibre](https://github.com/samuelsadok/fibre). Its home is located [here](https://github.com/samuelsadok/fibre/tree/master/python). There's also a standalone repository for this directory [here](https://github.com/samuelsadok/pyfibre). + +## Current Status + +Currently only client-side features are implemented, that means you can discover objects but you cannot publish objects. + +## How to use + +```python +import fibre + +with fibre.Domain("tcp-client:address=localhost,port=14220") as domain: + obj = domain.discover_one() + obj.test_function() +``` diff --git a/fibre/__init__.py b/fibre/__init__.py index 216f2e06..1ee712e9 100644 --- a/fibre/__init__.py +++ b/fibre/__init__.py @@ -1,4 +1,4 @@ from .utils import Event, Logger, TimeoutError from .shell import launch_shell -from .libfibre import find_all, find_any, ObjectLostError +from .libfibre import Domain, ObjectLostError diff --git a/fibre/libfibre-linux-amd64.so b/fibre/libfibre-linux-amd64.so new file mode 100755 index 00000000..00cbd444 --- /dev/null +++ b/fibre/libfibre-linux-amd64.so @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:d235ce21bf52f537f17cc317a141d0f9f0f2d1683c20222aecdae4cc1c5a4295 +size 3757680 diff --git a/fibre/libfibre-linux-armhf.so b/fibre/libfibre-linux-armhf.so new file mode 100755 index 00000000..2f3e9eb2 --- /dev/null +++ b/fibre/libfibre-linux-armhf.so @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:528c680e0417085e7187a231c05fc200ff03b9d903cb8418fbe63e9fbd66ac3d +size 750616 diff --git a/fibre/libfibre-macos-x86.dylib b/fibre/libfibre-macos-x86.dylib new file mode 100644 index 00000000..2af09dec --- /dev/null +++ b/fibre/libfibre-macos-x86.dylib @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:bf5105038e2787b8e3e05600a8d303a24dae6052364201c6ecb3233e3a947ddb +size 657528 diff --git a/fibre/libfibre-windows-amd64.dll b/fibre/libfibre-windows-amd64.dll new file mode 100755 index 00000000..868fc1e2 --- /dev/null +++ b/fibre/libfibre-windows-amd64.dll @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:727e24428ad48184b891a57e27763e3cd9be8e98a38befc5279d3fb11f11566d +size 1239566 diff --git a/fibre/libfibre.py b/fibre/libfibre.py index 038f51d7..b623fa56 100644 --- a/fibre/libfibre.py +++ b/fibre/libfibre.py @@ -9,8 +9,13 @@ from types import MethodType import concurrent import threading import time -from fibre.utils import Logger, Event import platform +from .utils import Logger, Event +import sys + +# Enable this for better tracebacks in some cases +#import tracemalloc +#tracemalloc.start(10) lib_names = { ('Linux', 'x86_64'): 'libfibre-linux-amd64.so', @@ -21,35 +26,54 @@ lib_names = { system_desc = (platform.system(), platform.machine()) -lib_dir = os.path.join( - os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))), - 'cpp') +script_dir = os.path.dirname(os.path.realpath(__file__)) +fibre_cpp_paths = [ + os.path.join(os.path.dirname(os.path.dirname(script_dir)), "cpp"), + os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(script_dir)))), "Firmware", "fibre-cpp") +] -def test_path(path): - return path if os.path.isfile(path) else None +def get_first(lst, predicate, default): + for item in lst: + if predicate(item): + return item + return default -lib_path = (test_path(os.path.join(lib_dir, 'libfibre.so')) or - test_path(os.path.join(lib_dir, 'libfibre.dll')) or - (test_path(os.path.join(lib_dir, lib_names[system_desc])) if (system_desc in lib_names) else None)) +if not system_desc in lib_names: + fibre_cpp_path = get_first(fibre_cpp_paths, os.path.isdir, None) + + if fibre_cpp_path is None: + instructions = ("Go to https://github.com/samuelsadok/fibre-cpp for " + "instructions on how to compile libfibre. Once you have compiled it, " + "add it to this folder.") + else: + instructions = ("Go to {} and run `make`. Then edit this file (libfibre.py) " + "to include the name of the binary that was generated by `make`.".format(fibre_cpp_path)) + + raise ModuleNotFoundError("libfibre is not supported on your platform ({} {}). {}".format(*system_desc, instructions)) + +lib_name = lib_names[system_desc] +search_paths = fibre_cpp_paths + [script_dir] + +lib_path = get_first( + (os.path.join(p, lib_name) for p in search_paths), + os.path.isfile, None) if lib_path is None: - raise ModuleNotFoundError("This package has no precompiled libfibre for your platform ({} {}). " - "Go to fibre/cpp/ and run `make` to compile libfibre for your platform.".format(*system_desc)) + raise ModuleNotFoundError("{} was not found in {}".format(lib_name, search_paths)) lib = windll.LoadLibrary(lib_path) if os.name == 'nt' else cdll.LoadLibrary(lib_path) # libfibre definitions --------------------------------------------------------# -PostSignature = CFUNCTYPE(c_void_p, CFUNCTYPE(None, c_void_p), POINTER(c_int)) -RegisterEventSignature = CFUNCTYPE(c_int, c_int, c_uint32, CFUNCTYPE(None, c_void_p), POINTER(c_int)) +PostSignature = CFUNCTYPE(c_int, CFUNCTYPE(None, c_void_p), POINTER(c_int)) +RegisterEventSignature = CFUNCTYPE(c_int, c_int, c_uint32, CFUNCTYPE(None, c_void_p, c_int), POINTER(c_int)) DeregisterEventSignature = CFUNCTYPE(c_int, c_int) CallLaterSignature = CFUNCTYPE(c_void_p, c_float, CFUNCTYPE(None, c_void_p), POINTER(c_int)) CancelTimerSignature = CFUNCTYPE(c_int, c_void_p) -ConstructObjectSignature = CFUNCTYPE(None, c_void_p, c_void_p, c_void_p, c_void_p, c_size_t) -DestroyObjectSignature = CFUNCTYPE(None, c_void_p, c_void_p) -OnFoundObjectSignature = CFUNCTYPE(None, c_void_p, c_void_p) +OnFoundObjectSignature = CFUNCTYPE(None, c_void_p, c_void_p, c_void_p) +OnLostObjectSignature = CFUNCTYPE(None, c_void_p, c_void_p) OnStoppedSignature = CFUNCTYPE(None, c_void_p, c_int) OnAttributeAddedSignature = CFUNCTYPE(None, c_void_p, c_void_p, c_void_p, c_size_t, c_void_p, c_void_p, c_size_t) @@ -57,13 +81,18 @@ OnAttributeRemovedSignature = CFUNCTYPE(None, c_void_p, c_void_p) OnFunctionAddedSignature = CFUNCTYPE(None, c_void_p, c_void_p, c_void_p, c_size_t, POINTER(c_char_p), POINTER(c_char_p), POINTER(c_char_p), POINTER(c_char_p)) OnFunctionRemovedSignature = CFUNCTYPE(None, c_void_p, c_void_p) -OnCallCompletedSignature = CFUNCTYPE(None, c_void_p, c_int, c_char_p) +OnCallCompletedSignature = CFUNCTYPE(c_int, c_void_p, c_int, c_void_p, c_void_p, POINTER(c_void_p), POINTER(c_size_t), POINTER(c_void_p), POINTER(c_size_t)) +OnTxCompletedSignature = CFUNCTYPE(None, c_void_p, c_void_p, c_int, c_void_p) +OnRxCompletedSignature = CFUNCTYPE(None, c_void_p, c_void_p, c_int, c_void_p) kFibreOk = 0 -kFibreCancelled = 1 -kFibreClosed = 2 -kFibreInvalidArgument = 3 -kFibreInternalError = 4 +kFibreBusy = 1 +kFibreCancelled = 2 +kFibreClosed = 3 +kFibreInvalidArgument = 4 +kFibreInternalError = 5 +kFibreProtocolError = 6 +kFibreHostUnreachable = 7 class LibFibreVersion(Structure): _fields_ = [ @@ -75,28 +104,45 @@ class LibFibreVersion(Structure): def __repr__(self): return "{}.{}.{}".format(self.major, self.minor, self.patch) +class LibFibreEventLoop(Structure): + _fields_ = [ + ("post", PostSignature), + ("register_event", RegisterEventSignature), + ("deregister_event", DeregisterEventSignature), + ("call_later", CallLaterSignature), + ("cancel_timer", CancelTimerSignature), + ] + libfibre_get_version = lib.libfibre_get_version libfibre_get_version.argtypes = [] libfibre_get_version.restype = POINTER(LibFibreVersion) version = libfibre_get_version().contents -if version.major != 0: +if (version.major, version.minor) != (0, 1): raise Exception("Incompatible libfibre version: {}".format(version)) libfibre_open = lib.libfibre_open -libfibre_open.argtypes = [PostSignature, RegisterEventSignature, DeregisterEventSignature, CallLaterSignature, CancelTimerSignature, ConstructObjectSignature, DestroyObjectSignature, c_void_p] +libfibre_open.argtypes = [LibFibreEventLoop] libfibre_open.restype = c_void_p libfibre_close = lib.libfibre_close libfibre_close.argtypes = [c_void_p] libfibre_close.restype = None +libfibre_open_domain = lib.libfibre_open_domain +libfibre_open_domain.argtypes = [c_void_p, c_char_p, c_size_t] +libfibre_open_domain.restype = c_void_p + +libfibre_close_domain = lib.libfibre_close_domain +libfibre_close_domain.argtypes = [c_void_p] +libfibre_close_domain.restype = None + libfibre_start_discovery = lib.libfibre_start_discovery -libfibre_start_discovery.argtypes = [c_void_p, c_char_p, c_size_t, c_void_p, OnFoundObjectSignature, OnStoppedSignature, c_void_p] -libfibre_start_discovery.restype = c_void_p +libfibre_start_discovery.argtypes = [c_void_p, c_void_p, OnFoundObjectSignature, OnLostObjectSignature, OnStoppedSignature, c_void_p] +libfibre_start_discovery.restype = None libfibre_stop_discovery = lib.libfibre_stop_discovery -libfibre_stop_discovery.argtypes = [c_void_p, c_void_p] +libfibre_stop_discovery.argtypes = [c_void_p] libfibre_stop_discovery.restype = None libfibre_subscribe_to_interface = lib.libfibre_subscribe_to_interface @@ -107,13 +153,25 @@ libfibre_get_attribute = lib.libfibre_get_attribute libfibre_get_attribute.argtypes = [c_void_p, c_void_p, POINTER(c_void_p)] libfibre_get_attribute.restype = c_int -libfibre_start_call = lib.libfibre_start_call -libfibre_start_call.argtypes = [c_void_p, c_void_p, c_char_p, c_size_t, c_char_p, c_size_t, c_void_p, OnCallCompletedSignature, c_void_p] -libfibre_start_call.restype = None +libfibre_call = lib.libfibre_call +libfibre_call.argtypes = [c_void_p, POINTER(c_void_p), c_int, c_void_p, c_size_t, c_void_p, c_size_t, POINTER(c_void_p), POINTER(c_void_p), OnCallCompletedSignature, c_void_p] +libfibre_call.restype = c_int -libfibre_cancel_call = lib.libfibre_cancel_call -libfibre_cancel_call.argtypes = [c_void_p] -libfibre_cancel_call.restype = None +libfibre_start_tx = lib.libfibre_start_tx +libfibre_start_tx.argtypes = [c_void_p, c_char_p, c_size_t, OnTxCompletedSignature, c_void_p] +libfibre_start_tx.restype = None + +libfibre_cancel_tx = lib.libfibre_cancel_tx +libfibre_cancel_tx.argtypes = [c_void_p] +libfibre_cancel_tx.restype = None + +libfibre_start_rx = lib.libfibre_start_rx +libfibre_start_rx.argtypes = [c_void_p, c_char_p, c_size_t, OnRxCompletedSignature, c_void_p] +libfibre_start_rx.restype = None + +libfibre_cancel_rx = lib.libfibre_cancel_rx +libfibre_cancel_rx.argtypes = [c_void_p] +libfibre_cancel_rx.restype = None # libfibre wrapper ------------------------------------------------------------# @@ -128,11 +186,15 @@ def _get_exception(status): elif status == kFibreCancelled: return asyncio.CancelledError() elif status == kFibreClosed: - return ObjectLostError() + return EOFError() elif status == kFibreInvalidArgument: return ArgumentError() elif status == kFibreInternalError: return Exception("internal libfibre error") + elif status == kFibreProtocolError: + return Exception("peer misbehaving") + elif status == kFibreHostUnreachable: + return ObjectLostError() else: return Exception("unknown libfibre error {}".format(status)) @@ -208,12 +270,237 @@ def run_coroutine_threadsafe(loop, func): future = concurrent.futures.Future() async def func_async(): try: - future.set_result(await func()) + result = func() + if hasattr(result, '__await__'): + result = await result + future.set_result(result) except Exception as ex: future.set_exception(ex) loop.call_soon_threadsafe(asyncio.ensure_future, func_async()) return future.result() +class TxStream(): + """Python wrapper for libfibre's LibFibreTxStream interface""" + + def __init__(self, libfibre, tx_stream_handle): + self._libfibre = libfibre + self._tx_stream_handle = tx_stream_handle + self._future = None + self._tx_buf = None + self._c_on_tx_completed = OnTxCompletedSignature(self._on_tx_completed) + self.is_closed = False + + def _on_tx_completed(self, ctx, tx_stream, status, tx_end): + tx_start = cast(self._tx_buf, c_void_p).value + + n_written = tx_end - tx_start + assert(n_written <= len(self._tx_buf)) + future = self._future + self._future = None + self._tx_buf = None + + if status == kFibreClosed: + self.is_closed = True + + if status == kFibreOk or status == kFibreClosed: + future.set_result(n_written) + else: + future.set_exception(_get_exception(status)) + + def write(self, data): + """ + Writes the provided data to the stream. Not all bytes are guaranteed to + be written. The caller should check the return value to determine the + actual number of bytes written. + + If a non-empty buffer is provided, this function will either write at + least one byte to the output, set is_closed to True or throw an + Exception (through the future). + + Currently only one write call may be active at a time (this may change + in the future). + + Returns: A future that completes with the number of bytes actually + written or an Exception. + """ + assert(self._future is None) + self._future = future = self._libfibre.loop.create_future() + self._tx_buf = data # Retain a reference to the buffer to prevent it from being garbage collected + + libfibre_start_tx(self._tx_stream_handle, + cast(self._tx_buf, c_char_p), len(self._tx_buf), + self._c_on_tx_completed, None) + + return future + + async def write_all(self, data): + """ + Writes all of the provided data to the stream or completes with an + Exception. + + If an empty buffer is provided, the underlying stream's write function + is still called at least once. + + Returns: A future that either completes with an empty result or with + an Exception. + """ + + while True: + n_written = await self.write(data) + data = data[n_written:] + if len(data) == 0: + break + elif self.is_closed: + raise EOFError("the TX stream was closed but there are still {} bytes left to send".format(len(data))) + assert(n_written > 0) # Ensure progress + +class RxStream(): + """Python wrapper for libfibre's LibFibreRxStream interface""" + + def __init__(self, libfibre, rx_stream_handle): + self._libfibre = libfibre + self._rx_stream_handle = rx_stream_handle + self._future = None + self._rx_buf = None + self._c_on_rx_completed = OnRxCompletedSignature(self._on_rx_completed) + self.is_closed = False + + def _on_rx_completed(self, ctx, rx_stream, status, rx_end): + rx_start = cast(self._rx_buf, c_void_p).value + + n_read = rx_end - rx_start + assert(n_read <= len(self._rx_buf)) + data = self._rx_buf[:n_read] + future = self._future + self._future = None + self._rx_buf = None + + if status == kFibreClosed: + self.is_closed = True + + if status == kFibreOk or status == kFibreClosed: + future.set_result(data) + else: + future.set_exception(_get_exception(status)) + + def read(self, n_read): + """ + Reads up to the specified number of bytes from the stream. + + If more than zero bytes are requested, this function will either read at + least one byte, set is_closed to True or throw an Exception (through the + future). + + Currently only one write call may be active at a time (this may change + in the future). + + Returns: A future that either completes with a buffer containing the + bytes that were read or completes with an Exception. + """ + assert(self._future is None) + self._future = future = self._libfibre.loop.create_future() + self._rx_buf = bytes(n_read) + + libfibre_start_rx(self._rx_stream_handle, + cast(self._rx_buf, c_char_p), len(self._rx_buf), + self._c_on_rx_completed, None) + + return future + + async def read_all(self, n_read): + """ + Reads the specified number of bytes from the stream or throws an + Exception. + + If zero bytes are requested, the underlying stream's read function + is still called at least once. + + Returns: A future that either completes with a buffer of size n_read or + an Exception. + """ + + data = bytes() + while True: + chunk = await self.read(n_read - len(data)) + data += chunk + if n_read == len(data): + break + elif self.is_closed: + raise EOFError() + assert(len(chunk) > 0) # Ensure progress + return data + + +class Call(object): + """ + This call behaves as you would expect an async generator to behave. This is + used to provide compatibility down to Python 3.5. + """ + def __init__(self, func): + self._func = func + self._call_handle = c_void_p(0) + self._is_started = False + self._should_close = False + self._is_closed = False + self._tx_buf = None + + def __aiter__(self): + return self + + async def asend(self, val): + assert(self._is_started == (not val is None)) + if not val is None: + self._tx_buf, self._rx_len, self._should_close = val + return await self.__anext__() + + async def __anext__(self): + if not self._is_started: + self._is_started = True + return None # This immitates the weird starting behavior of Python 3.6+ async generators iterators + + if self._is_closed: + raise StopAsyncIteration + + tx_end = c_void_p(0) + rx_end = c_void_p(0) + + rx_buf = b'\0' * self._rx_len + + call_id = insert_with_new_id(self._func._libfibre._calls, self) + + status = libfibre_call(self._func._func_handle, byref(self._call_handle), + kFibreClosed if self._should_close else kFibreOk, + cast(self._tx_buf, c_char_p), len(self._tx_buf), + cast(rx_buf, c_char_p), len(rx_buf), + byref(tx_end), byref(rx_end), self._func._libfibre.c_on_call_completed, call_id) + + if status == kFibreBusy: + self.ag_await = self._func._libfibre.loop.create_future() + status, tx_end, rx_end = await self.ag_await + self.ag_await = None + + if status != kFibreOk and status != kFibreClosed: + raise _get_exception(status) + + n_written = tx_end - cast(self._tx_buf, c_void_p).value + self._tx_buf = self._tx_buf[n_written:] + n_read = rx_end - cast(rx_buf, c_void_p).value + rx_buf = rx_buf[:n_read] + + if status != kFibreOk: + self._is_closed = True + return self._tx_buf, rx_buf, self._is_closed + + async def cancel(): + # TODO: this doesn't follow the official Python async generator protocol. Should implement aclose() instead. + status = libfibre_call(self._func._func_handle, byref(self._call_handle), kFibreOk, + 0, 0, 0, 0, 0, 0, self._func._libfibre.c_on_call_completed, call_id) + + #async def aclose(self): + # assert(self._is_started and not self._is_closed) + # return self._tx_buf, rx_buf, self._is_closed + + class RemoteFunction(object): """ Represents a callable function that maps to a function call on a remote object. @@ -224,33 +511,50 @@ class RemoteFunction(object): self._inputs = inputs self._outputs = outputs self._rx_size = sum(codec.get_length() for _, _, codec in self._outputs) - self._calls = {} - self._c_on_completed = OnCallCompletedSignature(self._on_completed) - def _on_completed(self, ctx, status, end_ptr): - call = self._calls.pop(ctx) + async def async_call(self, args, cancellation_token): + #print("making call on " + hex(args[0]._obj_handle)) + tx_buf = bytes() + for i, arg in enumerate(self._inputs): + tx_buf += arg[2].serialize(self._libfibre, args[i]) + rx_buf = bytes() - if status != kFibreOk: - call['future'].set_exception(_get_exception(status)) + agen = Call(self) + + if not cancellation_token is None: + cancellation_token.add_done_callback(agen.cancel) + + try: + assert(await agen.asend(None) is None) + + is_closed = False + while not is_closed: + tx_buf, rx_chunk, is_closed = await agen.asend((tx_buf, self._rx_size - len(rx_buf), True)) + rx_buf += rx_chunk + + finally: + if not cancellation_token is None: + cancellation_token.remove_done_callback(agen.cancel) + + assert(len(rx_buf) == self._rx_size) + + outputs = [] + for arg in self._outputs: + arg_length = arg[2].get_length() + outputs.append(arg[2].deserialize(self._libfibre, rx_buf[:arg_length])) + rx_buf = rx_buf[arg_length:] + + if len(outputs) == 0: + return + elif len(outputs) == 1: + return outputs[0] else: - pos = 0 - outputs = [] + return tuple(outputs) - for arg in self._outputs: - arg_length = arg[2].get_length() - outputs.append(arg[2].deserialize(self._libfibre, call['rx_buf'][pos:(pos + arg_length)])) - pos += arg_length - - if len(outputs) == 0: - call['future'].set_result(None) - elif len(outputs) == 1: - call['future'].set_result(outputs[0]) - else: - call['future'].set_result(tuple(outputs)) - - def __call__(self, instance, *args): + def __call__(self, *args, cancellation_token = None): """ - Starts invoking the function on the remote object. + Starts invoking the remote function. The first argument is usually a + remote object. If this function is called from the Fibre thread then it is nonblocking and returns an asyncio.Future. If it is called from another thread then it blocks until the function completes and returns the result(s) of the @@ -258,28 +562,13 @@ class RemoteFunction(object): """ if threading.current_thread() != libfibre_thread: - return run_coroutine_threadsafe(instance._libfibre.loop, lambda: self.__call__(instance, *args)) + return run_coroutine_threadsafe(self._libfibre.loop, lambda: self.__call__(*args)) if (len(self._inputs) != len(args)): raise TypeError("expected {} arguments but have {}".format(len(self._inputs), len(args))) - # All of these variables need to be protected from the garbage collector - # for the duration of the call. - call = { - 'handle': c_size_t(0), - 'tx_buf': b''.join(self._inputs[i][2].serialize(self._libfibre, args[i]) - for i in range(len(self._inputs))), # Assemble TX buffer - 'rx_buf': b'\0' * self._rx_size, # Allocate RX buffer - 'future': instance._libfibre.loop.create_future(), - } - call_id = insert_with_new_id(self._calls, call) - - libfibre_start_call(instance._obj_handle, self._func_handle, - cast(call['tx_buf'], c_char_p), len(call['tx_buf']), - cast(call['rx_buf'], c_char_p), len(call['rx_buf']), - byref(call['handle']), self._c_on_completed, call_id) - - return call['future'] + coro = self.async_call(args, cancellation_token) + return asyncio.ensure_future(coro, loop=self._libfibre.loop) def __get__(self, instance, owner): return MethodType(self, instance) if instance else self @@ -416,36 +705,41 @@ class LibFibre(): self.c_deregister_event = DeregisterEventSignature(self._deregister_event) self.c_call_later = CallLaterSignature(self._call_later) self.c_cancel_timer = CancelTimerSignature(self._cancel_timer) - self.c_construct_object = ConstructObjectSignature(self._construct_object) - self.c_destroy_object = DestroyObjectSignature(self._destroy_object) self.c_on_found_object = OnFoundObjectSignature(self._on_found_object) + self.c_on_lost_object = OnLostObjectSignature(self._on_lost_object) self.c_on_discovery_stopped = OnStoppedSignature(self._on_discovery_stopped) self.c_on_attribute_added = OnAttributeAddedSignature(self._on_attribute_added) self.c_on_attribute_removed = OnAttributeRemovedSignature(self._on_attribute_removed) self.c_on_function_added = OnFunctionAddedSignature(self._on_function_added) self.c_on_function_removed = OnFunctionRemovedSignature(self._on_function_removed) + self.c_on_call_completed = OnCallCompletedSignature(self._on_call_completed) self.timer_map = {} self.eventfd_map = {} self.interfaces = {} # key: libfibre handle, value: python class self.discovery_processes = {} # key: ID, value: python dict - self._objects = {} # key: libfibre handle, value: pyhton class + self._objects = {} # key: libfibre handle, value: python class + self._calls = {} # key: libfibre handle, value: Call object - self.ctx = c_void_p(libfibre_open( - self.c_post, - self.c_register_event, self.c_deregister_event, - self.c_call_later, self.c_cancel_timer, - self.c_construct_object, self.c_destroy_object, None)) + event_loop = LibFibreEventLoop() + event_loop.post = self.c_post + event_loop.register_event = self.c_register_event + event_loop.deregister_event = self.c_deregister_event + event_loop.call_later = self.c_call_later + event_loop.cancel_timer = self.c_cancel_timer + + self.ctx = c_void_p(libfibre_open(event_loop)) def _post(self, callback, ctx): self.loop.call_soon_threadsafe(callback, ctx) + return 0 def _register_event(self, event_fd, events, callback, ctx): self.eventfd_map[event_fd] = events if (events & 1): - self.loop.add_reader(event_fd, callback, ctx) + self.loop.add_reader(event_fd, lambda x: callback(x, 1), ctx) if (events & 4): - self.loop.add_writer(event_fd, callback, ctx) + self.loop.add_writer(event_fd, lambda x: callback(x, 4), ctx) if (events & 0xfffffffa): raise Exception("unsupported event mask " + str(events)) return 0 @@ -484,22 +778,45 @@ class LibFibre(): libfibre_subscribe_to_interface(intf_handle, self.c_on_attribute_added, self.c_on_attribute_removed, self.c_on_function_added, self.c_on_function_removed, intf_handle) return py_intf - def _construct_object(self, ctx, obj, intf, name, name_length): - #increment_libfibre_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 _load_py_obj(self, obj_handle, intf_handle): + if not obj_handle in self._objects: + name = None # TODO: load from libfibre + py_intf = self._load_py_intf(name, intf_handle) + py_obj = py_intf(self, obj_handle) + self._objects[obj_handle] = py_obj + else: + py_obj = self._objects[obj_handle] + py_obj._refcount += 1 + return py_obj - def _destroy_object(self, ctx, obj): + def _release_py_obj(self, obj_handle): + py_obj = self._objects[obj_handle] + py_obj._refcount -= 1 + 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() - 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)) + def _on_found_object(self, ctx, obj, intf): + py_obj = self._load_py_obj(obj, intf) + discovery = self.discovery_processes[ctx] + discovery._unannounced.append(py_obj) + old_future = discovery._future + discovery._future = self.loop.create_future() + old_future.set_result(None) + + def _on_lost_object(self, ctx, obj): + self._free_py_obj(obj) def _on_discovery_stopped(self, ctx, result): print("discovery stopped") @@ -517,7 +834,7 @@ class LibFibre(): setattr(intf, "_" + name + "_property", RemoteAttribute(self, attr, subintf, subintf_name, False, False)) def _on_attribute_removed(self, ctx, attr): - print("attribute removed") + print("attribute removed") # TODO def _on_function_added(self, ctx, func, name, name_length, input_names, input_codecs, output_names, output_codecs): name = string_at(name, name_length).decode('utf-8') @@ -527,25 +844,105 @@ class LibFibre(): setattr(intf, name, RemoteFunction(self, func, inputs, outputs)) def _on_function_removed(self, ctx, func): - print("function removed") + print("function removed") # TODO - def start_discovery(self, path, on_obj_discovered, cancellation_token): + def _on_call_completed(self, ctx, status, tx_end, rx_end, tx_buf, tx_len, rx_buf, rx_len): + call = self._calls.pop(ctx) + + call.ag_await.set_result((status, tx_end, rx_end)) + + 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(): + def __init__(self, domain): + self._domain = domain + self._id = 0 + self._discovery_handle = c_void_p(0) + self._unannounced = [] + self._future = domain._libfibre.loop.create_future() + + async def _next(self): + if len(self._unannounced) == 0: + await self._future + return self._unannounced.pop(0) + + def _stop(self): + self._domain._libfibre.discovery_processes.pop(self._id) + libfibre_stop_discovery(self._discovery_handle) + +class _Domain(): + def __init__(self, libfibre, handle): + self._libfibre = libfibre + self._domain_handle = handle + + def _close(self): + libfibre_close_domain(self._domain_handle) + self._domain_handle = None + #decrement_lib_refcount() + + def _start_discovery(self): + discovery = Discovery(self) + discovery._id = insert_with_new_id(self._libfibre.discovery_processes, discovery) + libfibre_start_discovery(self._domain_handle, byref(discovery._discovery_handle), self._libfibre.c_on_found_object, self._libfibre.c_on_lost_object, self._libfibre.c_on_discovery_stopped, discovery._id) + return discovery + + async def _discover_one(self): + discovery = self._start_discovery() + obj = await discovery._next() + discovery._stop() + return obj + + def discover_one(self): + return run_coroutine_threadsafe(self._libfibre.loop, self._discover_one) + + + +class Domain(): + def __init__(self, path): + increment_lib_refcount() + self._opened_domain = run_coroutine_threadsafe(libfibre.loop, lambda: Domain._open(path)) + + def _open(path): + assert(libfibre_thread == threading.current_thread()) buf = path.encode('ascii') + domain_handle = libfibre_open_domain(libfibre.ctx, buf, len(buf)) + return _Domain(libfibre, domain_handle) - discovery = { - 'handle': c_void_p(0), - 'callback': on_obj_discovered - } - discovery_id = insert_with_new_id(self.discovery_processes, discovery) - - cancellation_token.subscribe(lambda: libfibre_stop_discovery(self.ctx, discovery['handle'])) - libfibre_start_discovery(self.ctx, buf, len(buf), byref(discovery['handle']), self.c_on_found_object, self.c_on_discovery_stopped, discovery_id) - + def __enter__(self): + return self._opened_domain + def __exit__(self, type, value, traceback): + run_coroutine_threadsafe(self._opened_domain._libfibre.loop, self._opened_domain._close) + self._opened_domain = None + decrement_lib_refcount() libfibre = None -def run_event_loop(): +def _run_event_loop(): global libfibre global terminate_libfibre @@ -576,7 +973,7 @@ lock = threading.Lock() libfibre_refcount = 0 libfibre_thread = None -def increment_libfibre_refcount(): +def increment_lib_refcount(): global libfibre_refcount global libfibre_thread @@ -585,7 +982,7 @@ def increment_libfibre_refcount(): #print("inc refcount to {}".format(libfibre_refcount)) if libfibre_refcount == 1: - libfibre_thread = threading.Thread(target = run_event_loop) + libfibre_thread = threading.Thread(target = _run_event_loop) libfibre_thread.start() while libfibre is None: @@ -610,64 +1007,35 @@ def decrement_lib_refcount(): libfibre_thread = None -def find_all(path, serial_number, - on_object_discovered, - search_cancellation_token, - channel_termination_token, - logger): - """ - 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_libfibre_refcount() - channel_termination_token.subscribe(lambda: decrement_lib_refcount()) - if serial_number is None or (await fibre.utils.get_serial_number_str(obj)) == serial_number: - result = on_object_discovered(obj) - if not result is None: - await result - - increment_libfibre_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 find_any(path="usb", serial_number=None, - search_cancellation_token=None, channel_termination_token=None, - timeout=None, logger=Logger(verbose=False), find_multiple=False): - """ - Blocks until the first matching Fibre object is connected and then returns that object - """ - result = [] - done_signal = Event(search_cancellation_token) - def did_discover_object(obj): - result.append(obj) - if find_multiple: - if len(result) >= int(find_multiple): - done_signal.set() - else: - done_signal.set() - - find_all(path, serial_number, did_discover_object, done_signal, channel_termination_token, logger) - - try: - done_signal.wait(timeout=timeout) - except TimeoutError: - if not find_multiple: - return None - finally: - done_signal.set() # terminate find_all - - if find_multiple: - return result - else: - return result[0] if len(result) > 0 else 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): diff --git a/fibre/libwinpthread-1.dll b/fibre/libwinpthread-1.dll new file mode 100755 index 00000000..cc8ca20e Binary files /dev/null and b/fibre/libwinpthread-1.dll differ diff --git a/fibre/shell.py b/fibre/shell.py index 6fbfdf1f..31ba99a5 100644 --- a/fibre/shell.py +++ b/fibre/shell.py @@ -58,6 +58,7 @@ def get_user_name(interactive_variables, obj): return "anonymous_remote_object_" + str(self._obj_handle) def launch_shell(args, + object_filter, interactive_variables, print_banner, print_help, logger, app_shutdown_token, @@ -77,7 +78,7 @@ def launch_shell(args, # Connect to device logger.debug("Waiting for {}...".format(branding_long)) - fibre.find_all(args.path, args.serial_number, + 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, diff --git a/fibre/utils.py b/fibre/utils.py index 4a6f8965..2893bd10 100644 --- a/fibre/utils.py +++ b/fibre/utils.py @@ -23,12 +23,6 @@ if sys.version_info < (3, 3): else: TimeoutError = TimeoutError -def get_serial_number_str(device): - if hasattr(device, 'serial_number'): - return format(device.serial_number, 'x').upper() - else: - return "[unknown serial number]" - ## Threading utils ## class Event(): """