Merge commit '9768ee6392ee515ad2371a2bb5662011c9a87941' into libfibre

Firmware/fibre-cpp/Dockerfile

@@ -47,6 +47,9 @@ RUN mkdir -p "third_party/libusb-windows" && \
 # Make Emscripten build its standard libraries for the WebAssembly target
 RUN echo "void test() {}" | em++ -x c - -o /tmp/a.out
 
+# Install dependencies for interface_generator.py
+RUN pacman -S --noconfirm python-yaml python-jinja python-jsonschema
+
 ENV THIRD_PARTY=/
 
 # Set up entrypoint

Firmware/fibre-cpp/README.md

@@ -1,6 +1,6 @@
 # fibre-cpp
 
-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).
+This directory provides the 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).
 
 ## Overview
 

Firmware/fibre-cpp/Tupfile.lua

@@ -1,7 +1,14 @@
 
+-- Projects that include fibre-cpp and also use tup can place a Tuprules.lua file
+-- into their root directory with the line `no_libfibre = true` to prevent
+-- libfibre from building.
+if no_libfibre == true then
+    return
+end
+
 tup.include('package.lua')
 
-CFLAGS = {'-fPIC -std=c++11 -DFIBRE_COMPILE'}
+CFLAGS = {'-fPIC -std=c++11 -DFIBRE_COMPILE -Wall'}
 LDFLAGS = {'-static-libstdc++'}
 
 
@@ -28,8 +35,9 @@ end
 CFLAGS += tup.getconfig("CFLAGS")
 LDFLAGS += tup.getconfig("LDFLAGS")
 DEBUG = get_bool_config("DEBUG", true)
+STRICT = get_bool_config("STRICT", false)
 
-machine = run_now(CXX..' -dumpmachine') -- works with both clang and GCC
+machine = fibre_run_now(CXX..' -dumpmachine') -- works with both clang and GCC
 
 BUILD_TYPE='-shared'
 
@@ -64,6 +72,10 @@ else
     CFLAGS += '-O3' -- TODO: add back -lfto
 end
 
