diff --git a/Firmware/fibre-cpp/.gitignore b/Firmware/fibre-cpp/.gitignore index 7bbdcb78..d06737af 100644 --- a/Firmware/fibre-cpp/.gitignore +++ b/Firmware/fibre-cpp/.gitignore @@ -1 +1,4 @@ -/third_party \ No newline at end of file +/third_party +build/ +build-*/ +/.tup diff --git a/Firmware/fibre-cpp/Makefile b/Firmware/fibre-cpp/Makefile index 0a90c060..0fc88d9b 100644 --- a/Firmware/fibre-cpp/Makefile +++ b/Firmware/fibre-cpp/Makefile @@ -1,46 +1,6 @@ -ifeq ($(OS),Windows_NT) - target = windows - ifeq ($(PROCESSOR_ARCHITEW6432),AMD64) - outname = libfibre-windows-amd64.dll - else ifeq ($(PROCESSOR_ARCHITECTURE),AMD64) - outname = libfibre-windows-amd64.dll - else - $(error unsupported platform ${PROCESSOR_ARCHITECTURE}) - endif -else ifeq ($(shell uname -s),Linux) - target = linux - ifeq ($(shell uname -m),x86_64) - outname = libfibre-linux-amd64.so - else ifeq (($(shell uname -m),armv7l) - outname = libfibre-linux-armhf.so - else - $(error unsupported platform) - endif -else ifeq ($(shell uname -s),Darwin) - target = macos - ifeq ($(shell uname -m),x86_64) - outname = libfibre-macos-multiarch.dylib - else - $(error unsupported platform) - endif -endif - - -FILES=libfibre.cpp \ - platform_support/libusb_transport.cpp \ - legacy_protocol.cpp \ - legacy_object_client.cpp \ - logging.cpp - -linux: - g++ -shared -o $(outname) -fPIC -std=c++11 -I/usr/include/libusb-1.0 -I./include -DFIBRE_COMPILE -DFIBRE_ENABLE_CLIENT \ - $(FILES) \ - -lusb-1.0 -windows: - g++ -shared -o $(outname) -fPIC -std=c++11 -I./third_party/libusb-windows/libusb-1.0.23/include/libusb-1.0 -I./include -DFIBRE_COMPILE -DFIBRE_ENABLE_CLIENT \ - $(FILES) \ - -static-libgcc -Wl,-Bstatic -lstdc++ ./third_party/libusb-windows/libusb-1.0.23/MinGW64/static/libusb-1.0.a -Wl,-Bdynamic - -# To check exported symbols run: -# nm -D libfibre.so | grep ' T ' +all: + tup --no-environ-check build-local + tup --no-environ-check build-wasm + cp build-local/libfibre-* ../python/fibre/ + cp build-wasm/libfibre-* ../js/ diff --git a/Firmware/fibre-cpp/README.md b/Firmware/fibre-cpp/README.md index 33be1144..1c8438c7 100644 --- a/Firmware/fibre-cpp/README.md +++ b/Firmware/fibre-cpp/README.md @@ -1,17 +1,22 @@ +# fibre-cpp -## Platform Compatibility +This directory provides the C/C++ reference implementation of [Fibre](https://github.com/samuelsadok/fibre). Its home is located [here](https://github.com/samuelsadok/fibre/tree/master/cpp). There's also a standalone repository for this directory [here](https://github.com/samuelsadok/fibre-cpp). -| | Windows | macOS [1] | Linux | -|-----|---------|-----------|-------| -| USB | yes | yes | yes | +## How to use - - [1] macOS 10.9 (Mavericks) or later +The code files in this directory can be embedded directly into user applications by including the C and C++ files in the user application's compile and link process. This is the recommended approach when including Fibre in embedded systems. Currently there's no nice walkthrough for this but you can refer to the [ODrive Firmware](https://github.com/madcowswe/ODrive/tree/devel/Firmware) as an example. -## `libfibre` API +If your application runs on a major Operating System it's recommended that you link to **libfibre** instead. This is essentially fibre-cpp packaged as a nice shared library (aka DLL) accompanied by a header file. We provide precompiled libfibre binaries for most platforms on the main project's [release page](https://github.com/samuelsadok/fibre/releases). The API is documented in [libfibre.h](include/fibre/libfibre.h). -Refer to [libfibre.h](include/fibre/libfibre.h) for documentation of the API. +_Note:_ Libfibre's API currently only exposes the client role, not the server role. That means you can use it to discover and access remote objects but you cannot use it to expose local objects yet. For this you have to embed the code files directly. -## Build instructions +_Note:_ fibre-cpp makes use of dynamic memory if and only if the client role is used (this limitation might be lifted at some point). + +## How to embed + +Refer to + +## `libfibre` Build Instructions ### Windows 1. Download MinGW from [here](https://sourceforge.net/projects/mingw-w64/files/Toolchains%20targetting%20Win32/Personal%20Builds/mingw-builds/installer/mingw-w64-install.exe/download) and install it. diff --git a/Firmware/fibre-cpp/Tupfile.lua b/Firmware/fibre-cpp/Tupfile.lua new file mode 100644 index 00000000..f6e611d0 --- /dev/null +++ b/Firmware/fibre-cpp/Tupfile.lua @@ -0,0 +1,136 @@ + +tup.include('package.lua') + +CFLAGS = {'-I./include -fPIC -std=c++11 -DFIBRE_COMPILE -DFIBRE_ENABLE_CLIENT'} +LDFLAGS = {'-static-libstdc++'} + +-- Runs the specified shell command immediately (not as part of the dependency +-- graph). +-- Returns the values (return_code, stdout) where stdout has the trailing new +-- line removed. +function run_now(command) + local handle + handle = io.popen(command) + local output = handle:read("*a") + local rc = {handle:close()} + if not rc[1] then + error("failed to invoke "..command) + end + return string.sub(output, 0, -2) +end + +if tup.getconfig("CC") == "" then + CC = 'clang++' + LINKER = 'clang++' +else + CC = tup.getconfig("CC") + LINKER = tup.getconfig("CC") +end + +function get_bool_config(name, default) + if tup.getconfig(name) == "" then + return default + elseif tup.getconfig(name) == "true" then + return true + elseif tup.getconfig(name) == "false" then + return false + else + error(name.." ("..tup.getconfig(name).." must be 'true' or 'false'.") + end +end + +CFLAGS += tup.getconfig("CFLAGS") +LDFLAGS += tup.getconfig("LDFLAGS") +DEBUG = get_bool_config("DEBUG", true) +USE_PKGCONF = get_bool_config("USE_PKGCONF", true) +ENABLE_LIBUSB = get_bool_config("ENABLE_LIBUSB", true) + +if USE_PKGCONF and ENABLE_LIBUSB then + CFLAGS += run_now("pkgconf libusb-1.0 --cflags") + LDFLAGS += run_now("pkgconf libusb-1.0 --libs") +end + +machine = run_now(CC..' -dumpmachine') -- works with both clang and GCC + +BUILD_TYPE='-shared' + +if string.find(machine, "x86_64.*%-linux%-.*") then + outname = 'libfibre-linux-amd64.so' + LDFLAGS += '-lpthread -Wl,--version-script=libfibre.version -Wl,--gc-sections' + STRIP = not DEBUG +elseif string.find(machine, "arm.*%-linux%-.*") then + outname = 'libfibre-linux-armhf.so' + LDFLAGS += '-lpthread -Wl,--version-script=libfibre.version -Wl,--gc-sections' + STRIP = false +elseif string.find(machine, "x86_64.*-mingw.*") then + outname = 'libfibre-windows-amd64.dll' + LDFLAGS += '-lpthread -Wl,--version-script=libfibre.version' + STRIP = not DEBUG +elseif string.find(machine, "x86_64.*-apple-.*") then + outname = 'libfibre-macos-x86.dylib' + STRIP = false +elseif string.find(machine, "wasm.*") then + outname = 'libfibre-wasm.js' + STRIP = false + BUILD_TYPE = '' +else + error('unknown machine identifier '..machine) +end + +LDFLAGS += BUILD_TYPE + +if DEBUG then + CFLAGS += '-O1 -g' +else + CFLAGS += '-O3' -- TODO: add back -lfto +end + +function compile(src_file, obj_file) + tup.frule{ + inputs={src_file}, + command='^co^ '..CC..' -c %f '..tostring(CFLAGS)..' -fdebug-prefix-map=/Data/Projects/fibre/cpp/build-local=/Data/Projects/fibre/cpp -o %o', + outputs={obj_file} + } +end + +code_files = fibre_package.core_files + +if ENABLE_LIBUSB then + tup.append_table(code_files, fibre_package.features["LIBUSB"]) + CFLAGS += '-DFIBRE_ENABLE_LIBUSB=1' +end +if get_bool_config("ENABLE_LOGGING", true) then + tup.append_table(code_files, fibre_package.features["LOGGING"]) +end + +for _, src_file in pairs(code_files) do + obj_file = "build/"..src_file:gsub("/","_")..".o" + object_files += obj_file + compile(src_file, obj_file) +end + +if not STRIP then + compile_outname=outname +else + compile_outname=outname..'.fat' +end + +if tup.ext(outname) == 'js' then + extra_outputs = {tup.base(compile_outname)..'.wasm'} +else + extra_outputs = {} +end + +tup.frule{ + inputs=object_files, + command='^c^ '..LINKER..' %f '..tostring(CFLAGS)..' '..tostring(LDFLAGS)..' -o %o', + outputs={compile_outname, extra_outputs=extra_outputs} +} + +if STRIP then + tup.frule{ + inputs={compile_outname}, + command='strip --strip-all --discard-all %f -o %o', + outputs={outname} + } +end diff --git a/Firmware/fibre-cpp/channel_discoverer.hpp b/Firmware/fibre-cpp/channel_discoverer.hpp new file mode 100644 index 00000000..3082e806 --- /dev/null +++ b/Firmware/fibre-cpp/channel_discoverer.hpp @@ -0,0 +1,29 @@ +#ifndef __FIBRE_CHANNEL_DISCOVERER +#define __FIBRE_CHANNEL_DISCOVERER + +#include "async_stream.hpp" +#include + +namespace fibre { + +struct ChannelDiscoveryResult { + FibreStatus status; + AsyncStreamSource* rx_channel; + AsyncStreamSink* tx_channel; + size_t mtu; +}; + +struct ChannelDiscoveryContext {}; + +class ChannelDiscoverer { +public: + virtual void start_channel_discovery( + const char* specs, size_t specs_len, + ChannelDiscoveryContext** handle, + Completer& on_found_channels) = 0; + virtual int stop_channel_discovery(ChannelDiscoveryContext* handle) = 0; +}; + +} + +#endif // __FIBRE_CHANNEL_DISCOVERER \ No newline at end of file diff --git a/Firmware/fibre-cpp/compile_for_all_platforms.sh b/Firmware/fibre-cpp/compile_for_all_platforms.sh index 688d6ac7..9ef1a74b 100755 --- a/Firmware/fibre-cpp/compile_for_all_platforms.sh +++ b/Firmware/fibre-cpp/compile_for_all_platforms.sh @@ -9,11 +9,6 @@ set -euo pipefail # p7zip # apple-darwin-osxcross -# TODO: support C++11 - -CROSS_PLATFORM_CFLAGS='-O3 -fPIC -std=c++11 -DFIBRE_COMPILE -DFIBRE_ENABLE_CLIENT' -CROSS_PLATFORM_GCC_FLAGS='-flto -Wl,--version-script=libfibre.version -Wl,--gc-sections' - mkdir -p third_party # Usage: download_deb_pkg destination-dir url @@ -63,27 +58,6 @@ function compile_libusb() { popd > /dev/null } -FILES=('libfibre.cpp' - 'platform_support/libusb_transport.cpp' - 'legacy_protocol.cpp' - 'legacy_object_client.cpp' - 'logging.cpp') - -function build_for_linux() { - arch_name="$1" - "${CXX}" -shared -o libfibre-"$arch_name".so -I./include \ - ${CFLAGS} \ - ${CROSS_PLATFORM_CFLAGS} \ - ${CROSS_PLATFORM_GCC_FLAGS} \ - "${FILES[@]}" \ - ${LIBS} \ - -lpthread \ - -Wl,--unresolved-symbols=ignore-in-shared-libs -static-libstdc++ - - # Could reduce binary size by 35% by using something like - #arm-none-eabi-strip --strip-all --discard-all libfibre-linux-armhf.so -} - ### Download/compile prerequisites @@ -106,42 +80,7 @@ fi popd > /dev/null - -### Compile libfibre - -echo "building libfibre for Linux (AMD64)..." -CFLAGS="-I./third_party/libusb-dev-amd64/usr/include/libusb-1.0" \ -LIBS="third_party/libusb-amd64/lib/x86_64-linux-gnu/libusb-1.0.so.0.2.0" \ -CXX="x86_64-pc-linux-gnu-g++" \ - build_for_linux 'linux-amd64' - -echo "building libfibre for Linux (ARM)..." -CFLAGS="-I./third_party/libusb-dev-armhf/usr/include/libusb-1.0 -L./third_party/libstdc++-linux-armhf/usr/lib/gcc-cross/arm-linux-gnueabihf/10" \ -LIBS="third_party/libusb-armhf/lib/arm-linux-gnueabihf/libusb-1.0.so.0.2.0" \ -CXX="arm-linux-gnueabihf-g++" \ - build_for_linux 'linux-armhf' - - -### Windows - -echo "building libfibre for Windows (AMD64)..." -x86_64-w64-mingw32-g++ -shared -o libfibre-windows-amd64.dll -I./include \ - -I./third_party/libusb-windows/libusb-1.0.23/include/libusb-1.0 \ - ${CROSS_PLATFORM_CFLAGS} \ - ${CROSS_PLATFORM_GCC_FLAGS} \ - "${FILES[@]}" \ - -static-libgcc \ - -Wl,-Bstatic \ - -lstdc++ \ - ./third_party/libusb-windows/libusb-1.0.23/MinGW64/static/libusb-1.0.a \ - -Wl,-Bdynamic -cp /usr/x86_64-w64-mingw32/bin/libwinpthread-1.dll . - -# The windows compiler keeps a ton of debug symbols for some reason. Stripping -# the DLL reduces its size by a factor of 10. -x86_64-w64-mingw32-strip --strip-all --discard-all libfibre-windows-amd64.dll - -### macOS +### compile libusb for macOS # Link are broken: # …ions/Current/Headers $ ls -l IOReturn.h @@ -159,20 +98,85 @@ while IFS= read -r link; do fi done <<< "$(find /opt/osxcross/SDK/MacOSX10.13.sdk/System/Library/Frameworks/IOKit.framework -xtype l)" -export PATH="/opt/osxcross/bin:$PATH" -export LD_LIBRARY_PATH="/opt/osxcross/lib" -export CFLAGS='-I/opt/osxcross/SDK/MacOSX10.13.sdk/usr/include -arch i386 -arch x86_64' -export MACOSX_DEPLOYMENT_TARGET='10.9' -CC='o64-clang' \ +echo "building libusb for macOS..." + +CC='/opt/osxcross/bin/o64-clang' \ +LD_LIBRARY_PATH="/opt/osxcross/lib" \ +CFLAGS='-I/opt/osxcross/SDK/MacOSX10.13.sdk/usr/include -arch i386 -arch x86_64' \ +MACOSX_DEPLOYMENT_TARGET='10.9' \ compile_libusb 'macos-amd64' 'x86_64-apple-darwin17' -echo "building libfibre for macOS (x86)" -o64-clang++ -shared -o libfibre-macos-x86.dylib -I./include \ - -I./third_party/libusb-windows/libusb-1.0.23/include/libusb-1.0 \ - -arch x86_64 -arch i386 \ - ${CROSS_PLATFORM_CFLAGS} \ - "${FILES[@]}" \ - -static-libstdc++ \ - ./third_party/libusb-1.0.23/build-macos-amd64/libusb/.libs/libusb-1.0.a \ - -framework CoreFoundation -framework IOKit + + +### Prepare tup.config files + +mkdir -p build-linux-amd64 +cat < build-linux-amd64/tup.config +CONFIG_DEBUG=false +CONFIG_CC="clang++" +CONFIG_CFLAGS="-I./third_party/libusb-dev-armhf/usr/include/libusb-1.0" +CONFIG_LDFLAGS="-L./third_party/libusb-amd64/lib/x86_64-linux-gnu/libusb-1.0.so.0.2.0" +CONFIG_USE_PKGCONF=false +EOF + +mkdir -p build-linux-armhf +cat < build-linux-armhf/tup.config +CONFIG_DEBUG=false +CONFIG_CC="arm-linux-gnueabihf-g++" +CONFIG_CFLAGS="-I./third_party/libusb-dev-armhf/usr/include/libusb-1.0" +CONFIG_LDFLAGS="-L./third_party/libstdc++-linux-armhf/usr/lib/gcc-cross/arm-linux-gnueabihf/10 third_party/libusb-armhf/lib/arm-linux-gnueabihf/libusb-1.0.so.0.2.0" +CONFIG_USE_PKGCONF=false +EOF + +mkdir -p build-windows-amd64 +cat < build-windows-amd64/tup.config +CONFIG_DEBUG=false +CONFIG_CC="x86_64-w64-mingw32-g++" +CONFIG_CFLAGS="-I./third_party/libusb-windows/libusb-1.0.23/include/libusb-1.0" +CONFIG_LDFLAGS="-static-libgcc ./third_party/libusb-windows/libusb-1.0.23/MinGW64/static/libusb-1.0.a" +CONFIG_USE_PKGCONF=false +EOF + +mkdir -p build-macos-x86 +cat < build-macos-x86/tup.config +CONFIG_DEBUG=false +CONFIG_CC="LD_LIBRARY_PATH=/opt/osxcross/lib MACOSX_DEPLOYMENT_TARGET=10.9 /opt/osxcross/bin/o64-clang++" +CONFIG_CFLAGS="-I./third_party/libusb-1.0.23/libusb -arch i386 -arch x86_64" +CONFIG_LDFLAGS="./third_party/libusb-1.0.23/build-macos-amd64/libusb/.libs/libusb-1.0.a -framework CoreFoundation -framework IOKit" +CONFIG_USE_PKGCONF=false +EOF + +# Uncomment this to generate the WebAssembly build target. If you do this you +# have to finish a compile without tup before tup works. This is because +# emscripten generates some cache files which tup is unhappy about. +#mkdir -p build-wasm +#cat < build-wasm/tup.config +#CONFIG_DEBUG=true +#CONFIG_CC=/usr/lib/emscripten/em++ +#CONFIG_CFLAGS="-include emscripten.h -DFIBRE_PUBLIC=EMSCRIPTEN_KEEPALIVE" +#CONFIG_USE_PKGCONF=false +#CONFIG_ENABLE_LIBUSB=false +#EOF + +mkdir -p build-local +echo "" > build-local/tup.config + + +### Invoke tup for all configs + +tup --no-environ-check + + +### Copy to other locations + +function copy_to() { + cp build-linux-amd64/libfibre-linux-amd64.so "$1/" + cp build-linux-armhf/libfibre-linux-armhf.so "$1/" + cp build-windows-amd64/libfibre-windows-amd64.dll "$1/" + cp /usr/x86_64-w64-mingw32/bin/libwinpthread-1.dll "$1/" + cp build-macos-x86/libfibre-macos-x86.dylib "$1/" +} + +copy_to ../python/fibre/ +cp build-wasm/libfibre-* ../js/ diff --git a/Firmware/fibre-cpp/include/fibre/libfibre.h b/Firmware/fibre-cpp/include/fibre/libfibre.h index dd60832f..03e0aa06 100644 --- a/Firmware/fibre-cpp/include/fibre/libfibre.h +++ b/Firmware/fibre-cpp/include/fibre/libfibre.h @@ -40,7 +40,9 @@ #endif #ifdef FIBRE_COMPILE -# define FIBRE_PUBLIC DLL_EXPORT +# ifndef FIBRE_PUBLIC +# define FIBRE_PUBLIC DLL_EXPORT +# endif # define FIBRE_PRIVATE DLL_LOCAL #else # define FIBRE_PUBLIC DLL_IMPORT @@ -54,12 +56,15 @@ extern "C" { #endif struct LibFibreCtx; +struct LibFibreChannelDiscoveryCtx; struct LibFibreDiscoveryCtx; struct LibFibreCallContext; struct LibFibreObject; struct LibFibreInterface; struct LibFibreFunction; struct LibFibreAttribute; +struct LibFibreTxStream; +struct LibFibreRxStream; enum FibreStatus { kFibreOk, @@ -94,38 +99,98 @@ typedef int (*cancel_timer_cb_t)(struct EventLoopTimer* timer); * previous object. * The interface handle is valid until the last object that implements it * is destroyed. - * @param intf_name: The ASCII-encoded name of the interface. Can be NULL be for + * @param intf_name: The ASCII-encoded name of the interface. Can be NULL for * anonymous interfaces. If not NULL, it is only valid for the duration * of the callback and must not be freed by the application. */ typedef void (*construct_object_cb_t)(void* ctx, LibFibreObject* obj, LibFibreInterface* intf, const char* intf_name, size_t intf_name_length); - typedef void (*destroy_object_cb_t)(void* ctx, LibFibreObject* obj); -typedef void (*on_found_object_cb_t)(void*, LibFibreObject*); -typedef void (*on_stopped_cb_t)(void*, FibreStatus); -typedef void (*on_attribute_added_cb_t)(void*, LibFibreAttribute*, const char* name, size_t name_length, LibFibreInterface*, const char* intf_name, size_t intf_name_length); -typedef void (*on_attribute_removed_cb_t)(void*, LibFibreAttribute*); -typedef void (*on_function_added_cb_t)(void*, LibFibreFunction*, const char* name, size_t name_length, const char** input_names, const char** input_codecs, const char** output_names, const char** output_codecs); -typedef void (*on_function_removed_cb_t)(void*, LibFibreFunction*); /** - * @brief Completion callback type for libfibre_start_call(). + * @brief on_start_discovery callback type for libfibre_register_discoverer(). * - * @param ctx: The user data that was passed to libfibre_start_call(). - * @param status: The status of the function call. - * kFibreOk indicates successful completion of the function call. All - * other error codes make no guarantee whether the call was executed or - * not. - * kFibreClosed indicates that the underlying object was lost during the - * function call. The call may or may not have succeeded. - * @param rx_end: Points to the address after the last byte written to the - * output buffer. This pointer always points to a valid position in the - * buffer (or the end of the buffer), even if the call failed. However if - * the status is not kFibreOk then the pointer may not precisely indicate - * the received data range. + * For every channel pair that the application finds that matches the filter of + * this discoverer the application should call libfibre_add_channels(). + * + * @param discovery_handle: An opaque handle that libfibre will pass to the + * corresponding on_stop_discovery callback to stop the discovery. + * @param specs, specs_length: The specs string that specifies discoverer-specific + * filter parameters. */ -typedef void (*on_call_completed_cb_t)(void* ctx, FibreStatus status, uint8_t* rx_end); +typedef void (*on_start_discovery_cb_t)(void* ctx, LibFibreChannelDiscoveryCtx* discovery_ctx, const char* specs, size_t specs_length); +typedef void (*on_stop_discovery_cb_t)(void* ctx, LibFibreChannelDiscoveryCtx* discovery_ctx); +typedef void (*on_found_object_cb_t)(void*, LibFibreObject*); +typedef void (*on_stopped_cb_t)(void*, FibreStatus); + +typedef void (*on_attribute_added_cb_t)(void*, LibFibreAttribute*, const char* name, size_t name_length, LibFibreInterface*, const char* intf_name, size_t intf_name_length); +typedef void (*on_attribute_removed_cb_t)(void*, LibFibreAttribute*); + +/** + * @brief on_function_added callback type for libfibre_subscribe_to_interface(). + * + * @param ctx: The user data that was passed to libfibre_subscribe_to_interface(). + * @param func: A handle for the function. Remains valid until the corresponding + * call to on_function_removed(). + * @param name: The ASCII-encoded name of the function. + * @param name_length: Length in bytes of the name. + * @param input_names: A null-terminated list of null-terminated ASCII-encoded + * strings. Each string corresponds to the name of one input argument. + * The list and the string buffers are only valid for the duration of the + * callback. They must not be freed by the application. + * @param input_codecs: A null-terminated list of null-terminated ASCII-encoded + * strings. Each string names the codec of one input argument. + * The list and the string buffers are only valid for the duration of the + * callback. They must not be freed by the application. + * @param output_names: Analogous to input_names. + * @param output_codecs: Analogous to output names. + */ +typedef void (*on_function_added_cb_t)(void* ctx, LibFibreFunction* func, const char* name, size_t name_length, const char** input_names, const char** input_codecs, const char** output_names, const char** output_codecs); + +typedef void (*on_function_removed_cb_t)(void*, LibFibreFunction*); +typedef void (*on_call_completed_cb_t)(void*, FibreStatus); + +/** + * @brief TX completion callback type for libfibre_start_tx(). + * + * @param ctx: The user data that was passed to libfibre_start_tx(). + * @param tx_stream: The TX stream on which the TX operation completed. + * @param status: The status of the last TX operation. + * - kFibreOk: The indicated range of the TX buffer was successfully + * transmitted and the stream might accept more data. + * - kFibreClosed: The indicated range of the TX buffer was successfully + * transmitted and the stream will no longer accept any data. + * - Any other status: Successful transmission of the data cannot be + * guaranteed and no more data can be sent on this stream. + * @param tx_end: Points to the address after the last byte read from the + * TX buffer. This pointer always points to a valid position in the + * buffer (or the end of the buffer), even if the transmission failed. + * However if the status is something other than kFibreOk and + * kFibreClosed then the pointer may not precisely indicate the + * transmitted data range. + */ +typedef void (*on_tx_completed_cb_t)(void* ctx, LibFibreTxStream* tx_stream, FibreStatus status, const uint8_t* tx_end); + +/** + * @brief RX completion callback type for libfibre_start_rx(). + * + * @param ctx: The user data that was passed to libfibre_start_rx(). + * @param rx_stream: The RX stream on which the RX operation completed. + * @param status: The status of the last RX operation. + * - kFibreOk: The indicated range of the RX buffer was successfully + * filled with received data and the stream might emit more data. + * - kFibreClosed: The indicated range of the RX buffer was successfully + * filled with received data and the stream will emit no more data. + * - Any other status: Successful transmission of the data cannot be + * guaranteed and no more data can be sent on this stream. + * @param rx_end: Points to the address after the last byte written to the + * RX buffer. This pointer always points to a valid position in the + * buffer (or the end of the buffer), even if the reception failed. + * However if the status is something other than kFibreOk and + * kFibreClosed then the pointer may not precisely indicate the + * received data range. + */ +typedef void (*on_rx_completed_cb_t)(void* ctx, LibFibreRxStream* rx_stream, FibreStatus status, uint8_t* rx_end); /** * @brief Returns the version of the libfibre library. @@ -139,7 +204,7 @@ typedef void (*on_call_completed_cb_t)(void* ctx, FibreStatus status, uint8_t* r * Even if breaking changes are introduced, we promise to keep this function * backwards compatible. */ -const struct LibFibreVersion* libfibre_get_version(); +FIBRE_PUBLIC const struct LibFibreVersion* libfibre_get_version(); /** * @brief Opens and initializes a Fibre context. @@ -192,6 +257,24 @@ FIBRE_PUBLIC struct LibFibreCtx* libfibre_open( */ FIBRE_PUBLIC void libfibre_close(struct LibFibreCtx* ctx); +/** + * @brief Registers an external channel discovery provider. + * + * Libfibre starts and stops the discoverer on demand as a result of calls + * to libfibre_start_discovery() and libfibre_stop_discovery(). + * This can be used by applications to implement transport providers which are + * not supported natively in libfibre. + */ +FIBRE_PUBLIC void libfibre_register_discoverer(LibFibreCtx* ctx, const char* name, size_t name_length, on_start_discovery_cb_t on_start_discovery, on_stop_discovery_cb_t on_stop_discovery, void* cb_ctx); + +/** + * @brief Registers new TX and RX channels as part of an ongoing discovery + * operation. + * + * The channels can be closed with libfibre_close_tx() and libfibre_close_rx(). + */ +FIBRE_PUBLIC void libfibre_add_channels(LibFibreCtx* ctx, LibFibreChannelDiscoveryCtx* discovery_ctx, LibFibreRxStream** tx_channel, LibFibreTxStream** rx_channel, size_t mtu); + /** * @brief Starts looking for Fibre objects that match the specifications. * @@ -221,7 +304,8 @@ FIBRE_PUBLIC void libfibre_close(struct LibFibreCtx* ctx); * @param cb_ctx: Arbitrary user data passed to the callbacks. * @returns: An opaque handle which should be passed to libfibre_stop_discovery(). */ -FIBRE_PUBLIC void libfibre_start_discovery(LibFibreCtx* ctx, const char* specs, size_t specs_len, struct LibFibreDiscoveryCtx** handle, +FIBRE_PUBLIC void libfibre_start_discovery(LibFibreCtx* ctx, + const char* specs, size_t specs_len, LibFibreDiscoveryCtx** handle, on_found_object_cb_t on_found_object, on_stopped_cb_t on_stopped, void* cb_ctx); @@ -245,10 +329,7 @@ FIBRE_PUBLIC void libfibre_stop_discovery(LibFibreCtx* ctx, LibFibreDiscoveryCtx * @param interface: An interface handle that was obtained in the callback of * libfibre_start_discovery(). * @param on_attribute_added: Invoked when an attribute is added to the - * interface. The name and intf_name buffers are only valid for the - * duration of the callback and must not be freed by the application. - * The attribute handle remains valid until the corresponding call to - * on_attribute_removed(). + * interface. * @param on_attribute_removed: Invoked when an attribute is removed from the * interface, including when the interface is being torn down. This is * called exactly once for every call to on_attribute_added(). @@ -299,47 +380,143 @@ FIBRE_PUBLIC void libfibre_subscribe_to_interface(LibFibreInterface* interface, FIBRE_PUBLIC FibreStatus libfibre_get_attribute(LibFibreObject* parent_obj, LibFibreAttribute* attr, LibFibreObject** child_obj_ptr); /** - * @brief Starts the invokation of a function call. + * @brief Starts a remote procedure call. * - * Once the call has completed (whether successful, failed or cancelled), the - * provided callback will be invoked. - * Until then, the ongoing call can be aborted with libfibre_cancel_call(). + * This function returns a TX/RX pair of streams that can be used by the + * application to send inputs to the remote procedure and receive outputs from + * it. + * + * libfibre is not required to send anything on the underlying transport + * layer(s) until a TX operation is started on the call's tx_stream. This means + * that for functions with no input the application must start a zero-length + * operation for the call to be started. + * + * The operation must be considered in progress until on_completed is invoked. + * This usually happens directly after both the tx_stream and rx_stream are + * closed (or failed), either by the server or due to a call to + * libfibre_cancel_call(). * * @param obj: An object handle that was obtained in the callback of * libfibre_start_discovery() or from a call to libfibre_get_attribute(). * @param func: A function handle that was obtained in the on_function_added() * callback of libfibre_subscribe_to_interface(). - * @param input: The buffer that contains the input arguments. Must remain valid - * until on_completed() is called. - * @param output: The buffer where the output arguments will be written. Must - * remain valid until on_completed() is called. * @param handle: The variable being pointed to by this argument is set to a * handle that can be passed to libfibre_cancel_call() to cancel the * started call. If on_complete() is invoked directly during * libfibre_start_call() then the handle variable is not updated later - * than this invokation. - * @param on_completed: Called when the operation completes, whether successful - * or not. + * than that invokation. + * @param tx_stream: The variable being pointed to by this argument is set to + * a TX stream handle that can be used to send data on this call. + * This handle remains valid until the application calls + * libfibre_end_call(). + * @param rx_stream: The variable being pointed to by this argument is set to + * an RX stream handle that can be used to receive data on this call. + * This handle remains valid until the application calls + * libfibre_end_call(). + * @param on_completed: Called when the call completes, whether successful or + * not. + * @param cb_ctx: An opaque handle which will be passed to on_completed(). */ -FIBRE_PUBLIC void libfibre_start_call(LibFibreObject* obj, LibFibreFunction* func, const uint8_t *input, size_t input_length, uint8_t *output, size_t output_length, LibFibreCallContext** handle, on_call_completed_cb_t on_completed, void* ctx); +FIBRE_PUBLIC void libfibre_start_call(LibFibreObject* obj, LibFibreFunction* func, LibFibreCallContext** handle, LibFibreTxStream** tx_stream, LibFibreRxStream** rx_stream, on_call_completed_cb_t on_completed, void* cb_ctx); /** - * @brief Cancels an ongoing function call. + * @brief Ends an ongoing function call. * - * This must not be called twice for the same call and must not be called once - * the completion callback of the function call is invoked. + * Note that this does not request semantic cancellation (or reversal) of + * actions triggered by this call. * - * The completion callback associated with this call will still be invoked after - * the call is cancelled. Until then, the call must still be considered in - * progress. - * - * After calling this function, the function call that was cancelled may or may - * not still go into effect. + * The application must still wait for the on_complete callback to be called + * before the call can be considered finished. + * In the meantime if the TX and/or RX stream of this call is still open then + * it is closed. + * libfibre_end_call must not be called twice for the same call. + * The completion callback may be called with kFibreCancelled or any other + * status. * * @param handle: The function call handle that was obtained by a call to - * libfibre_cancel_call(). + * libfibre_start_call(). */ -FIBRE_PUBLIC void libfibre_cancel_call(LibFibreCallContext* handle); +FIBRE_PUBLIC void libfibre_end_call(LibFibreCallContext* handle); + +/** + * @brief Starts sending data on the specified TX stream. + * + * The TX operation must be considered in progress until the on_completed + * callback is called. Until then the application must not start another TX + * operation on the same stream. In the meantime the application can call + * libfibre_cancel_tx() at any time to abort the operation. + * + * @param tx_stream: The stream on which to send data. + * @param tx_buf: The buffer to transmit. Must remain valid until the operation + * completes. + * @param tx_len: Length of tx_buf. + * @param on_completed: Called when the operation completes, whether successful + * or not. + * @param ctx: Arbitrary user data passed to the on_completed callback. + */ +FIBRE_PUBLIC void libfibre_start_tx(LibFibreTxStream* tx_stream, const uint8_t* tx_buf, size_t tx_len, on_tx_completed_cb_t on_completed, void* ctx); + +/** + * @brief Cancels an ongoing TX operation. + * + * Must only be called if there is actually a TX operation in progress for which + * cancellation has not yet been requested. + * The application must still wait for the on_complete callback to be called + * before the operation can be considered finished. The completion callback may + * be called with kFibreCancelled or any other status. + * + * TODO: specify if streams can be restarted (current doc of on_tx_completed_cb_t implies no) + * + * @param tx_stream: The TX stream on which to cancel the ongoing TX operation. + */ +FIBRE_PUBLIC void libfibre_cancel_tx(LibFibreTxStream* tx_stream); + +/** + * @brief Permanently close TX stream. + * + * Must not be called while a transfer is ongoing. + */ +FIBRE_PUBLIC void libfibre_close_tx(LibFibreTxStream* tx_stream, FibreStatus status); + +/** + * @brief Starts receiving data on the specified RX stream. + * + * The RX operation must be considered in progress until the on_completed + * callback is called. Until then the application must not start another RX + * operation on the same stream. In the meantime the application can call + * libfibre_cancel_rx() at any time to abort the operation. + * + * @param rx_stream: The stream on which to receive data. + * @param rx_buf: The buffer to receive to. Must remain valid until the + * operation completes. + * @param rx_len: Length of rx_buf. + * @param on_completed: Called when the operation completes, whether successful + * or not. + * @param ctx: Arbitrary user data passed to the on_completed callback. + */ +FIBRE_PUBLIC void libfibre_start_rx(LibFibreRxStream* rx_stream, uint8_t* rx_buf, size_t rx_len, on_rx_completed_cb_t on_completed, void* ctx); + +/** + * @brief Cancels an ongoing RX operation. + * + * Must only be called if there is actually a RX operation in progress for which + * cancellation has not yet been requested. + * The application must still wait for the on_complete callback to be called + * before the operation can be considered finished. The completion callback may + * be called with kFibreCancelled or any other status. + * + * TODO: specify if streams can be restarted (current doc of on_rx_completed_cb_t implies no) + * + * @param rx_stream: The RX stream on which to cancel the ongoing RX operation. + */ +FIBRE_PUBLIC void libfibre_cancel_rx(LibFibreRxStream* rx_stream); + +/** + * @brief Permanently close RX stream. + * + * Must not be called while a transfer is ongoing. + */ +FIBRE_PUBLIC void libfibre_close_rx(LibFibreRxStream* rx_stream, FibreStatus status); #ifdef __cplusplus } diff --git a/Firmware/fibre-cpp/legacy_object_client.cpp b/Firmware/fibre-cpp/legacy_object_client.cpp index abc1262d..321997d5 100644 --- a/Firmware/fibre-cpp/legacy_object_client.cpp +++ b/Firmware/fibre-cpp/legacy_object_client.cpp @@ -177,8 +177,8 @@ std::unordered_map codecs = { {"uint16", 2}, {"int32", 4}, {"uint32", 4}, - {"int64", 6}, - {"uint64", 6}, + {"int64", 8}, + {"uint64", 8}, {"float", 4}, {"endpoint_ref", 4} }; @@ -226,37 +226,24 @@ void LegacyObjectClient::start(Completer& completer) { +void LegacyObjectClient::start_call(size_t ep_num, LegacyFibreFunction* func, CallContext** handle, Completer& completer) { CallContext* call = new CallContext(); call->ep_num = ep_num; - call->tx_buf = input; - call->rx_buf = output; call->func = func; - call->completer = &completer; - - if (op_handle_) { - FIBRE_LOG(D) << "Call in progress. Enqueuing this call."; - // An operation is already in progress. Enqueue this one. - pending_calls_.push_back(call); - } else { - // No endpoint operation is in progress. Start this call immediately - FIBRE_LOG(D) << "No call in progress. Starting call now."; - call_ = call; - complete({kStreamOk, nullptr}); + call->protocol_ = protocol_; + call->completer_ = &completer; + + if (handle) { + *handle = call; } } void LegacyObjectClient::cancel_call(CallContext* handle) { - if (call_ == handle) { - protocol_->cancel_endpoint_operation(op_handle_); + if (handle->op_handle_) { + handle->protocol_->cancel_endpoint_operation(op_handle_); + handle->cancelling_ = true; } else { - auto it = std::find(pending_calls_.begin(), pending_calls_.end(), handle); - if (it != pending_calls_.end()) { - CallContext* call = *it; - pending_calls_.erase(it); - safe_complete(call->completer, {kFibreCancelled, call->rx_buf.end()}); - delete call; - } + handle->complete_call(kFibreCancelled); } } @@ -366,118 +353,197 @@ void LegacyObjectClient::receive_more_json() { } void LegacyObjectClient::complete(EndpointOperationResult result) { + // The JSON read operation completed + op_handle_ = 0; if (result.status == kStreamCancelled) { - if (call_) { - auto call = call_; - call_ = nullptr; - safe_complete(call->completer, {kFibreCancelled, call->rx_buf.end()}); - delete call; - } return; } else if (result.status == kStreamClosed) { - if (call_) { - auto call = call_; - call_ = nullptr; - safe_complete(call->completer, {kFibreClosed, call->rx_buf.end()}); - delete call; - } return; } else if (result.status != kStreamOk) { - FIBRE_LOG(W) << "endpoint operation failed"; // TODO: add retry logic - if (call_) { - auto call = call_; - call_ = nullptr; - safe_complete(call->completer, {kFibreInternalError, call->rx_buf.end()}); - delete call; - } + FIBRE_LOG(W) << "JSON read operation failed"; // TODO: add retry logic return; } - if (call_) { - // The endpoint operation that completed belongs to the active call + size_t n_received = result.rx_end - json_.data() - json_.size() + 1024; + json_.resize(json_.size() - 1024 + n_received); - LegacyFibreFunction* func = call_->func; + if (n_received) { + receive_more_json(); - if (call_->ep_num && !call_->progress) { - // Read/write/exchange property - FIBRE_LOG(D) << "starting property transaction on " << call_->ep_num << " with tx buf len " << call_->tx_buf.size() << " and rx len " << call_->rx_buf.size(); - call_->progress++; - protocol_->start_endpoint_operation(call_->ep_num, call_->tx_buf, call_->rx_buf, &op_handle_, *this); + } else { - } else if (!call_->ep_num && call_->progress < func->inputs.size()) { - // Write input arg - size_t argnum = call_->progress; - call_->progress++; - cbufptr_t current_buf = call_->tx_buf.take(func->inputs[argnum].size); - call_->tx_buf = call_->tx_buf.skip(func->inputs[argnum].size); - protocol_->start_endpoint_operation(func->inputs[argnum].ep_num, current_buf, call_->rx_buf.take(0), &op_handle_, *this); + FIBRE_LOG(D) << "received JSON of length " << json_.size(); + //FIBRE_LOG(D) << "JSON: " << str{json_.data(), json_.data() + json_.size()}; - } else if (!call_->ep_num && call_->progress == func->inputs.size()) { - // Trigger - // call_->tx_buf should be empty by now - call_->progress++; - protocol_->start_endpoint_operation(func->ep_num, call_->tx_buf, call_->rx_buf.take(0), &op_handle_, *this); + const char *begin = reinterpret_cast(json_.data()); + auto val = json_parse(&begin, begin + json_.size()); - } else if (!call_->ep_num && call_->progress < func->inputs.size() + 1 + func->outputs.size()) { - // Read output arg - size_t argnum = call_->progress - func->inputs.size() - 1; - call_->progress++; - bufptr_t current_buf = call_->rx_buf.take(func->outputs[argnum].size); - call_->rx_buf = call_->rx_buf.skip(func->outputs[argnum].size); - protocol_->start_endpoint_operation(func->outputs[argnum].ep_num, call_->tx_buf, current_buf, &op_handle_, *this); + if (json_is_err(val)) { + size_t pos = json_as_err(val).ptr - reinterpret_cast(json_.data()); + FIBRE_LOG(E) << "JSON parsing error: " << json_as_err(val).str << " at position " << pos; + return; + } else if (!json_is_list(val)) { + FIBRE_LOG(E) << "JSON data must be a list"; + return; + } + FIBRE_LOG(D) << "sucessfully parsed JSON"; + root_obj_ = load_object(val); + json_crc_ = calc_crc16(PROTOCOL_VERSION, json_.data(), json_.size()); + if (root_obj_) { + safe_complete(on_found_root_object_, this, root_obj_); + } + } +} + +void LegacyObjectClient::CallContext::start_write(cbufptr_t buffer, TransferHandle* handle, Completer& completer) { + if (tx_completer_) { + FIBRE_LOG(W) << "TX operation already in progress"; + completer.complete({kStreamError, buffer.begin()}); + return; + } + + tx_completer_ = &completer; + + if (handle) { + *handle = reinterpret_cast(this); + } + + if (ep_num) { + // Single-endpoint function + if (progress == 0) { + protocol_->start_endpoint_operation(ep_num, buffer, rx_buf_, &op_handle_, *this); } else { - CallContext* call = call_; - call_ = nullptr; - FIBRE_LOG(D) << "call completed!"; - safe_complete(call->completer, {kFibreOk, call->rx_buf.end()}); - delete call; + safe_complete(tx_completer_, {kStreamClosed, buffer.begin()}); } } else { - // The endpoint operation that completed belongs to the JSON fetch process - - size_t n_received = result.rx_end - json_.data() - json_.size() + 1024; - json_.resize(json_.size() - 1024 + n_received); - - if (n_received) { - receive_more_json(); - return; - + // Multi-endpoint function (deprecated) + if (progress < func->inputs.size()) { + // Write input arg + size_t argnum = progress; + if (buffer.size() < func->inputs[argnum].size) { + FIBRE_LOG(W) << "TX granularity too small"; + safe_complete(tx_completer_, {kStreamError, buffer.begin()}); + } else { + protocol_->start_endpoint_operation(func->inputs[argnum].ep_num, + buffer.take(func->inputs[argnum].size), {}, &op_handle_, *this); + } + } else if (progress == func->inputs.size()) { + // Trigger function + protocol_->start_endpoint_operation(func->ep_num, buffer.take(0), {}, &op_handle_, *this); } else { - - FIBRE_LOG(D) << "received JSON of length " << json_.size(); - //FIBRE_LOG(D) << "JSON: " << str{json_.data(), json_.data() + json_.size()}; - - const char *begin = reinterpret_cast(json_.data()); - auto val = json_parse(&begin, begin + json_.size()); - - if (json_is_err(val)) { - size_t pos = json_as_err(val).ptr - reinterpret_cast(json_.data()); - FIBRE_LOG(E) << "JSON parsing error: " << json_as_err(val).str << " at position " << pos; - return; - } else if (!json_is_list(val)) { - FIBRE_LOG(E) << "JSON data must be a list"; - return; - } - - FIBRE_LOG(D) << "sucessfully parsed JSON"; - root_obj_ = load_object(val); - json_crc_ = calc_crc16(PROTOCOL_VERSION, json_.data(), json_.size()); - if (root_obj_) { - safe_complete(on_found_root_object_, this, root_obj_); - } + safe_complete(tx_completer_, {kStreamClosed, buffer.begin()}); } } +} - // Start next call in the queue if any - // It's possible that the next function call was already started on one of - // the callbacks above. - if (!call_ && pending_calls_.size()) { - call_ = pending_calls_[0]; - pending_calls_.erase(pending_calls_.begin()); - complete({kStreamOk, nullptr}); +void LegacyObjectClient::CallContext::cancel_write(TransferHandle transfer_handle) { + // not implemented +} + +void LegacyObjectClient::CallContext::start_read(bufptr_t buffer, TransferHandle* handle, Completer& completer) { + if (rx_completer_) { + FIBRE_LOG(W) << "RX operation already in progress"; + completer.complete({kStreamError, buffer.begin()}); + return; } -} \ No newline at end of file + + rx_completer_ = &completer; + + if (handle) { + *handle = reinterpret_cast(this); + } + + if (ep_num) { + // Single-endpoint function + if (progress == 0 && !op_handle_) { + // Transfer has not yet started. Prepare RX buffer for when it starts. + rx_buf_ = buffer; + } else { + // Transfer has already started or completed. Cannot start RX anymore. + safe_complete(rx_completer_, {kStreamClosed, buffer.begin()}); + } + + } else { + // Multi-endpoint function (deprecated) + if (progress <= func->inputs.size()) { + // Not yet in the receive phase. Store the RX pointer for later use. + rx_buf_ = buffer; + } else if (progress < func->inputs.size() + 1 + func->outputs.size()) { + // Read output arg + size_t argnum = progress - func->inputs.size() - 1; + if (buffer.size() < func->outputs[argnum].size) { + FIBRE_LOG(W) << "RX granularity too small"; + safe_complete(rx_completer_, {kStreamError, buffer.begin()}); + } else { + protocol_->start_endpoint_operation(func->inputs[argnum].ep_num, + {}, buffer.take(func->outputs[argnum].size), &op_handle_, *this); + } + } else { + safe_complete(rx_completer_, {kStreamClosed, buffer.begin()}); + } + } +} + +void LegacyObjectClient::CallContext::cancel_read(TransferHandle transfer_handle) { + // not implemented +} + +void LegacyObjectClient::CallContext::complete(EndpointOperationResult result) { + if (result.status == kStreamCancelled || (result.status == kStreamOk && cancelling_)) { + safe_complete(tx_completer_, {kStreamCancelled, result.tx_end}); + safe_complete(rx_completer_, {kStreamCancelled, result.rx_end}); + complete_call(kFibreCancelled); + return; + } else if (result.status == kStreamClosed) { + safe_complete(tx_completer_, {kStreamClosed, result.tx_end}); + safe_complete(rx_completer_, {kStreamClosed, result.rx_end}); + complete_call(kFibreClosed); + return; + } else if (result.status != kStreamOk) { + FIBRE_LOG(W) << "endpoint operation failed"; // TODO: add retry logic + safe_complete(tx_completer_, {kStreamError, result.tx_end}); + safe_complete(rx_completer_, {kStreamError, result.rx_end}); + complete_call(kFibreInternalError); + return; + } + + progress++; + + if (ep_num) { + // Single-endpoint function + if (progress == 1) { + safe_complete(tx_completer_, {kStreamClosed, result.tx_end}); + safe_complete(rx_completer_, {kStreamClosed, result.rx_end}); + complete_call(kFibreOk); + } + + } else { + // Multi-endpoint function (deprecated) + + if (progress < func->inputs.size() + 1) { + safe_complete(tx_completer_, {kStreamOk, result.tx_end}); + } else if (progress == func->inputs.size() + 1) { + safe_complete(tx_completer_, {kStreamClosed, result.tx_end}); + // If the application already prepared an RX operation complete this + // operation with length 0 to make the application restart this + // operation. + safe_complete(rx_completer_, {kStreamOk, rx_buf_.begin()}); + } else if (progress < func->inputs.size() + 1 + func->outputs.size()) { + safe_complete(rx_completer_, {kStreamOk, result.rx_end}); + } else if (progress == func->inputs.size() + 1 + func->outputs.size()) { + safe_complete(rx_completer_, {kStreamClosed, result.rx_end}); + complete_call(kFibreOk); + } else { + FIBRE_LOG(W) << "progress is further than expected"; + } + } +} + +void LegacyObjectClient::CallContext::complete_call(FibreStatus result) { + safe_complete(completer_, result); + delete this; +} diff --git a/Firmware/fibre-cpp/legacy_object_client.hpp b/Firmware/fibre-cpp/legacy_object_client.hpp index 1f2e8453..b6397672 100644 --- a/Firmware/fibre-cpp/legacy_object_client.hpp +++ b/Firmware/fibre-cpp/legacy_object_client.hpp @@ -15,6 +15,7 @@ namespace fibre { struct EndpointOperationResult { StreamStatus status; + const uint8_t* tx_end; uint8_t* rx_end; }; @@ -58,24 +59,32 @@ struct LegacyObject { class LegacyObjectClient : Completer { public: - struct CallResult { - FibreStatus status; - uint8_t* end; - }; - struct CallContext { + struct CallContext : AsyncStreamSink, AsyncStreamSource, Completer { size_t progress = 0; - size_t ep_num; - cbufptr_t tx_buf; - bufptr_t rx_buf; - LegacyFibreFunction* func; - Completer* completer; + size_t ep_num = 0; + bufptr_t rx_buf_ = {}; + LegacyFibreFunction* func = nullptr; + Completer* tx_completer_ = nullptr; + Completer* rx_completer_ = nullptr; + Completer* completer_ = nullptr; + EndpointOperationHandle op_handle_ = 0; + LegacyProtocolPacketBased* protocol_ = nullptr; + bool cancelling_ = false; + + void start_write(cbufptr_t buffer, TransferHandle* handle, Completer& completer) final; + void cancel_write(TransferHandle transfer_handle) final; + void start_read(bufptr_t buffer, TransferHandle* handle, Completer& completer) final; + void cancel_read(TransferHandle transfer_handle) final; + + void complete(EndpointOperationResult result); + void complete_call(FibreStatus result); }; LegacyObjectClient(LegacyProtocolPacketBased* protocol) : protocol_(protocol) {} void start(Completer>& on_found_root_object, Completer& on_lost_root_object); - void start_call(size_t ep_num, LegacyFibreFunction* func, cbufptr_t input, bufptr_t output, CallContext** handle, Completer& completer); + void start_call(size_t ep_num, LegacyFibreFunction* func, CallContext** handle, Completer& completer); void cancel_call(CallContext* handle); // For direct access by LegacyProtocolPacketBased and libfibre.cpp @@ -96,7 +105,6 @@ private: uint8_t tx_buf_[4] = {0xff, 0xff, 0xff, 0xff}; EndpointOperationHandle op_handle_ = 0; std::vector json_; - CallContext* call_ = nullptr; // active call std::vector pending_calls_; std::unordered_map> rw_property_interfaces; std::unordered_map> ro_property_interfaces; diff --git a/Firmware/fibre-cpp/legacy_protocol.cpp b/Firmware/fibre-cpp/legacy_protocol.cpp index 359cf888..e3aeca4e 100644 --- a/Firmware/fibre-cpp/legacy_protocol.cpp +++ b/Firmware/fibre-cpp/legacy_protocol.cpp @@ -201,16 +201,6 @@ void PacketUnwrapper::complete(ReadResult result) { * completer. */ void LegacyProtocolPacketBased::start_endpoint_operation(uint16_t endpoint_id, cbufptr_t tx_buf, bufptr_t rx_buf, EndpointOperationHandle* handle, Completer& completer) { - if (tx_buf.size() + 8 >= tx_mtu_) { - FIBRE_LOG(E) << "packet too large"; - completer.complete({kStreamError, rx_buf.begin()}); - } - - if (rx_buf.size() > 0xffff) { - FIBRE_LOG(E) << "receive size larger than 65535 currently not supported"; - completer.complete({kStreamError, rx_buf.begin()}); - } - outbound_seq_no_ = ((outbound_seq_no_ + 1) & 0x7fff); EndpointOperation op = { @@ -233,7 +223,7 @@ void LegacyProtocolPacketBased::start_endpoint_operation(uint16_t endpoint_id, c // Previous endpoint operation was not yet sent. We don't support // enqueuing multiple endpoint operations while the first didn't send yet. FIBRE_LOG(E) << "previous endpoint operation still not sent"; - completer.complete({kStreamError, rx_buf.begin()}); + completer.complete({kStreamError, tx_buf.begin(), rx_buf.begin()}); } else { // Control is returned to start_endpoint_operation once TX completes pending_operation_ = op; @@ -249,16 +239,19 @@ void LegacyProtocolPacketBased::start_endpoint_operation(EndpointOperation op) { write_le(op.endpoint_id | 0x8000, tx_buf_ + 2); write_le(op.rx_buf.size(), tx_buf_ + 4); - memcpy(tx_buf_ + 6, op.tx_buf.begin(), op.tx_buf.size()); + size_t mtu = std::min(sizeof(tx_buf_), tx_mtu_); + size_t n_payload = std::min(std::max(mtu, (size_t)8) - 8, op.tx_buf.size()); + + memcpy(tx_buf_ + 6, op.tx_buf.begin(), n_payload); uint16_t trailer = (op.endpoint_id & 0x7fff) == 0 ? PROTOCOL_VERSION : client_.json_crc_; - write_le(trailer, tx_buf_ + 6 + op.tx_buf.size()); + write_le(trailer, tx_buf_ + 6 + n_payload); expected_acks_[op.seqno] = op; transmitting_op_ = op.seqno | 0xffff0000; - tx_channel_->start_write(cbufptr_t{tx_buf_}.take(8 + op.tx_buf.size()), &tx_handle_, *static_cast(this)); + tx_channel_->start_write(cbufptr_t{tx_buf_}.take(8 + n_payload), &tx_handle_, *static_cast(this)); } @@ -270,10 +263,12 @@ void LegacyProtocolPacketBased::cancel_endpoint_operation(EndpointOperationHandl uint16_t seqno = static_cast(handle & 0xffff); Completer* completer; + const uint8_t* tx_end = nullptr; uint8_t* rx_end = nullptr; if (pending_operation_.seqno == seqno) { completer = pending_operation_.completer; + tx_end = pending_operation_.tx_buf.begin(); rx_end = pending_operation_.rx_buf.begin(); pending_operation_ = {}; } @@ -282,6 +277,7 @@ void LegacyProtocolPacketBased::cancel_endpoint_operation(EndpointOperationHandl if (it != expected_acks_.end()) { completer = it->second.completer; + tx_end = it->second.tx_buf.begin(); rx_end = it->second.rx_buf.begin(); expected_acks_.erase(it); } @@ -293,7 +289,7 @@ void LegacyProtocolPacketBased::cancel_endpoint_operation(EndpointOperationHandl } else { // Either we're waiting for an ack on this operation or it has not yet // been sent. In both cases we can just complete immediately. - safe_complete(completer, {kStreamCancelled, rx_end}); + safe_complete(completer, {kStreamCancelled, tx_end, rx_end}); } } @@ -339,14 +335,18 @@ void LegacyProtocolPacketBased::on_write_finished(WriteResult result) { uint16_t seqno = transmitting_op_ & 0xffff; transmitting_op_ = 0; + auto it = expected_acks_.find(seqno); + size_t n_sent = std::max((size_t)(result.end - tx_buf_), (size_t)8) - 8; + it->second.tx_buf = it->second.tx_buf.skip(n_sent); + // If the TX task was a remote endpoint operation but didn't succeed // we terminate that operation if (result.status != kStreamOk) { - auto it = expected_acks_.find(seqno); auto completer = it->second.completer; + auto tx_end = it->second.tx_buf.begin(); auto rx_end = it->second.rx_buf.begin(); expected_acks_.erase(it); - safe_complete(completer, {result.status, rx_end}); + safe_complete(completer, {result.status, result.end, rx_end}); } } #endif @@ -415,10 +415,11 @@ void LegacyProtocolPacketBased::on_read_finished(ReadResult result) { } else { size_t n_copy = std::min((size_t)(result.end - rx_buf.begin()), it->second.rx_buf.size()); memcpy(it->second.rx_buf.begin(), rx_buf.begin(), n_copy); + const uint8_t* tx_end = it->second.tx_buf.begin(); uint8_t* rx_end = it->second.rx_buf.begin() + n_copy; auto completer = it->second.completer; expected_acks_.erase(it); - safe_complete(completer, {kStreamOk, rx_end}); + safe_complete(completer, {kStreamOk, tx_end, rx_end}); } #else @@ -506,14 +507,14 @@ void LegacyProtocolPacketBased::on_rx_tx_closed(StreamStatus status) { #ifdef FIBRE_ENABLE_CLIENT // Cancel pending endpoint operation if (pending_operation_.completer) { - pending_operation_.completer->complete({status, pending_operation_.rx_buf.begin()}); + pending_operation_.completer->complete({status, pending_operation_.tx_buf.begin(), pending_operation_.rx_buf.begin()}); pending_operation_ = {}; } // Cancel all ongoing endpoint operations for (auto& item: expected_acks_) { if (item.second.completer) { - (*item.second.completer).complete({status, item.second.rx_buf.begin()}); + (*item.second.completer).complete({status, item.second.tx_buf.begin(), item.second.rx_buf.begin()}); } } expected_acks_.clear(); diff --git a/Firmware/fibre-cpp/libfibre.cpp b/Firmware/fibre-cpp/libfibre.cpp index 3e88172b..2c98d945 100644 --- a/Firmware/fibre-cpp/libfibre.cpp +++ b/Firmware/fibre-cpp/libfibre.cpp @@ -1,15 +1,19 @@ #include -#include "platform_support/libusb_transport.hpp" #include "logging.hpp" #include "print_utils.hpp" #include "legacy_protocol.hpp" #include "legacy_object_client.hpp" -#include "stdio.h" // TODO: remove +#include "event_loop.hpp" +#include "channel_discoverer.hpp" #include "string.h" #include #include "fibre/simple_serdes.hpp" +#ifdef FIBRE_ENABLE_LIBUSB +#include "platform_support/libusb_transport.hpp" +#endif + DEFINE_LOG_TOPIC(LIBFIBRE); USE_LOG_TOPIC(LIBFIBRE); @@ -56,13 +60,95 @@ private: cancel_timer_cb_t cancel_timer_; }; +namespace fibre { + +class AsyncStreamLink : public AsyncStreamSink, public AsyncStreamSource { +public: + void start_write(cbufptr_t buffer, TransferHandle* handle, Completer& completer) final; + void cancel_write(TransferHandle transfer_handle) final; + void start_read(bufptr_t buffer, TransferHandle* handle, Completer& completer) final; + void cancel_read(TransferHandle transfer_handle) final; + void close(StreamStatus status); + + Completer* read_completer_ = nullptr; + bufptr_t read_buf_; + Completer* write_completer_ = nullptr; + cbufptr_t write_buf_; +}; + +void AsyncStreamLink::start_write(cbufptr_t buffer, TransferHandle* handle, Completer& completer) { + if (read_completer_) { + size_t n_copy = std::min(read_buf_.size(), buffer.size()); + memcpy(read_buf_.begin(), buffer.begin(), n_copy); + safe_complete(read_completer_, {kStreamOk, read_buf_.begin() + n_copy}); + completer.complete({kStreamOk, buffer.begin() + n_copy}); + } else { + if (handle) { + *handle = reinterpret_cast(this); + } + write_buf_ = buffer; + write_completer_ = &completer; + } +} + +void AsyncStreamLink::cancel_write(TransferHandle transfer_handle) { + safe_complete(write_completer_, {kStreamCancelled, write_buf_.begin()}); +} + +void AsyncStreamLink::start_read(bufptr_t buffer, TransferHandle* handle, Completer& completer) { + if (write_completer_) { + FIBRE_LOG(W) << "start_read: completing writer"; + size_t n_copy = std::min(buffer.size(), write_buf_.size()); + memcpy(buffer.begin(), write_buf_.begin(), n_copy); + safe_complete(write_completer_, {kStreamOk, write_buf_.begin() + n_copy}); + completer.complete({kStreamOk, buffer.begin() + n_copy}); + } else { + //FIBRE_LOG(W) << "start_read: waiting for writer"; + if (handle) { + *handle = reinterpret_cast(this); + } + read_buf_ = buffer; + read_completer_ = &completer; + } +} + +void AsyncStreamLink::cancel_read(TransferHandle transfer_handle) { + safe_complete(read_completer_, {kStreamCancelled, read_buf_.begin()}); +} + +void AsyncStreamLink::close(StreamStatus status) { + safe_complete(write_completer_, {status, write_buf_.begin()}); + safe_complete(read_completer_, {status, read_buf_.begin()}); +} + +} + +FibreStatus convert_status(fibre::StreamStatus status) { + switch (status) { + case fibre::kStreamOk: return kFibreOk; + case fibre::kStreamCancelled: return kFibreCancelled; + case fibre::kStreamClosed: return kFibreClosed; + default: return kFibreInternalError; // TODO: this may not always be appropriate + } +} + +fibre::StreamStatus convert_status(FibreStatus status) { + switch (status) { + case kFibreOk: return fibre::kStreamOk; + case kFibreCancelled: return fibre::kStreamCancelled; + case kFibreClosed: return fibre::kStreamClosed; + default: return fibre::kStreamError; // TODO: this may not always be appropriate + } +} + struct FIBRE_PRIVATE LibFibreCtx { ExternalEventLoop* event_loop; construct_object_cb_t on_construct_object; destroy_object_cb_t on_destroy_object; void* cb_ctx; - fibre::LibusbDiscoverer libusb_discoverer; size_t n_discoveries = 0; + + std::unordered_map> discoverers; }; struct FIBRE_PRIVATE LibFibreDiscoveryCtx : @@ -76,7 +162,8 @@ struct FIBRE_PRIVATE LibFibreDiscoveryCtx : void complete(fibre::LegacyObjectClient* obj_client) final; void complete(fibre::LegacyProtocolPacketBased* protocol, fibre::StreamStatus status) final; - fibre::LibusbDiscoverer::ChannelDiscoveryContext* libusb_discovery_ctx = nullptr; + std::unordered_map context_handles; + on_found_object_cb_t on_found_object; void* cb_ctx; LibFibreCtx* ctx; @@ -87,6 +174,35 @@ struct FIBRE_PRIVATE LibFibreDiscoveryCtx : size_t use_count = 1; }; +struct LibFibreTxStream : fibre::Completer { + void complete(fibre::WriteResult result) { + if (on_completed) { + (*on_completed)(ctx, this, convert_status(result.status), result.end); + } + } + + fibre::AsyncStreamSink* sink; + fibre::TransferHandle handle; + on_tx_completed_cb_t on_completed; + void* ctx; + void (*on_closed)(LibFibreTxStream*, void*, fibre::StreamStatus); + void* on_closed_ctx; +}; + +struct LibFibreRxStream : fibre::Completer { + void complete(fibre::ReadResult result) { + if (on_completed) { + (*on_completed)(ctx, this, convert_status(result.status), result.end); + } + } + + fibre::AsyncStreamSource* source; + fibre::TransferHandle handle; + on_rx_completed_cb_t on_completed; + void* ctx; + void (*on_closed)(LibFibreRxStream*, void*, fibre::StreamStatus); + void* on_closed_ctx; +}; // Callback for start_channel_discovery() @@ -103,11 +219,9 @@ void LibFibreDiscoveryCtx::complete(fibre::ChannelDiscoveryResult result) { return; } - const size_t mtu = 64; // TODO: get MTU from channel specific data - use_count++; - auto protocol = new fibre::LegacyProtocolPacketBased(result.rx_channel, result.tx_channel, mtu); + auto protocol = new fibre::LegacyProtocolPacketBased(result.rx_channel, result.tx_channel, result.mtu); protocol->client_.user_data_ = ctx; protocol->start(*this, *this, *this); } @@ -177,22 +291,26 @@ LibFibreCtx* libfibre_open( destroy_object_cb_t destroy_object, void* cb_ctx) { - if (!register_event || !deregister_event) { - FIBRE_LOG(E) << "invalid argument"; - return nullptr; - } - + //if (!register_event || !deregister_event) { + // FIBRE_LOG(E) << "invalid argument"; + // return nullptr; + //} + FIBRE_LOG(D) << "object constructor: " << reinterpret_cast(construct_object); LibFibreCtx* ctx = new LibFibreCtx(); ctx->event_loop = new ExternalEventLoop(post, register_event, deregister_event, call_later, cancel_timer); ctx->on_construct_object = construct_object; ctx->on_destroy_object = destroy_object; ctx->cb_ctx = cb_ctx; - if (ctx->libusb_discoverer.init(ctx->event_loop) != 0) { +#ifdef FIBRE_ENABLE_LIBUSB + auto libusb_discoverer = std::make_shared(); + if (libusb_discoverer->init(ctx->event_loop) != 0) { delete ctx; FIBRE_LOG(E) << "failed to init libusb transport layer"; return nullptr; } + ctx->discoverers["usb"] = libusb_discoverer; +#endif FIBRE_LOG(D) << "opened (" << fibre::as_hex((uintptr_t)ctx) << ")"; return ctx; @@ -203,13 +321,99 @@ void libfibre_close(LibFibreCtx* ctx) { FIBRE_LOG(W) << "there are still discovery processes ongoing"; } - ctx->libusb_discoverer.deinit(); + ctx->discoverers.clear(); + delete ctx->event_loop; delete ctx; FIBRE_LOG(D) << "closed (" << fibre::as_hex((uintptr_t)ctx) << ")"; } +struct LibFibreChannelDiscoveryCtx : fibre::ChannelDiscoveryContext { + fibre::Completer* completer; +}; + +class ExternalDiscoverer : public fibre::ChannelDiscoverer { + void start_channel_discovery( + const char* specs, size_t specs_len, + fibre::ChannelDiscoveryContext** handle, + fibre::Completer& on_found_channels) final; + int stop_channel_discovery(fibre::ChannelDiscoveryContext* handle) final; +public: + on_start_discovery_cb_t on_start_discovery; + on_stop_discovery_cb_t on_stop_discovery; + void* cb_ctx; +}; + +void ExternalDiscoverer::start_channel_discovery(const char* specs, size_t specs_len, fibre::ChannelDiscoveryContext** handle, fibre::Completer& on_found_channels) { + LibFibreChannelDiscoveryCtx* ctx = new LibFibreChannelDiscoveryCtx{}; + ctx->completer = &on_found_channels; + if (handle) { + *handle = ctx; + } + if (on_start_discovery) { + (*on_start_discovery)(cb_ctx, ctx, specs, specs_len); + } +} + +int ExternalDiscoverer::stop_channel_discovery(fibre::ChannelDiscoveryContext* handle) { + LibFibreChannelDiscoveryCtx* ctx = static_cast(handle); + if (on_stop_discovery) { + (*on_stop_discovery)(cb_ctx, ctx); + } + delete ctx; + return 0; +} + +void libfibre_register_discoverer(LibFibreCtx* ctx, const char* name, size_t name_length, on_start_discovery_cb_t on_start_discovery, on_stop_discovery_cb_t on_stop_discovery, void* cb_ctx) { + std::string name_str = {name, name + name_length}; + if (ctx->discoverers.find(name_str) != ctx->discoverers.end()) { + FIBRE_LOG(W) << "Discoverer " << name_str << " already registered"; + return; // TODO: report status + } + + auto disc = std::make_shared(); + disc->on_start_discovery = on_start_discovery; + disc->on_stop_discovery = on_stop_discovery; + disc->cb_ctx = cb_ctx; + ctx->discoverers[name_str] = disc; +} + +void libfibre_add_channels(LibFibreCtx* ctx, LibFibreChannelDiscoveryCtx* discovery_ctx, LibFibreRxStream** tx_channel, LibFibreTxStream** rx_channel, size_t mtu) { + fibre::AsyncStreamLink* tx_link = new fibre::AsyncStreamLink(); + fibre::AsyncStreamLink* rx_link = new fibre::AsyncStreamLink(); + LibFibreRxStream* tx = new LibFibreRxStream(); + LibFibreTxStream* rx = new LibFibreTxStream(); + tx->source = tx_link; + rx->sink = rx_link; + + tx->on_closed = [](LibFibreRxStream* stream, void* ctx, fibre::StreamStatus status) { + auto link = reinterpret_cast(ctx); + link->close(status); + delete link; + delete stream; + }; + tx->on_closed_ctx = tx_link; + rx->on_closed = [](LibFibreTxStream* stream, void* ctx, fibre::StreamStatus status) { + auto link = reinterpret_cast(ctx); + link->close(status); + delete link; + delete stream; + }; + rx->on_closed_ctx = rx_link; + + if (tx_channel) { + *tx_channel = tx; + } + + if (rx_channel) { + *rx_channel = rx; + } + + fibre::ChannelDiscoveryResult result = {kFibreOk, rx_link, tx_link, mtu}; + discovery_ctx->completer->complete(result); +} + void libfibre_start_discovery(LibFibreCtx* ctx, const char* specs, size_t specs_len, struct LibFibreDiscoveryCtx** handle, on_found_object_cb_t on_found_object, on_stopped_cb_t on_stopped, void* cb_ctx) { if (!ctx) { @@ -237,12 +441,16 @@ void libfibre_start_discovery(LibFibreCtx* ctx, const char* specs, size_t specs_ const char* next_delim = std::find(prev_delim, specs + specs_len, ';'); const char* colon = std::find(prev_delim, next_delim, ':'); const char* colon_end = std::min(colon + 1, next_delim); - - if ((colon - prev_delim) == strlen("usb") && std::equal(prev_delim, colon, "usb")) { - ctx->libusb_discoverer.start_channel_discovery(colon_end, next_delim - colon_end, - &discovery_ctx->libusb_discovery_ctx, *discovery_ctx); + + std::string name{prev_delim, colon}; + auto it = ctx->discoverers.find(name); + + if (it == ctx->discoverers.end()) { + FIBRE_LOG(W) << "transport layer \"" << name << "\" not implemented"; } else { - FIBRE_LOG(W) << "transport layer \"" << std::string(prev_delim, colon - prev_delim) << "\" not implemented"; + discovery_ctx->context_handles[name] = nullptr; + it->second->start_channel_discovery(colon_end, next_delim - colon_end, + &discovery_ctx->context_handles[name], *discovery_ctx); } prev_delim = std::min(next_delim + 1, specs + specs_len); @@ -258,10 +466,10 @@ void libfibre_stop_discovery(LibFibreCtx* ctx, LibFibreDiscoveryCtx* discovery_c ctx->n_discoveries--; } - if (discovery_ctx->libusb_discovery_ctx) { - // TODO: implement "stopped" callback - ctx->libusb_discoverer.stop_channel_discovery(discovery_ctx->libusb_discovery_ctx); + for (auto& it: discovery_ctx->context_handles) { + ctx->discoverers[it.first]->stop_channel_discovery(it.second); } + discovery_ctx->context_handles.clear(); if (--discovery_ctx->use_count == 0) { FIBRE_LOG(D) << "deleting discovery context"; @@ -379,93 +587,62 @@ void resize_at(std::vector& vec, size_t pos, ssize_t delta) { } } -void transcode(fibre::LegacyObjectClient* client, std::vector& buffer, const std::vector& args, bool to) { - size_t offset = 0; +class ArgEncoder : public fibre::AsyncStreamSink, fibre::Completer { + void start_write(fibre::cbufptr_t buffer, fibre::TransferHandle* handle, Completer& completer) final; + void cancel_write(fibre::TransferHandle transfer_handle) final; + void complete(fibre::WriteResult result) final; - for (auto& arg: args) { - if (to && arg.codec == "endpoint_ref") { - fibre::cbufptr_t orig_range = fibre::cbufptr_t{buffer}.skip(offset).take(sizeof(uintptr_t)); +public: + LibFibreCallContext* call_ = nullptr; + fibre::AsyncStreamSink* encoded_stream_ = nullptr; + fibre::cbufptr_t decoded_buf_; // application-owned buffer + std::vector encoded_buf_; // libfibre-owned buffer used after transcoding from application buffer + size_t encoded_offset_ = 0; // offset in the TX stream after transcoding from application-facing format + fibre::TransferHandle transfer_handle_ = 0; + fibre::Completer* completer_ = nullptr; +}; - uintptr_t val = *reinterpret_cast(orig_range.begin()); +class ArgDecoder : public fibre::AsyncStreamSource, fibre::Completer { + void start_read(fibre::bufptr_t buffer, fibre::TransferHandle* handle, Completer& completer) final; + void cancel_read(fibre::TransferHandle transfer_handle) final; + void complete(fibre::ReadResult result) final; - resize_at(buffer, offset, (ssize_t)4 - (ssize_t)sizeof(uintptr_t)); - fibre::bufptr_t transcoded_range = fibre::bufptr_t{buffer}.skip(offset).take(4); +public: + LibFibreCallContext* call_ = nullptr; + fibre::AsyncStreamSource* encoded_stream_ = nullptr; + fibre::bufptr_t decoded_buf_; // application-owned buffer + std::vector encoded_buf_; // libfibre-owned buffer used before transcoding to application buffer + size_t encoded_offset_ = 0; // offset in the RX stream before transcoding to application-facing format + fibre::TransferHandle transfer_handle_ = 0; + fibre::Completer* completer_ = nullptr; +}; - auto obj = reinterpret_cast(val); - write_le(obj ? obj->ep_num : 0, &transcoded_range); - write_le(obj ? obj->client->json_crc_ : 0, &transcoded_range); +struct FIBRE_PRIVATE LibFibreCallContext : fibre::Completer { + void complete(FibreStatus status) final; - offset += 4; - - } else if (!to && arg.codec == "endpoint_ref") { - fibre::cbufptr_t orig_range = fibre::cbufptr_t{buffer}.skip(offset).take(4); - - uint16_t ep_num = *read_le(&orig_range); - uint16_t json_crc = *read_le(&orig_range); - - resize_at(buffer, offset, (ssize_t)sizeof(uintptr_t) - (ssize_t)4); - fibre::bufptr_t transcoded_range = fibre::bufptr_t{buffer}.skip(offset).take(sizeof(uintptr_t)); - - fibre::LegacyObject* obj_ptr = nullptr; - - if (ep_num && json_crc == client->json_crc_) { - for (auto& known_obj: client->objects_) { - if (known_obj->ep_num == ep_num) { - obj_ptr = known_obj.get(); - } - } - } - - FIBRE_LOG(D) << "placing transcoded ptr " << reinterpret_cast(obj_ptr); - *reinterpret_cast(transcoded_range.begin()) = reinterpret_cast(obj_ptr); - - offset += sizeof(uintptr_t); - - } else { - offset += arg.size; - } - } -} - -struct FIBRE_PRIVATE LibFibreCallContext : fibre::Completer { - void complete(fibre::LegacyObjectClient::CallResult output) final { - - FIBRE_LOG(D) << "received " << (output.end - rx_vec.data()) << " bytes "; - - // Prune vector to the end of valid data - rx_vec.erase(rx_vec.begin() + (output.end - rx_vec.data()), rx_vec.end()); - - transcode(obj->client, rx_vec, func->outputs, false); - - size_t len = std::min(rx_vec.size(), rx_buf.size()); - std::copy(rx_vec.begin(), rx_vec.begin() + len, rx_buf.begin()); - FIBRE_LOG(D) << "result is [" << as_hex(rx_buf) << "]"; - - if (on_completed) { - (*on_completed)(ctx, output.status, rx_buf.begin() + len); - } - delete this; - } + template + bool iterate_over_args_at(size_t encoded_offset, size_t max_encoded_length, size_t max_decoded_length, Func visitor); + uint8_t n_active_transfers = 0; fibre::LegacyObject* obj = nullptr; fibre::LegacyFibreFunction* func = nullptr; - void (*on_completed)(void*, FibreStatus, uint8_t*); - void* ctx; - fibre::cbufptr_t tx_buf; // application-owned buffer - fibre::bufptr_t rx_buf; // application-owned buffer - std::vector tx_vec; // libfibre-owned buffer used after transcoding from application buffer - std::vector rx_vec; // libfibre-owned buffer used before transcoding to application buffer - fibre::LegacyObjectClient::CallContext* handle; + on_call_completed_cb_t on_call_completed_ = nullptr; + void* call_cb_ctx_ = nullptr; + fibre::LegacyObjectClient::CallContext* handle_ = nullptr; + LibFibreTxStream tx_stream_; + LibFibreRxStream rx_stream_; + ArgDecoder decoder_; + ArgEncoder encoder_; }; void libfibre_start_call(LibFibreObject* obj, LibFibreFunction* func, - const uint8_t *input, size_t input_length, - uint8_t *output, size_t output_length, LibFibreCallContext** handle, + LibFibreTxStream** tx_stream, + LibFibreRxStream** rx_stream, on_call_completed_cb_t on_completed, void* cb_ctx) { - if (!obj || !func || !input || !output) { + if (!obj || !func) { if (on_completed) { - (*on_completed)(cb_ctx, kFibreInvalidArgument, output); + (*on_completed)(cb_ctx, kFibreInvalidArgument); } return; } @@ -481,43 +658,316 @@ void libfibre_start_call(LibFibreObject* obj, LibFibreFunction* func, if (!is_member) { FIBRE_LOG(W) << "attempt to invoke function on an object that does not implement it"; if (on_completed) { - (*on_completed)(cb_ctx, kFibreInvalidArgument, output); + (*on_completed)(cb_ctx, kFibreInvalidArgument); } return; } - std::vector tx_vec; - tx_vec.insert(tx_vec.begin(), input, input + input_length); - auto completer = new LibFibreCallContext(); completer->obj = obj_cast; completer->func = func_cast; - completer->on_completed = on_completed; - completer->ctx = cb_ctx; - completer->tx_buf = fibre::cbufptr_t{input, input + input_length}; - completer->rx_buf = fibre::bufptr_t{output, output + output_length}; - - completer->tx_vec.insert(completer->tx_vec.begin(), completer->tx_buf.begin(), completer->tx_buf.end()); - transcode(obj_cast->client, completer->tx_vec, func_cast->inputs, true); - - size_t output_size = 0; - for (auto& arg: func_cast->outputs) - output_size += arg.size; - FIBRE_LOG(D) << "sizing output to " << output_size; - completer->rx_vec.resize(output_size); + completer->on_call_completed_ = on_completed; + completer->call_cb_ctx_ = cb_ctx; + completer->encoder_.call_ = completer; + completer->decoder_.call_ = completer; + completer->tx_stream_.sink = &completer->encoder_; + completer->rx_stream_.source = &completer->decoder_; if (handle) { *handle = completer; } + if (tx_stream) { + *tx_stream = &completer->tx_stream_; + } + if (rx_stream) { + *rx_stream = &completer->rx_stream_; + } obj_cast->client->start_call(obj_cast->ep_num, func_cast, - fibre::cbufptr_t{completer->tx_vec}, fibre::bufptr_t{completer->rx_vec}, - &completer->handle, *completer); + &completer->handle_, *completer); + + completer->encoder_.encoded_stream_ = completer->handle_; + completer->decoder_.encoded_stream_ = completer->handle_; } -void libfibre_cancel_call(LibFibreCallContext* handle) { - if (handle && handle->on_completed) { - handle->obj->client->cancel_call(handle->handle); +void libfibre_end_call(LibFibreCallContext* handle) { + if (handle) { + handle->obj->client->cancel_call(handle->handle_); } } + +void LibFibreCallContext::complete(FibreStatus status) { + if (on_call_completed_) { + (*on_call_completed_)(call_cb_ctx_, status); + } + delete this; +} + +bool encode_for_transport(fibre::LegacyObjectClient* client, fibre::cbufptr_t src, fibre::bufptr_t dst, const fibre::LegacyFibreArg& arg) { + if (arg.codec == "endpoint_ref") { + if (src.size() < sizeof(uintptr_t) || dst.size() < 4) { + return false; + } + + uintptr_t val = *reinterpret_cast(src.begin()); + auto obj = reinterpret_cast(val); + write_le(obj ? obj->ep_num : 0, &dst); + write_le(obj ? obj->client->json_crc_ : 0, &dst); + } else { + if (src.size() < arg.size || dst.size() < arg.size) { + return false; + } + + memcpy(dst.begin(), src.begin(), arg.size); + } + + return true; +} + +bool decode_from_transport(fibre::LegacyObjectClient* client, fibre::cbufptr_t src, fibre::bufptr_t dst, const fibre::LegacyFibreArg& arg) { + if (arg.codec == "endpoint_ref") { + if (src.size() < 4 || dst.size() < sizeof(uintptr_t)) { + return false; + } + + uint16_t ep_num = *read_le(&src); + uint16_t json_crc = *read_le(&src); + + fibre::LegacyObject* obj_ptr = nullptr; + + if (ep_num && json_crc == client->json_crc_) { + for (auto& known_obj: client->objects_) { + if (known_obj->ep_num == ep_num) { + obj_ptr = known_obj.get(); + } + } + } + + FIBRE_LOG(D) << "placing transcoded ptr " << reinterpret_cast(obj_ptr); + *reinterpret_cast(dst.begin()) = reinterpret_cast(obj_ptr); + + } else { + if (src.size() < arg.size || dst.size() < arg.size) { + return false; + } + + memcpy(dst.begin(), src.begin(), arg.size); + } + + return true; +} + +/** + * @brief func: A functor that takes these arguments: + * - size_t rel_encoded_offset (relative to the starting pos described by encoded_offset) + * - size_t rel_decoded_offset (relative to the starting pos described by encoded_offset) + * - size_t encoded_length + * - size_t decoded_length + */ +template +bool LibFibreCallContext::iterate_over_args_at(size_t encoded_offset, size_t max_encoded_length, size_t max_decoded_length, Func visitor) { + ssize_t arg_offset = 0; + ssize_t len_diff = 0; + + for (auto& arg: func->outputs) { + if (arg.size >= SIZE_MAX || arg_offset + arg.size > encoded_offset) { + ssize_t rel_encoded_offset = arg_offset - (ssize_t)encoded_offset; + ssize_t rel_decoded_offset = arg_offset - (ssize_t)encoded_offset - len_diff; + + if (rel_decoded_offset >= max_decoded_length || rel_encoded_offset >= max_encoded_length) { + break; + } + + size_t encoded_arg_size = arg.size; + size_t decoded_arg_size = arg.codec == "endpoint_ref" ? sizeof(uintptr_t) : arg.size; + len_diff += encoded_arg_size - decoded_arg_size; + + if (rel_encoded_offset < 0) { + encoded_arg_size += rel_encoded_offset; + } + + if (rel_decoded_offset < 0) { + decoded_arg_size += rel_decoded_offset; + } + + if (encoded_arg_size < SIZE_MAX && rel_encoded_offset + encoded_arg_size > max_encoded_length) { + encoded_arg_size -= rel_encoded_offset + encoded_arg_size - max_encoded_length; + } + + if (decoded_arg_size < SIZE_MAX && rel_decoded_offset + decoded_arg_size > max_decoded_length) { + decoded_arg_size -= rel_decoded_offset + decoded_arg_size - max_decoded_length; + } + + if (!visitor(arg, rel_encoded_offset, rel_decoded_offset, encoded_arg_size, decoded_arg_size)) { + return false; + } + } + + arg_offset += arg.size; + } + + return true; +} + +void libfibre_start_tx(LibFibreTxStream* tx_stream, + const uint8_t* tx_buf, size_t tx_len, on_tx_completed_cb_t on_completed, + void* ctx) { + tx_stream->on_completed = on_completed; + tx_stream->ctx = ctx; + tx_stream->sink->start_write({tx_buf, tx_len}, &tx_stream->handle, *tx_stream); +} + +void libfibre_cancel_tx(LibFibreTxStream* tx_stream) { + tx_stream->sink->cancel_write(tx_stream->handle); +} + +void libfibre_close_tx(LibFibreTxStream* tx_stream, FibreStatus status) { + if (tx_stream->on_closed) { + (tx_stream->on_closed)(tx_stream, tx_stream->on_closed_ctx, convert_status(status)); + } +} + +void libfibre_start_rx(LibFibreRxStream* rx_stream, + uint8_t* rx_buf, size_t rx_len, on_rx_completed_cb_t on_completed, + void* ctx) { + rx_stream->on_completed = on_completed; + rx_stream->ctx = ctx; + rx_stream->source->start_read({rx_buf, rx_len}, &rx_stream->handle, *rx_stream); +} + +void libfibre_cancel_rx(LibFibreRxStream* rx_stream) { + rx_stream->source->cancel_read(rx_stream->handle); +} + +void libfibre_close_rx(LibFibreRxStream* rx_stream, FibreStatus status) { + if (rx_stream->on_closed) { + (rx_stream->on_closed)(rx_stream, rx_stream->on_closed_ctx, convert_status(status)); + } +} + +void ArgEncoder::start_write(fibre::cbufptr_t buffer, fibre::TransferHandle* handle, Completer& completer) { + // Allocate libfibre-internal TX buffer into which the application buffer + // will be encoded. The size can still change during transcoding. + encoded_buf_ = std::vector{}; + encoded_buf_.reserve(buffer.size()); + + // Transcode application buffer to stream buffer + bool ok = call_->iterate_over_args_at(encoded_offset_, SIZE_MAX, buffer.size(), [&]( + const fibre::LegacyFibreArg& arg, + ssize_t rel_encoded_offset, ssize_t rel_decoded_offset, + size_t encoded_arg_size, size_t decoded_arg_size) { + encoded_buf_.resize(rel_encoded_offset + encoded_arg_size); + fibre::bufptr_t encoded_buf = {encoded_buf_.data() + rel_encoded_offset, encoded_arg_size}; + fibre::cbufptr_t decoded_buf = {buffer.begin() + rel_decoded_offset, decoded_arg_size}; + return encode_for_transport(call_->obj->client, decoded_buf, encoded_buf, arg); + }); + + if (!ok) { + FIBRE_LOG(W) << "Transcoding before TX failed. Note that partial transcoding of arguments is not supported."; + completer.complete({fibre::kStreamError, buffer.begin()}); + return; + } + + decoded_buf_ = buffer; + call_->n_active_transfers++; + completer_ = &completer; + + encoded_stream_->start_write(encoded_buf_, &transfer_handle_, *this); +} + +void ArgEncoder::cancel_write(fibre::TransferHandle transfer_handle) { + encoded_stream_->cancel_write(transfer_handle_); +} + +void ArgEncoder::complete(fibre::WriteResult result) { + size_t n_sent = (result.end - encoded_buf_.data()); + FIBRE_LOG(D) << "sent " << n_sent << " bytes "; + + if (n_sent > encoded_buf_.size()) { + FIBRE_LOG(E) << "internal error: sent more bytes than expected"; + } + + ssize_t len_diff = encoded_buf_.size() - decoded_buf_.size(); + const uint8_t* tx_end = decoded_buf_.begin() + n_sent - len_diff; + + decoded_buf_ = {}; + encoded_buf_ = {}; + encoded_offset_ += n_sent; + call_->n_active_transfers--; + + safe_complete(completer_, {result.status, tx_end}); +} + +void ArgDecoder::start_read(fibre::bufptr_t buffer, fibre::TransferHandle* handle, Completer& completer) { + size_t encoded_size = 0; + + bool ok = call_->iterate_over_args_at(encoded_offset_, SIZE_MAX, buffer.size(), [&]( + const fibre::LegacyFibreArg& arg, + ssize_t rel_encoded_offset, ssize_t rel_decoded_offset, + size_t encoded_arg_size, size_t decoded_arg_size) { + encoded_size += encoded_arg_size; + return true; + }); + + if (!ok) { + FIBRE_LOG(W) << "Transcoding preparation before RX failed. Note that partial transcoding of arguments is not supported."; + completer.complete({fibre::kStreamError, buffer.begin()}); + return; + } + + encoded_buf_ = {}; + encoded_buf_.resize(encoded_size); + decoded_buf_ = buffer; + completer_ = &completer; + call_->n_active_transfers++; + + encoded_stream_->start_read(encoded_buf_, &transfer_handle_, *this); +} + +void ArgDecoder::cancel_read(fibre::TransferHandle transfer_handle) { + encoded_stream_->cancel_read(transfer_handle_); +} + +void ArgDecoder::complete(fibre::ReadResult result) { + transfer_handle_ = 0; + + size_t n_recv = result.end - encoded_buf_.data(); + FIBRE_LOG(D) << "received " << n_recv << " bytes "; + + if (n_recv > encoded_buf_.size()) { + FIBRE_LOG(E) << "internal error: received more bytes than expected"; + } + + ssize_t len_diff = encoded_buf_.size() - decoded_buf_.size(); + uint8_t* rx_end = decoded_buf_.begin() + n_recv - len_diff; + + size_t arg_offset = 0; + + // Transcode stream buffer to application buffer + bool ok = call_->iterate_over_args_at(encoded_offset_, n_recv, decoded_buf_.size(), [&]( + const fibre::LegacyFibreArg& arg, + ssize_t rel_encoded_offset, ssize_t rel_decoded_offset, + size_t encoded_arg_size, size_t decoded_arg_size) { + fibre::cbufptr_t encoded_buf = {encoded_buf_.data() + rel_encoded_offset, encoded_arg_size}; + fibre::bufptr_t decoded_buf = {decoded_buf_.begin() + rel_decoded_offset, decoded_arg_size}; + return decode_from_transport(call_->obj->client, encoded_buf, decoded_buf, arg); + }); + + if (!ok) { + FIBRE_LOG(W) << "Transcoding after RX failed. Partial transcoding of arguments is not supported."; + rx_end = decoded_buf_.begin(); + result.status = fibre::kStreamError; + } else if (rx_end > decoded_buf_.end()) { + FIBRE_LOG(E) << "miscalculated pointer: beyond buffer end"; + rx_end = decoded_buf_.end(); + result.status = fibre::kStreamError; + } + + decoded_buf_ = {}; + encoded_buf_ = {}; + encoded_offset_ += n_recv; + call_->n_active_transfers--; + + safe_complete(completer_, {result.status, rx_end}); +} diff --git a/Firmware/fibre-cpp/logging.cpp b/Firmware/fibre-cpp/logging.cpp index 31f37113..8ed92180 100644 --- a/Firmware/fibre-cpp/logging.cpp +++ b/Firmware/fibre-cpp/logging.cpp @@ -1,7 +1,7 @@ #include "logging.hpp" -#if !defined(_WIN32) && !defined(_WIN64) && !defined(__linux__) && !defined(__APPLE__) +#if !defined(_WIN32) && !defined(_WIN64) && !defined(__linux__) && !defined(__APPLE__) && !defined(EMSCRIPTEN) namespace std { StdoutStream cerr; diff --git a/Firmware/fibre-cpp/logging.hpp b/Firmware/fibre-cpp/logging.hpp index a4aeac62..05751e97 100644 --- a/Firmware/fibre-cpp/logging.hpp +++ b/Firmware/fibre-cpp/logging.hpp @@ -56,7 +56,7 @@ #define __FIBRE_LOGGING_HPP // TODO: support lite-version of logging on embedded systems -#if defined(_WIN32) || defined(_WIN64) || defined(__linux__) || defined(__APPLE__) +#if defined(_WIN32) || defined(_WIN64) || defined(__linux__) || defined(__APPLE__) || defined(EMSCRIPTEN) #include @@ -67,7 +67,7 @@ #include "windows.h" #endif -#if defined(_WIN32) || defined(_WIN64) || defined(__linux__) || defined(__APPLE__) +#if defined(_WIN32) || defined(_WIN64) || defined(__linux__) || defined(__APPLE__) || defined(EMSCRIPTEN) #include #include #else diff --git a/Firmware/fibre-cpp/package.lua b/Firmware/fibre-cpp/package.lua index 53101e1a..41bfafc1 100644 --- a/Firmware/fibre-cpp/package.lua +++ b/Firmware/fibre-cpp/package.lua @@ -1,8 +1,12 @@ -tup.include('../tupfiles/build.lua') - -fibre_package = define_package{ - sources={'protocol.cpp', 'posix_tcp.cpp', 'posix_udp.cpp'}, - libs={'pthread'}, - headers={'include'} +fibre_package = { + core_files = { + 'libfibre.cpp', + 'legacy_protocol.cpp', + 'legacy_object_client.cpp', + }, + features = { + LIBUSB={'platform_support/libusb_transport.cpp'}, + LOGGING={'logging.cpp'}, + } } diff --git a/Firmware/fibre-cpp/platform_support/libusb_transport.cpp b/Firmware/fibre-cpp/platform_support/libusb_transport.cpp index 7b2a15f4..31de2bf1 100644 --- a/Firmware/fibre-cpp/platform_support/libusb_transport.cpp +++ b/Firmware/fibre-cpp/platform_support/libusb_transport.cpp @@ -252,14 +252,16 @@ void LibusbDiscoverer::start_channel_discovery(const char* specs, size_t specs_l if (!success) { FIBRE_LOG(E) << "could not interpret channel discovery specs"; - on_found_channels.complete({kFibreInvalidArgument, nullptr, nullptr}); + on_found_channels.complete({kFibreInvalidArgument, nullptr, nullptr, 0}); return; } prev_delim = std::min(next_delim + 1, specs + specs_len); } - ChannelDiscoveryContext* subscription = new ChannelDiscoveryContext{interface_specs, &on_found_channels}; + MyChannelDiscoveryContext* subscription = new MyChannelDiscoveryContext{}; + subscription->interface_specs = interface_specs; + subscription->on_found_channels = &on_found_channels; subscriptions_.push_back(subscription); for (auto& dev: known_devices_) { @@ -456,7 +458,7 @@ void LibusbDiscoverer::poll_devices_now() { } } -void LibusbDiscoverer::consider_device(struct libusb_device *device, ChannelDiscoveryContext* subscription) { +void LibusbDiscoverer::consider_device(struct libusb_device *device, MyChannelDiscoveryContext* subscription) { uint8_t bus_number = libusb_get_bus_number(device); uint8_t dev_number = libusb_get_device_address(device); @@ -532,9 +534,12 @@ void LibusbDiscoverer::consider_device(struct libusb_device *device, ChannelDisc continue; } + size_t mtu = SIZE_MAX; + LibusbBulkInEndpoint* ep_in = new LibusbBulkInEndpoint(); if (libusb_ep_in && ep_in->init(this, my_dev.handle, libusb_ep_in->bEndpointAddress)) { my_dev.ep_in.push_back(ep_in); + mtu = std::min(mtu, (size_t)libusb_ep_in->wMaxPacketSize); } else { delete ep_in; ep_in = nullptr; @@ -543,13 +548,14 @@ void LibusbDiscoverer::consider_device(struct libusb_device *device, ChannelDisc LibusbBulkOutEndpoint* ep_out = new LibusbBulkOutEndpoint(); if (libusb_ep_out && ep_out->init(this, my_dev.handle, libusb_ep_out->bEndpointAddress)) { my_dev.ep_out.push_back(ep_out); + mtu = std::min(mtu, (size_t)libusb_ep_out->wMaxPacketSize); } else { delete ep_out; ep_out = nullptr; } if (subscription->on_found_channels) { - subscription->on_found_channels->complete({kFibreOk, ep_in, ep_out}); + subscription->on_found_channels->complete({kFibreOk, ep_in, ep_out, mtu}); } } } diff --git a/Firmware/fibre-cpp/platform_support/libusb_transport.hpp b/Firmware/fibre-cpp/platform_support/libusb_transport.hpp index 2d8ea2b0..33e1a2db 100644 --- a/Firmware/fibre-cpp/platform_support/libusb_transport.hpp +++ b/Firmware/fibre-cpp/platform_support/libusb_transport.hpp @@ -3,7 +3,7 @@ #include "../event_loop.hpp" #include "../async_stream.hpp" -#include +#include "../channel_discoverer.hpp" #include #include @@ -15,15 +15,9 @@ namespace fibre { class LibusbBulkInEndpoint; class LibusbBulkOutEndpoint; -struct ChannelDiscoveryResult { - FibreStatus status; - AsyncStreamSource* rx_channel; - AsyncStreamSink* tx_channel; -}; - template class FIBRE_PRIVATE LibusbBulkEndpoint; -class FIBRE_PRIVATE LibusbDiscoverer { +class FIBRE_PRIVATE LibusbDiscoverer : public ChannelDiscoverer { public: struct InterfaceSpecs { @@ -36,15 +30,17 @@ public: int interface_protocol = -1; // -1 to ignore }; - struct ChannelDiscoveryContext { + struct MyChannelDiscoveryContext : ChannelDiscoveryContext { InterfaceSpecs interface_specs; Completer* on_found_channels; }; + ~LibusbDiscoverer() { deinit(); } + int init(EventLoop* event_loop); int deinit() { return deinit(INT_MAX); } - void start_channel_discovery(const char* specs, size_t specs_len, ChannelDiscoveryContext** handle, Completer& on_found_channels); - int stop_channel_discovery(ChannelDiscoveryContext* handle); + void start_channel_discovery(const char* specs, size_t specs_len, ChannelDiscoveryContext** handle, Completer& on_found_channels) final; + int stop_channel_discovery(ChannelDiscoveryContext* handle) final; private: friend class LibusbBulkEndpoint; @@ -64,7 +60,7 @@ private: void on_remove_pollfd(int fd); int on_hotplug(struct libusb_device *dev, libusb_hotplug_event event); void poll_devices_now(); - void consider_device(struct libusb_device *device, ChannelDiscoveryContext* subscription); + void consider_device(struct libusb_device *device, MyChannelDiscoveryContext* subscription); EventLoop* event_loop_ = nullptr; bool using_sparate_libusb_thread_; // true on Windows. Initialized in init() @@ -75,7 +71,7 @@ private: EventLoopTimer* device_polling_timer_; EventLoopTimer* event_loop_timer_ = nullptr; std::unordered_map known_devices_; // key: bus_number << 8 | dev_number - std::vector subscriptions_; + std::vector subscriptions_; }; template diff --git a/Firmware/fibre-cpp/print_utils.hpp b/Firmware/fibre-cpp/print_utils.hpp index 07a23ee3..39825631 100644 --- a/Firmware/fibre-cpp/print_utils.hpp +++ b/Firmware/fibre-cpp/print_utils.hpp @@ -3,6 +3,7 @@ #include #include +#include namespace fibre { diff --git a/Firmware/fibre-cpp/.gitattributes b/tools/odrive/pyfibre/.gitattributes similarity index 100% rename from Firmware/fibre-cpp/.gitattributes rename to tools/odrive/pyfibre/.gitattributes diff --git a/tools/odrive/pyfibre/fibre/libfibre-linux-amd64.so b/tools/odrive/pyfibre/fibre/libfibre-linux-amd64.so index 6d22256e..41ffc404 100755 Binary files a/tools/odrive/pyfibre/fibre/libfibre-linux-amd64.so and b/tools/odrive/pyfibre/fibre/libfibre-linux-amd64.so differ diff --git a/tools/odrive/pyfibre/fibre/libfibre-linux-armhf.so b/tools/odrive/pyfibre/fibre/libfibre-linux-armhf.so index cca886ad..2f3e9eb2 100755 Binary files a/tools/odrive/pyfibre/fibre/libfibre-linux-armhf.so and b/tools/odrive/pyfibre/fibre/libfibre-linux-armhf.so differ diff --git a/tools/odrive/pyfibre/fibre/libfibre-macos-x86.dylib b/tools/odrive/pyfibre/fibre/libfibre-macos-x86.dylib index 98873047..2af09dec 100644 Binary files a/tools/odrive/pyfibre/fibre/libfibre-macos-x86.dylib and b/tools/odrive/pyfibre/fibre/libfibre-macos-x86.dylib differ diff --git a/tools/odrive/pyfibre/fibre/libfibre-windows-amd64.dll b/tools/odrive/pyfibre/fibre/libfibre-windows-amd64.dll index 14628c94..868fc1e2 100755 Binary files a/tools/odrive/pyfibre/fibre/libfibre-windows-amd64.dll and b/tools/odrive/pyfibre/fibre/libfibre-windows-amd64.dll differ diff --git a/tools/odrive/pyfibre/fibre/libfibre.py b/tools/odrive/pyfibre/fibre/libfibre.py index df8e7e6c..348bdc35 100644 --- a/tools/odrive/pyfibre/fibre/libfibre.py +++ b/tools/odrive/pyfibre/fibre/libfibre.py @@ -12,6 +12,10 @@ import time import platform from .utils import Logger, Event +# Enable this for better tracebacks in some cases +#import tracemalloc +#tracemalloc.start(10) + lib_names = { ('Linux', 'x86_64'): 'libfibre-linux-amd64.so', ('Linux', 'armv7l'): 'libfibre-linux-armhf.so', @@ -77,7 +81,9 @@ 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(None, c_void_p, c_int) +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 @@ -100,7 +106,7 @@ 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 @@ -128,12 +134,28 @@ 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.argtypes = [c_void_p, c_void_p, POINTER(c_void_p), POINTER(c_void_p), POINTER(c_void_p), OnCallCompletedSignature, c_void_p] libfibre_start_call.restype = None -libfibre_cancel_call = lib.libfibre_cancel_call -libfibre_cancel_call.argtypes = [c_void_p] -libfibre_cancel_call.restype = None +libfibre_end_call = lib.libfibre_end_call +libfibre_end_call.argtypes = [c_void_p] +libfibre_end_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 ------------------------------------------------------------# @@ -148,7 +170,7 @@ 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: @@ -234,6 +256,158 @@ def run_coroutine_threadsafe(loop, func): 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 RemoteFunction(object): """ Represents a callable function that maps to a function call on a remote object. @@ -245,30 +419,83 @@ class RemoteFunction(object): 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) + self._c_on_call_completed = OnCallCompletedSignature(self._on_completed) - def _on_completed(self, ctx, status, end_ptr): + def _on_completed(self, ctx, status): call = self._calls.pop(ctx) if status != kFibreOk: - call['future'].set_exception(_get_exception(status)) + call.set_exception(_get_exception(status)) else: - pos = 0 - outputs = [] + call.set_result(None) - 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 + def start_call(self, instance, cancellation_token): + """ + Starts invoking the function on the remote object. + Must be called from the Fibre thread. - 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)) + cancellation_token: A future that starts cancellation of the function + call when completed. + Returns: A tuple (tx_stream, rx_stream, future). The tx_stream can be + used to send input arguments to the function. The rx_stream can + be used to receive output arguments from the function. The + future is completed once the function call is fully terminated. + """ + assert(asyncio.get_event_loop() == instance._libfibre.loop) - def __call__(self, instance, *args): + future = instance._libfibre.loop.create_future() + call_id = insert_with_new_id(self._calls, future) + + call_handle = c_void_p(0) + tx_stream_handle = c_void_p(0) + rx_stream_handle = c_void_p(0) + + libfibre_start_call(instance._obj_handle, self._func_handle, + byref(call_handle), byref(tx_stream_handle), byref(rx_stream_handle), + self._c_on_call_completed, call_id) + + if not cancellation_token is None: + on_cancel = lambda: libfibre_end_call(call_handle) + cancellation_token.add_done_callback(on_cancel) + future.add_done_callback(cancellation_token.remove_done_callback(on_cancel)) + + return TxStream(self._libfibre, tx_stream_handle), RxStream(self._libfibre, rx_stream_handle), future + + async def async_call(self, instance, args, cancellation_token): + tx_stream, rx_stream, call_future = self.start_call(instance, cancellation_token) + + tx_buf = bytes() + for i, arg in enumerate(self._inputs): + tx_buf += arg[2].serialize(self._libfibre, args[i]) + + 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 + + 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: + return tuple(outputs) + + def __call__(self, instance, *args, cancellation_token = None): """ Starts invoking the function on the remote object. If this function is called from the Fibre thread then it is nonblocking @@ -283,23 +510,8 @@ class RemoteFunction(object): 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(instance, args, cancellation_token) + return asyncio.ensure_future(coro, loop=instance._libfibre.loop) def __get__(self, instance, owner): return MethodType(self, instance) if instance else self @@ -537,7 +749,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') @@ -547,7 +759,7 @@ 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): buf = path.encode('ascii')