Merge #34: Add shared_ptr callback support

9ce0335 Add comment saying how to fix clientInvoke missing Proxy.Context assert (Russell Yanofsky)
31b4f1b Add shared_ptr ownership and lifetime support (Russell Yanofsky)
27f8a35 Add saveCallback / callbackSaved test setup (Russell Yanofsky)
5390a1b Add support for passing shared_ptrs without extending lifetime (Russell Yanofsky)
39bbf74 Add CustomReadField priority param for more flexibility and consistency with CustomBuildField (Russell Yanofsky)
da489be Fix bugs in PassField overload for callback objects passed by reference (Russell Yanofsky)
ab4568b Add test coverage for thread map and callbacks (Russell Yanofsky)

Pull request description:

  This is needed after bitcoin/bitcoin#18338 which changed `handleNotifications()` `Notifications&` callback argument to `std::shared_ptr<Notifications>`.

  Easiest way to support this was to change `ProxyServerBase` reference from a raw pointer to shared pointer and make necessary BuildField changes to pass along the `shared_ptr`. Alternative might have been to add more generic cleanup support to `ProxyServerBase` instead of hardcoding `shared_ptr`.

  The change also required making the ReadField callback overload more generic, which was a straightforward but kind of big change that touched a lot of code.

  There weren't any unit tests for callback support previously, so a lot of new test coverage was added. It includes coverage for `shared_ptr` lifetime correctness, making sure there's an IPC call decrementing server `shared_ptr` reference count when client `shared_ptr` proxy is reset.

Top commit has no ACKs.

Tree-SHA512: 96607bb339a5184ab34f9c133f7d5a9a6ac37a7a71187bdd6bd15235b44690cf19d8c09ad6e6966e51886cb00036cecf268406af7afd749b659970e9b00179ee

Author: ryanofsky

include/mp/proxy-io.h

@@ -412,28 +412,25 @@ ProxyClientBase<Interface, Impl>::~ProxyClientBase() noexcept
 }
 
 template <typename Interface, typename Impl>