+if STRICT then
+    CFLAGS += '-Werror'
+end
+
 function compile(src_file)
     obj_file = 'build/'..tup.file(src_file)..'.o'
     tup.frule{
@@ -81,7 +93,7 @@ pkg = get_fibre_package({
     enable_tcp_client_backend=get_bool_config("ENABLE_TCP_CLIENT_BACKEND", true),
     enable_libusb_backend=get_bool_config("ENABLE_LIBUSB_BACKEND", true),
     allow_heap=true,
-    pkgconf=tup.getconfig("USE_PKGCONF") or nil
+    pkgconf=(tup.getconfig("USE_PKGCONF") != "") and tup.getconfig("USE_PKGCONF") or nil
 })
 
 CFLAGS += pkg.cflags

Firmware/fibre-cpp/channel_discoverer.cpp

@@ -7,7 +7,7 @@
 using namespace fibre;
 
 bool ChannelDiscoverer::try_parse_key(const char* begin, const char* end, const char* key, const char** val_begin, const char** val_end) {
-    size_t keylen = strlen(key);
+    ssize_t keylen = strlen(key);
 
     while (begin != end) {
         const char* next_delim = std::find(begin, end, ',');

Firmware/fibre-cpp/configs/linux-amd64.config

@@ -1,5 +1,6 @@
 CONFIG_DEBUG=false
+CONFIG_STRICT=true
 CONFIG_CC="clang++"
 CONFIG_CFLAGS="-I$THIRD_PARTY./third_party/libusb-dev-armhf/usr/include/libusb-1.0"
-CONFIG_LDFLAGS="-L$THIRD_PARTY./third_party/libusb-amd64/lib/x86_64-linux-gnu/libusb-1.0.so.0.2.0"
+CONFIG_LDFLAGS="$THIRD_PARTY./third_party/libusb-amd64/lib/x86_64-linux-gnu/libusb-1.0.so.0.2.0"
 CONFIG_USE_PKGCONF=false

Firmware/fibre-cpp/configs/linux-armhf.config

@@ -1,4 +1,5 @@
 CONFIG_DEBUG=false
+CONFIG_STRICT=true
 CONFIG_CC="arm-linux-gnueabihf-g++"
 CONFIG_CFLAGS="-I$THIRD_PARTY./third_party/libusb-dev-armhf/usr/include/libusb-1.0"
 CONFIG_LDFLAGS="-L$THIRD_PARTY./third_party/libstdc++-linux-armhf/usr/lib/gcc-cross/arm-linux-gnueabihf/10 $THIRD_PARTY./third_party/libusb-armhf/usr/lib/arm-linux-gnueabihf/libusb-1.0.so.0"

Firmware/fibre-cpp/configs/macos-x86.config

@@ -1,4 +1,5 @@
 CONFIG_DEBUG=false
+CONFIG_STRICT=true
 CONFIG_CC="LD_LIBRARY_PATH=/opt/osxcross/lib MACOSX_DEPLOYMENT_TARGET=10.9 /opt/osxcross/bin/o64-clang++"
 CONFIG_CFLAGS="-I$THIRD_PARTY./third_party/libusb-1.0.23/libusb -arch i386 -arch x86_64"
 CONFIG_LDFLAGS="$THIRD_PARTY./third_party/libusb-1.0.23/build-macos-amd64/libusb/.libs/libusb-1.0.a -framework CoreFoundation -framework IOKit"

Firmware/fibre-cpp/configs/wasm.config

@@ -1,4 +1,5 @@
 CONFIG_DEBUG=true
+CONFIG_STRICT=true
 CONFIG_CC=/usr/lib/emscripten/em++
 CONFIG_CFLAGS=-include emscripten.h -DFIBRE_PUBLIC=EMSCRIPTEN_KEEPALIVE -s RESERVED_FUNCTION_POINTERS=1
 CONFIG_LDFLAGS=-s EXPORT_ES6=1 -s MODULARIZE=1 -s USE_ES6_IMPORT_META=0 -s 'EXTRA_EXPORTED_RUNTIME_METHODS=[addFunction, stringToUTF8Array, UTF8ArrayToString, ENV]'

Firmware/fibre-cpp/configs/windows-amd64.config

@@ -1,4 +1,5 @@
 CONFIG_DEBUG=false
+CONFIG_STRICT=true
 CONFIG_CC="x86_64-w64-mingw32-g++"
 CONFIG_CFLAGS="-I$THIRD_PARTY./third_party/libusb-windows/libusb-1.0.23/include/libusb-1.0"
 CONFIG_LDFLAGS="-static-libgcc $THIRD_PARTY./third_party/libusb-windows/libusb-1.0.23/MinGW64/static/libusb-1.0.a"

Firmware/fibre-cpp/endpoints_template.j2

@@ -15,6 +15,7 @@
 
 #include <fibre/introspection.hpp>
 #include <fibre/../../legacy_protocol.hpp>
+#include <fibre/../../crc.hpp>
 
 // Note: with -Og the functions with large switch statements reserves a huge amount
 // of stack space because they reserves separate space for the stack frame of each

Firmware/fibre-cpp/fibre.cpp

@@ -6,6 +6,7 @@
 #include "print_utils.hpp"
 #include <memory>
 #include <algorithm>
+#include <array>
 
 #if FIBRE_ALLOW_HEAP
 #include <unordered_map>
@@ -102,6 +103,17 @@ struct BackendDeinitializer {
     Context* ctx;
 };
 
+template<typename ... T, size_t ... Is>
+bool all(std::tuple<T...> args, std::index_sequence<Is...>) {
+    std::array<bool, sizeof...(T)> arr = { std::get<Is>(args) ... };
+    return std::all_of(arr.begin(), arr.end(), [](bool val) { return val; });
+}
+
+template<typename ... T>
+bool all(std::tuple<T...> args) {
+    return all(args, std::make_index_sequence<sizeof...(T)>());
+}
+
 Context* fibre::open(EventLoop* event_loop) {
     Context* ctx = my_alloc<Context>();
     if (!ctx) {
@@ -115,9 +127,9 @@ Context* fibre::open(EventLoop* event_loop) {
 
     // TODO: check static_backends_good
 
-    //if (std::all(static_backends_good)) {
-    //    return nullptr;
-    //}
+    if (all(static_backends_good)) {
+        return nullptr;
+    }
 
     return ctx;
 }
@@ -133,6 +145,7 @@ void fibre::close(Context* ctx) {
     my_free<Context>(ctx);
 }
 
+#if FIBRE_ALLOW_HEAP
 Domain* Context::create_domain(std::string specs) {
     FIBRE_LOG(D) << "creating domain with path \"" << specs << "\"";
 
@@ -191,6 +204,7 @@ void Context::deregister_backend(std::string name) {
     
     discoverers.erase(it);
 }
+#endif
 
 #if FIBRE_ENABLE_CLIENT
 void Domain::start_discovery(Callback<void, Object*, Interface*> on_found_object, Callback<void, Object*> on_lost_object) {

Firmware/fibre-cpp/include/fibre/cpp_utils.hpp

@@ -528,6 +528,9 @@ optional<T> make_optional(T& val) {
 
 } // namespace std
 
+#else
+#include <variant>
+#include <optional>
 #endif
 
 /* Stuff that should be in the STL but isn't ---------------------------------*/

Firmware/fibre-cpp/include/fibre/fibre.hpp

@@ -38,7 +38,7 @@ struct Function {
 
 struct Object;
 struct Interface;
-struct Domain;
+class Domain;
 
 template<typename T>
 struct StaticBackend {
@@ -87,11 +87,11 @@ struct Context {
 
 // TODO: don't declare these types here
 struct LegacyProtocolPacketBased;
-struct LegacyObjectClient;
+class LegacyObjectClient;
 struct LegacyObject;
 
 class Domain {
-    friend class Context;
+    friend struct Context;
 public:
 #if FIBRE_ENABLE_CLIENT
     // TODO: add interface argument

Firmware/fibre-cpp/interfaces_template.j2

@@ -12,6 +12,8 @@
 #ifndef __FIBRE_INTERFACES_HPP
 #define __FIBRE_INTERFACES_HPP
 
+[[userdata.c_preamble]]
+
 #include <fibre/../../protocol.hpp>
 
 #pragma GCC push_options
@@ -28,7 +30,7 @@ std::tuple<[% for arg in func.out.values() %][[arg.type.c_name]][[', ' if not lo
 [%- endmacro %]
 
 [%- macro render_interface(intf) %]
-class [[intf.name | to_pascal_case]]Intf {
+class [[intf.name | to_pascal_case]]Intf[% if intf.implements %] :[%- for base_intf in intf.implements %] public [[base_intf.c_name]][% endfor %][% endif %] {
 public:
 [%- for intf in intf.interfaces -%]
 [[render_interface(intf) | indent(4)]]
@@ -36,7 +38,7 @@ public:
 [%- for enum in intf.enums %]
     enum [[enum.name | to_pascal_case]] {
 [%- for k, value in enum['values'].items() %]
-        [[((enum.name + k) | to_macro_case).ljust(32)]] = [% if enum.is_flags %]0x[['%08x' | format(value.value)]][% else %][[value.value]][% endif %],
+        [[((enum.name | to_macro_case) + "_" + (k | to_macro_case)).ljust(32)]] = [% if enum.is_flags %]0x[['%08x' | format(value.value)]][% else %][[value.value]][% endif %],
 [%- endfor %]
     };
 [%- endfor %]

Firmware/fibre-cpp/legacy_object_client.cpp

@@ -469,6 +469,7 @@ void LegacyCallContext::resume_from_protocol(EndpointOperationResult result) {
         ep_result = result;
         return;
     }
+    op_handle_ = 0;
 
     std::variant<ResultFromApp, ResultFromProtocol> res = result; 
 

Firmware/fibre-cpp/legacy_object_client.hpp

@@ -20,6 +20,8 @@ struct EndpointOperationResult {
     uint8_t* rx_end;
 };
 
+// Lower 16 bits are the seqno. Upper 16 bits are all 1 for valid handles
+// (such that seqno 0 doesn't cause the handle to be 0)
 using EndpointOperationHandle = uint32_t;
 
 struct LegacyProtocolPacketBased;
@@ -51,7 +53,7 @@ struct LegacyFunction : Function {
 };
 
 struct FibreInterface;
-struct LegacyObjectClient;
+class LegacyObjectClient;
 
 struct LegacyFibreAttribute {
     std::shared_ptr<LegacyObject> object;

Firmware/fibre-cpp/legacy_protocol.cpp

@@ -273,7 +273,7 @@ void LegacyProtocolPacketBased::cancel_endpoint_operation(EndpointOperationHandl
         pending_operation_ = std::nullopt;
     }
 
-    auto it = expected_acks_.find(handle);
+    auto it = expected_acks_.find(seqno);
 
     if (it != expected_acks_.end()) {
         callback = it->second.callback;
@@ -336,17 +336,26 @@ void LegacyProtocolPacketBased::on_write_finished(WriteResult result) {
         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);
+        it->second.tx_done = true;
 
-        // If the TX task was a remote endpoint operation but didn't succeed
-        // we terminate that operation
-        if (result.status != kStreamOk) {
-            auto callback = it->second.callback;
-            auto tx_end = it->second.tx_buf.begin();
-            auto rx_end = it->second.rx_buf.begin();
+        if (it->second.rx_done) {
+            // It's possible that the RX operation completes before the TX operation
+            auto op = it->second;
             expected_acks_.erase(it);
-            callback.invoke_and_clear({result.status, result.end, rx_end});
+            op.callback.invoke_and_clear({kStreamOk, op.tx_buf.begin(), op.rx_buf.begin()});
+        } else if (result.status != kStreamOk) {
+            // If the TX task was a remote endpoint operation but didn't succeed
+            // we terminate that operation
+            auto op = it->second;
+            expected_acks_.erase(it);
+            op.callback.invoke_and_clear({result.status, result.end, op.rx_buf.begin()});
+        }
+
+        if (transmitting_op_) {
+            return;
         }
     }
 #endif
@@ -360,7 +369,11 @@ void LegacyProtocolPacketBased::on_write_finished(WriteResult result) {
         uint8_t* rx_end = rx_end_;
         rx_end_ = nullptr;
         on_read_finished({kStreamOk, rx_end});
-        return;
+#if FIBRE_ENABLE_CLIENT
+        if (transmitting_op_) {
+            return;
+        }
+#endif
     }
 #endif
 
@@ -371,7 +384,9 @@ void LegacyProtocolPacketBased::on_write_finished(WriteResult result) {
         EndpointOperation op = *pending_operation_;
         pending_operation_ = std::nullopt;
         start_endpoint_operation(op);
-        return;
+        if (transmitting_op_) {
+            return;
+        }
     }
 #endif
 }
@@ -415,11 +430,16 @@ 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 callback = it->second.callback;
-            expected_acks_.erase(it);
-            callback.invoke_and_clear({kStreamOk, tx_end, rx_end});
+            it->second.rx_buf = it->second.rx_buf.skip(n_copy);
+            it->second.rx_done = true;
+            FIBRE_LOG(T) << "received ACK: " << (*seq_no & 0x7fff);
+
+            // It's possible that the RX operation completes before the TX operation
+            if (it->second.tx_done) {
+                auto op = it->second;
+                expected_acks_.erase(it);
+                op.callback.invoke_and_clear({kStreamOk, op.tx_buf.begin(), op.rx_buf.begin()});
+            }
         }
 
 #else

Firmware/fibre-cpp/legacy_protocol.hpp

@@ -126,7 +126,9 @@ private:
         uint16_t seqno;
         uint16_t endpoint_id;
         cbufptr_t tx_buf;
+        bool tx_done;
         bufptr_t rx_buf;
+        bool rx_done;
         Callback<void, EndpointOperationResult> callback;
     };
 

Firmware/fibre-cpp/libfibre.cpp

@@ -12,7 +12,7 @@ USE_LOG_TOPIC(LIBFIBRE);
 
 static const struct LibFibreVersion libfibre_version = { 0, 1, 0 };
 
-class FIBRE_PRIVATE ExternalEventLoop : public fibre::EventLoop {
+class FIBRE_PRIVATE ExternalEventLoop final : public fibre::EventLoop {
 public:
     ExternalEventLoop(LibFibreEventLoop impl) : impl_(impl) {}
 
@@ -81,7 +81,7 @@ int ExternalDiscoverer::stop_channel_discovery(fibre::ChannelDiscoveryContext* h
 
 namespace fibre {
 
-class AsyncStreamLink : public AsyncStreamSink, public AsyncStreamSource {
+class AsyncStreamLink final : public AsyncStreamSink, public AsyncStreamSource {
 public:
     void start_write(cbufptr_t buffer, TransferHandle* handle, Callback<void, WriteResult> completer) final;
     void cancel_write(TransferHandle transfer_handle) final;
@@ -457,7 +457,7 @@ LibFibreStatus libfibre_get_attribute(LibFibreObject* parent_obj, LibFibreAttrib
         return kFibreInvalidArgument;
     }
 
-    LibFibreCtx* libfibre_ctx = reinterpret_cast<LibFibreCtx*>(parent_obj_cast->client->user_data_);
+    //LibFibreCtx* libfibre_ctx = reinterpret_cast<LibFibreCtx*>(parent_obj_cast->client->user_data_);
     fibre::LegacyObject* child_obj = attr_cast->object.get();
 
     if (!attr_cast->object->known_to_application) {

Firmware/fibre-cpp/package.lua

@@ -1,4 +1,5 @@
 
+fibre_root = tup.getcwd()
 
 -- Returns a table that contains the Fibre code files and the flags required to
 -- compile and link those files.
@@ -30,6 +31,7 @@
 --           linking the object files.
 function get_fibre_package(args)
     pkg = {
+        root = fibre_root,
         code_files = {
             'fibre.cpp',
             'channel_discoverer.cpp',
@@ -58,9 +60,11 @@ function get_fibre_package(args)
         end
 
     elseif args.pkgconf == false then
+        print("not using pkgconf")
         pkgconf_file = nil
         pkgconf = null_pkgconf
     else
+        print("using pkgconf: "..args.pkgconf)
         pkgconf_file = args.pkgconf
         pkgconf = real_pkgconf
     end
@@ -108,7 +112,7 @@ end
 -- graph).
 -- Returns the values (return_code, stdout) where stdout has the trailing new
 -- line removed.
-function run_now(command)
+function fibre_run_now(command)
     local handle
     handle = io.popen(command)
     local output = handle:read("*a")
@@ -117,13 +121,13 @@ function run_now(command)
 end
 
 function test_pkgconf(name)
-    local str, rc = run_now(name.." --version 2>&1 >/dev/null")
+    local str, rc = fibre_run_now(name.." --version 2>&1 >/dev/null")
     return rc
 end
 
 function real_pkgconf(pkg, lib)
-    pkg.cflags += run_now(pkgconf_file..' '..lib..' --cflags')
-    pkg.ldflags += run_now(pkgconf_file..' '..lib..' --libs')
+    pkg.cflags += fibre_run_now(pkgconf_file..' '..lib..' --cflags')
+    pkg.ldflags += fibre_run_now(pkgconf_file..' '..lib..' --libs')
 end
 
 function null_pkgconf(pkg, lib)

Firmware/fibre-cpp/platform_support/libusb_transport.cpp

@@ -216,8 +216,6 @@ bool LibusbDiscoverer::deinit(int stage) {
 void LibusbDiscoverer::start_channel_discovery(const char* specs, size_t specs_len, ChannelDiscoveryContext** handle, Callback<void, ChannelDiscoveryResult> on_found_channels) {
     FIBRE_LOG(D) << "starting discovery with filter \"" << std::string(specs, specs_len) << "\"";
 
-    const char* prev_delim = specs;
-
     InterfaceSpecs interface_specs;
 
     try_parse_key(specs, specs + specs_len, "bus", &interface_specs.bus);
@@ -450,7 +448,6 @@ void LibusbDiscoverer::consider_device(struct libusb_device *device, MyChannelDi
         }
     }
     
-    static libusb_device_handle *handle = nullptr;
     struct libusb_config_descriptor* config_desc = nullptr;
 
     if (libusb_get_active_config_descriptor(device, &config_desc) != LIBUSB_SUCCESS) {
@@ -645,7 +642,7 @@ void LibusbBulkEndpoint<TRes>::on_transfer_finished() {
         ssize_t n_devices = libusb_get_device_list(parent_->libusb_ctx_, &list);
 
         if (n_devices >= 0) {
-            for (size_t i = 0; i < n_devices; ++i) {
+            for (size_t i = 0; i < (size_t)n_devices; ++i) {
                 if (list[i] == dev) {
                     break;
                 }

Firmware/fibre-cpp/platform_support/libusb_transport.hpp

@@ -95,7 +95,7 @@ private:
     Callback<void, TRes> completer_ = nullptr;
 };
 
-class LibusbBulkInEndpoint : public LibusbBulkEndpoint<ReadResult>, public AsyncStreamSource {
+class LibusbBulkInEndpoint final : public LibusbBulkEndpoint<ReadResult>, public AsyncStreamSource {
 public:
     void start_read(bufptr_t buffer, TransferHandle* handle, Callback<void, ReadResult> completer) final {
         start_transfer(buffer, handle, completer);
@@ -106,7 +106,7 @@ public:
     }
 };
 
-class LibusbBulkOutEndpoint : public LibusbBulkEndpoint<WriteResult>, public AsyncStreamSink {
+class LibusbBulkOutEndpoint final : public LibusbBulkEndpoint<WriteResult>, public AsyncStreamSink {
 public:
     void start_write(cbufptr_t buffer, TransferHandle* handle, Callback<void, WriteResult> completer) final {
         start_transfer({

Firmware/fibre-cpp/platform_support/posix_socket.cpp

@@ -409,7 +409,7 @@ std::optional<ReadResult> PosixSocket::read_sync(bufptr_t buffer) {
             return {{kStreamError, buffer.end()}}; // the function might have written to the buffer
         }
 
-    } else if (n_received > buffer.size()) {
+    } else if ((size_t)n_received > buffer.size()) {
         FIBRE_LOG(E) << "received too many bytes";
         return {{kStreamError, buffer.end()}};
 
@@ -441,7 +441,7 @@ std::optional<WriteResult> PosixSocket::write_sync(cbufptr_t buffer) {
             return {{kStreamError, buffer.end()}}; // the function might have written to the buffer
         }
 
-    } else if (n_sent > buffer.size()) {
+    } else if ((size_t)n_sent > buffer.size()) {
         FIBRE_LOG(E) << "sent too many bytes";
         return {{kStreamError, buffer.end()}};
 

Firmware/fibre-cpp/platform_support/posix_socket.hpp

@@ -107,7 +107,7 @@ void stop_listening(ConnectionContext* ctx);
  * 
  * Note: To make this work on Windows, a "poll"-based worker must be implemented.
  */
-class PosixSocket : public AsyncStreamSource, public AsyncStreamSink {
+class PosixSocket final : public AsyncStreamSource, public AsyncStreamSink {
 public:
     /**
      * @brief Initializes the object with the given socket ID.

Firmware/fibre-cpp/stream_utils.hpp

@@ -1,13 +1,13 @@
 #ifndef __FIBRE_STREAM_UTILS_HPP
 #define __FIBRE_STREAM_UTILS_HPP
 
-#include "async_stream.hpp"
+#include <fibre/async_stream.hpp>
 #include <string.h>
 
 namespace fibre {
 
 template<size_t I>
-class BufferedStreamSink : Completer<WriteResult> {
+class BufferedStreamSink {
 public:
     BufferedStreamSink(AsyncStreamSink& sink) : sink_(sink) {}
 
@@ -50,17 +50,17 @@ public:
             // nothing to do
         } else if (read_idx_ < write_idx_) {
             is_active_ = true;
-            sink_.start_write({buffer_ + read_idx_, buffer_ + write_idx_}, &transfer_handle_, *this);
+            sink_.start_write({buffer_ + read_idx_, buffer_ + write_idx_}, &transfer_handle_, MEMBER_CB(this, on_write_complete));
         } else if (read_idx_ > write_idx_) {
             is_active_ = true;
-            sink_.start_write({buffer_ + read_idx_, buffer_ + I}, &transfer_handle_, *this);
+            sink_.start_write({buffer_ + read_idx_, buffer_ + I}, &transfer_handle_, MEMBER_CB(this, on_write_complete));
         } else {
             // nothing to do
         }
     }
 
 private:
-    void complete(WriteResult result) final {
+    void on_write_complete(WriteResult result) {
         is_active_ = false;
         transfer_handle_ = 0;
 
@@ -93,7 +93,7 @@ private:
  * operation at a time but are written to by multiple independent sources.
  */
 template<size_t NSlots>
-class AsyncStreamSinkMultiplexer : public AsyncStreamSink, Completer<WriteResult> {
+class AsyncStreamSinkMultiplexer : public AsyncStreamSink {
 public:
     AsyncStreamSinkMultiplexer(AsyncStreamSink& sink) : sink_(sink) {}
 
@@ -102,7 +102,7 @@ public:
             auto& [slot_in_use, slot_buf, slot_completer] = slots_[i];
             if (!__atomic_exchange_n(&slot_in_use, true, __ATOMIC_SEQ_CST)) {
                 slot_buf = buffer;
-                slot_completer = &completer;
+                slot_completer = completer;
 
                 if (handle) {
                     *handle = i + 1; // returning a valid handle of 0 is not a good idea
@@ -111,7 +111,7 @@ public:
                 // If the underlying sink wasn't busy, start it now.
                 if (active_slot_ == 0) {
                     active_slot_ = i + 1;
-                    sink_.start_write(slot_buf, &transfer_handle_, *this);
+                    sink_.start_write(slot_buf, &transfer_handle_, MEMBER_CB(this, on_write_complete));
                 }
 
                 return;
@@ -121,7 +121,7 @@ public:
         if (handle) {
             *handle = 0;
         }
-        completer.complete({kStreamError, buffer.begin()});
+        completer.invoke({kStreamError, buffer.begin()});
     }
 
     void cancel_write(TransferHandle transfer_handle) final {
@@ -134,12 +134,12 @@ public:
             auto completer = slot_completer;
             auto end = slot_buf.end();
             slot_in_use = false;
-            safe_complete(completer, {kStreamCancelled, end});
+            completer.invoke_and_clear({kStreamCancelled, end});
         }
     }
 
 private:
-    void complete(fibre::WriteResult result) final {
+    void on_write_complete(fibre::WriteResult result) {
         transfer_handle_ = 0;
 
         auto& [slot_in_use, slot_buf, slot_completer] =  slots_[active_slot_ - 1];
@@ -147,7 +147,7 @@ private:
         auto completer = slot_completer;
         slot_in_use = false;
 
-        safe_complete(completer, result);
+        completer.invoke_and_clear(result);
 
         // Select new slot before announcing completion of the old
         size_t active_slot = 0;
@@ -167,12 +167,12 @@ private:
             auto& [slot_in_use, slot_buf, slot_completer] = slots_[active_slot - 1];
             (void) slot_in_use;
             (void) slot_completer;
-            sink_.start_write(slot_buf, &transfer_handle_, *this);
+            sink_.start_write(slot_buf, &transfer_handle_, MEMBER_CB(this, on_write_complete));
         }
     }
     
     AsyncStreamSink& sink_;
-    std::tuple<bool, cbufptr_t, Completer<WriteResult>*> slots_[NSlots];
+    std::tuple<bool, cbufptr_t, Callback<void, WriteResult>> slots_[NSlots];
     size_t active_slot_ = 0;
     TransferHandle transfer_handle_ = 0;
 };