-ProxyServerBase<Interface, Impl>::ProxyServerBase(Impl* impl, bool owned, Connection& connection)
-    : m_impl(impl), m_owned(owned), m_connection(connection)
+ProxyServerBase<Interface, Impl>::ProxyServerBase(std::shared_ptr<Impl> impl, Connection& connection)
+    : m_impl(std::move(impl)), m_connection(connection)
 {
-    assert(impl != nullptr);
+    assert(m_impl);
     std::unique_lock<std::mutex> lock(m_connection.m_loop.m_mutex);
     m_connection.m_loop.addClient(lock);
 }
 
 template <typename Interface, typename Impl>
 ProxyServerBase<Interface, Impl>::~ProxyServerBase()
 {
-    if (Impl* impl = m_impl) {
+    if (m_impl) {
         // If impl is non-null, it means client was not destroyed cleanly (was
         // killed or disconnected). Since client isn't providing thread to run
         // destructor on, run asynchronously. Do not run destructor on current
         // (event loop) thread since destructors could be making IPC calls or
         // doing expensive cleanup.
-        if (m_owned) {
-            m_connection.addAsyncCleanup([impl] { delete impl; });
-        }
-        m_impl = nullptr;
-        m_owned = false;
+        auto impl = std::move(m_impl);
+        m_connection.addAsyncCleanup([impl]() mutable { impl.reset(); });
     }
     std::unique_lock<std::mutex> lock(m_connection.m_loop.m_mutex);
     m_connection.m_loop.removeClient(lock);
@@ -442,9 +439,7 @@ ProxyServerBase<Interface, Impl>::~ProxyServerBase()
 template <typename Interface, typename Impl>
 void ProxyServerBase<Interface, Impl>::invokeDestroy()
 {
-    if (m_owned) delete m_impl;
-    m_impl = nullptr;
-    m_owned = false;
+    m_impl.reset();
 }
 
 struct ThreadContext
@@ -509,9 +504,10 @@ template <typename InitInterface, typename InitImpl>
 void _Serve(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream, InitImpl& init)
 {
     loop.m_incoming_connections.emplace_front(loop, kj::mv(stream), [&](Connection& connection) {
-        // Set owned to false so proxy object doesn't attempt to delete init
-        // object on disconnect/close.
-        return kj::heap<mp::ProxyServer<InitInterface>>(&init, false, connection);
+        // Disable deleter so proxy server object doesn't attempt to delete the
+        // init implementation when the proxy client is destroyed or
+        // disconnected.
+        return kj::heap<ProxyServer<InitInterface>>(std::shared_ptr<InitImpl>(&init, [](InitImpl*){}), connection);
     });
     auto it = loop.m_incoming_connections.begin();
     it->onDisconnect([&loop, it] {

include/mp/proxy-types.h

@@ -235,6 +235,7 @@ struct ReadDestValue
 
 template <typename LocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<std::optional<LocalType>>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -256,6 +257,7 @@ decltype(auto) CustomReadField(TypeList<std::optional<LocalType>>,
 
 template <typename LocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<std::shared_ptr<LocalType>>,
+    Priority<0>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -277,6 +279,7 @@ decltype(auto) CustomReadField(TypeList<std::shared_ptr<LocalType>>,
 
 template <typename LocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<LocalType*>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -290,6 +293,7 @@ decltype(auto) CustomReadField(TypeList<LocalType*>,
 
 template <typename LocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<std::shared_ptr<const LocalType>>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -309,6 +313,7 @@ decltype(auto) CustomReadField(TypeList<std::shared_ptr<const LocalType>>,
 
 template <typename LocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<std::vector<LocalType>>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -329,6 +334,7 @@ decltype(auto) CustomReadField(TypeList<std::vector<LocalType>>,
 
 template <typename LocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<std::set<LocalType>>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -347,6 +353,7 @@ decltype(auto) CustomReadField(TypeList<std::set<LocalType>>,
 
 template <typename KeyLocalType, typename ValueLocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<std::map<KeyLocalType, ValueLocalType>>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -367,6 +374,7 @@ decltype(auto) CustomReadField(TypeList<std::map<KeyLocalType, ValueLocalType>>,
 
 template <typename KeyLocalType, typename ValueLocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<std::pair<KeyLocalType, ValueLocalType>>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -391,6 +399,7 @@ decltype(auto) CustomReadField(TypeList<std::pair<KeyLocalType, ValueLocalType>>
 
 template <typename KeyLocalType, typename ValueLocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<std::tuple<KeyLocalType, ValueLocalType>>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -407,6 +416,7 @@ decltype(auto) CustomReadField(TypeList<std::tuple<KeyLocalType, ValueLocalType>
 
 template <typename LocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<LocalType>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest,
@@ -417,6 +427,7 @@ decltype(auto) CustomReadField(TypeList<LocalType>,
 
 template <typename LocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<LocalType>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest,
@@ -431,6 +442,7 @@ decltype(auto) CustomReadField(TypeList<LocalType>,
 
 template <typename LocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<LocalType>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest,
@@ -443,6 +455,7 @@ decltype(auto) CustomReadField(TypeList<LocalType>,
 
 template <typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<std::string>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -453,6 +466,7 @@ decltype(auto) CustomReadField(TypeList<std::string>,
 
 template <size_t size, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<unsigned char[size]>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -478,6 +492,23 @@ std::unique_ptr<Impl> CustomMakeProxyClient(InvokeContext& context, typename Int
 
 template <typename LocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<std::unique_ptr<LocalType>>,
+    Priority<1>,
+    InvokeContext& invoke_context,
+    Input&& input,
+    ReadDest&& read_dest,
+    typename Decay<decltype(input.get())>::Calls* enable = nullptr)
+{
+    using Interface = typename Decay<decltype(input.get())>::Calls;
+    if (input.has()) {
+        return read_dest.construct(
+                                   CustomMakeProxyClient<Interface, LocalType>(invoke_context, std::move(input.get())));
+    }
+    return read_dest.construct();
+}
+
+template <typename LocalType, typename Input, typename ReadDest>
+decltype(auto) CustomReadField(TypeList<std::shared_ptr<LocalType>>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest,
@@ -505,6 +536,7 @@ struct ProxyCallFn
 
 template <typename FnR, typename... FnParams, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<std::function<FnR(FnParams...)>>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest)
@@ -546,6 +578,7 @@ void ReadOne(TypeList<LocalType> param,
 
 template <typename LocalType, typename Input, typename ReadDest>
 decltype(auto) CustomReadField(TypeList<LocalType> param,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     ReadDest&& read_dest,
@@ -557,7 +590,7 @@ decltype(auto) CustomReadField(TypeList<LocalType> param,
 template <typename... LocalTypes, typename... Args>
 void ReadField(TypeList<LocalTypes...>, Args&&... args)
 {
-    CustomReadField(TypeList<RemoveCvRef<LocalTypes>...>(), std::forward<Args>(args)...);
+    CustomReadField(TypeList<RemoveCvRef<LocalTypes>...>(), Priority<2>(), std::forward<Args>(args)...);
 }
 
 template <typename LocalType, typename Input>
@@ -642,18 +675,18 @@ void CustomBuildField(TypeList<std::function<FnR(FnParams...)>>,
         using Interface = typename decltype(output.get())::Calls;
         using Callback = ProxyCallbackImpl<FnR, FnParams...>;
         output.set(kj::heap<ProxyServer<Interface>>(
-            new Callback(std::forward<Value>(value)), true /* owned */, invoke_context.connection));
+            std::make_shared<Callback>(std::forward<Value>(value)), invoke_context.connection));
     }
 }
 
 template <typename Interface, typename Impl>
-kj::Own<typename Interface::Server> MakeProxyServer(InvokeContext& context, std::unique_ptr<Impl>&& impl)
+kj::Own<typename Interface::Server> MakeProxyServer(InvokeContext& context, std::shared_ptr<Impl> impl)
 {
-    return kj::heap<ProxyServer<Interface>>(impl.release(), true /* owned */, context.connection);
+    return kj::heap<ProxyServer<Interface>>(std::move(impl), context.connection);
 }
 
 template <typename Interface, typename Impl>
-kj::Own<typename Interface::Server> CustomMakeProxyServer(InvokeContext& context, std::unique_ptr<Impl>&& impl)
+kj::Own<typename Interface::Server> CustomMakeProxyServer(InvokeContext& context, std::shared_ptr<Impl>&& impl)
 {
     return MakeProxyServer<Interface, Impl>(context, std::move(impl));
 }
@@ -665,25 +698,40 @@ void CustomBuildField(TypeList<std::unique_ptr<Impl>>,
     Value&& value,
     Output&& output,
     typename Decay<decltype(output.get())>::Calls* enable = nullptr)
+{
+    if (value) {
+        using Interface = typename decltype(output.get())::Calls;
+        output.set(CustomMakeProxyServer<Interface, Impl>(invoke_context, std::shared_ptr<Impl>(value.release())));
+    }
+}
+
+template <typename Impl, typename Value, typename Output>
+void CustomBuildField(TypeList<std::shared_ptr<Impl>>,
+    Priority<2>,
+    InvokeContext& invoke_context,
+    Value&& value,
+    Output&& output,
+    typename Decay<decltype(output.get())>::Calls* enable = nullptr)
 {
     if (value) {
         using Interface = typename decltype(output.get())::Calls;
         output.set(CustomMakeProxyServer<Interface, Impl>(invoke_context, std::move(value)));
     }
 }
 
-template <typename LocalType, typename Output>
-void CustomBuildField(TypeList<LocalType&>,
+template <typename Impl, typename Output>
+void CustomBuildField(TypeList<Impl&>,
     Priority<1>,
     InvokeContext& invoke_context,
-    LocalType& value,
+    Impl& value,
     Output&& output,
     typename decltype(output.get())::Calls* enable = nullptr)
 {
-    // Set owned to false so proxy object doesn't attempt to delete interface
-    // reference when it is discarded remotely, or on disconnect.
-    output.set(kj::heap<ProxyServer<typename decltype(output.get())::Calls>>(
-        &value, false /* owned */, invoke_context.connection));
+    // Disable deleter so proxy server object doesn't attempt to delete the
+    // wrapped implementation when the proxy client is destroyed or
+    // disconnected.
+    using Interface = typename decltype(output.get())::Calls;
+    output.set(CustomMakeProxyServer<Interface, Impl>(invoke_context, std::shared_ptr<Impl>(&value, [](Impl*){})));
 }
 
 template <typename LocalType, typename Value, typename Output>
@@ -977,7 +1025,7 @@ auto PassField(TypeList<LocalType&>, ServerContext& server_context, Fn&& fn, Arg
     const auto& params = server_context.call_context.getParams();
     const auto& input = Make<StructField, Accessor>(params);
     using Interface = typename Decay<decltype(input.get())>::Calls;
-    auto param = std::make_unique<ProxyClient<Interface>>(input.get(), *server_context.proxy_server.m_connection);
+    auto param = std::make_unique<ProxyClient<Interface>>(input.get(), &server_context.proxy_server.m_connection, false);
     fn.invoke(server_context, std::forward<Args>(args)..., *param);
 }
 
@@ -1053,6 +1101,7 @@ void CustomBuildField(TypeList<>,
 
 template <typename Input>
 decltype(auto) CustomReadField(TypeList<>,
+    Priority<1>,
     InvokeContext& invoke_context,
     Input&& input,
     typename std::enable_if<std::is_same<decltype(input.get()), ThreadMap::Client>::value>::type* enable = nullptr)
@@ -1325,6 +1374,14 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel
     if (!g_thread_context.waiter) {
         assert(g_thread_context.thread_name.empty());
         g_thread_context.thread_name = ThreadName(proxy_client.m_connection->m_loop.m_exe_name);
+        // If next assert triggers, it means clientInvoke is being called from
+        // the capnp event loop thread. This can happen when a ProxyServer
+        // method implementation that runs synchronously on the event loop
+        // thread tries to make a blocking callback to the client. Any server
+        // method that makes a blocking callback or blocks in general needs to
+        // run asynchronously off the event loop thread. This is easy to fix by
+        // just adding a 'context :Proxy.Context' argument to the capnp method
+        // declaration so the server method runs in a dedicated thread.
         assert(!g_thread_context.loop_thread);
         g_thread_context.waiter = std::make_unique<Waiter>();
         proxy_client.m_connection->m_loop.logPlain()

include/mp/proxy.h

@@ -82,16 +82,15 @@ struct ProxyServerBase : public virtual Interface_::Server
     using Interface = Interface_;
     using Impl = Impl_;
 
-    ProxyServerBase(Impl* impl, bool owned, Connection& connection);
+    ProxyServerBase(std::shared_ptr<Impl> impl, Connection& connection);
     virtual ~ProxyServerBase();
     void invokeDestroy();
 
-    Impl* m_impl;
     /**
-     * Whether or not to delete native interface pointer when this capnp server
-     * goes out of scope. This is true for servers created to wrap
-     * unique_ptr<Impl> method arguments, but false for servers created to wrap
-     * Impl& method arguments.
+     * Implementation pointer that may or may not be owned and deleted when this
+     * capnp server goes out of scope. It is owned for servers created to wrap
+     * unique_ptr<Impl> method arguments, but unowned for servers created to
+     * wrap Impl& method arguments.
      *
      * In the case of Impl& arguments, custom code is required on other side of
      * the connection to delete the capnp client & server objects since native
@@ -100,7 +99,7 @@ struct ProxyServerBase : public virtual Interface_::Server
      * this is implemented with addCloseHook callbacks to delete clients at
      * appropriate times depending on semantics of the particular method being
      * wrapped. */
-    bool m_owned;
+    std::shared_ptr<Impl> m_impl;
     Connection& m_connection;
 };
 
@@ -159,7 +158,7 @@ struct ProxyMethodTraits<MethodParams, Require<decltype(ProxyMethod<MethodParams
 {
     template <typename ServerContext, typename... Args>
     static auto invoke(ServerContext& server_context, Args&&... args) -> AUTO_RETURN(
-        (server_context.proxy_server.m_impl->*ProxyMethod<MethodParams>::impl)(std::forward<Args>(args)...))
+        (server_context.proxy_server.m_impl.get()->*ProxyMethod<MethodParams>::impl)(std::forward<Args>(args)...))
 };
 
 //! Customizable (through template specialization) traits class used in generated ProxyClient implementations from

src/mp/test/foo.capnp

@@ -14,6 +14,17 @@ interface FooInterface $Proxy.wrap("mp::test::FooImplementation") {
     mapSize @1 (map :List(Pair(Text, Text))) -> (result :Int32);
     pass @2 (arg :FooStruct) -> (result :FooStruct);
     raise @3 (arg :FooStruct) -> (error :FooStruct $Proxy.exception("mp::test::FooStruct"));
+    initThreadMap @4 (threadMap: Proxy.ThreadMap) -> (threadMap :Proxy.ThreadMap);
+    callback @5 (context :Proxy.Context, callback :FooCallback, arg: Int32) -> (result :Int32);
+    callbackUnique @6 (context :Proxy.Context, callback :FooCallback, arg: Int32) -> (result :Int32);
+    callbackShared @7 (context :Proxy.Context, callback :FooCallback, arg: Int32) -> (result :Int32);
+    saveCallback @8 (context :Proxy.Context, callback :FooCallback) -> ();
+    callbackSaved @9 (context :Proxy.Context, arg: Int32) -> (result :Int32);
+}
+
+interface FooCallback $Proxy.wrap("mp::test::FooCallback") {
+    destroy @0 (context :Proxy.Context) -> ();
+    call @1 (context :Proxy.Context, arg :Int32) -> (result :Int32);
 }
 
 struct FooStruct $Proxy.wrap("mp::test::FooStruct") {