diff --git a/ChangeLog b/ChangeLog index fbc8b7d..488cf91 100644 --- a/ChangeLog +++ b/ChangeLog @@ -262,6 +262,7 @@ v2.0.0 - Fix for external event loops in which the event loop thread ID was not correctly initialized (now fixed and simplified by not needing the thread ID anymore) - Introduce native integration for sd-event - Add method to get currently processed message also to `IConnection` +- Add Slot-returning overloads of `callMethodAsync()` functions - `[[nodiscard]]` attribute has been added to relevant API methods. - Add new `SDBUSCPP_SDBUS_LIB` CMake configuration variable determining which sd-bus library shall be picked - Switch to C++20 standard (but C++20 is not required, and the used C++20 features are conditionally compiled) diff --git a/docs/using-sdbus-c++.md b/docs/using-sdbus-c++.md index e83dae5..7d6be9f 100644 --- a/docs/using-sdbus-c++.md +++ b/docs/using-sdbus-c++.md @@ -1231,6 +1231,10 @@ int main(int argc, char *argv[]) Empty `error` parameter means that no D-Bus error occurred while making the call, and subsequent arguments are valid D-Bus method return values. However, `error` parameter containing a value means that an error occurred during the call (and subsequent arguments are simply default-constructed), and the underlying `Error` instance provides us with the error name and message. +> **_Tip_:** The function returns the `sdbus::PendingAsyncCall` object, a non-owning, observing handle to the async call. It can be used to query whether the call is still in progress, and to cancel the call. + +> **_Tip_:** There is also the `.uponReplyInvoke(callback, sdbus::return_slot);` variant with the `return_slot` tag, which returns `Slot` object, an owning RAII handle to the async call. This makes the client an owner of the pending async call. Letting go of the handle means cancelling the call. + Another option is to finish the async call statement with `getResultAsFuture()`, which is a template function which takes the list of types returned by the D-Bus method (empty list in case of `void`-returning method) which returns a `std::future` object, which will later, when the reply arrives, be set to contain the return value(s). Or if the call returns an error, `sdbus::Error` will be thrown by `std::future::get()`. The future object will contain void for a void-returning D-Bus method, a single type for a single value returning D-Bus method, and a `std::tuple` to hold multiple return values of a D-Bus method. diff --git a/include/sdbus-c++/ConvenienceApiClasses.h b/include/sdbus-c++/ConvenienceApiClasses.h index 6a1d5c0..d644ab8 100644 --- a/include/sdbus-c++/ConvenienceApiClasses.h +++ b/include/sdbus-c++/ConvenienceApiClasses.h @@ -131,6 +131,7 @@ namespace sdbus { AsyncMethodInvoker& withTimeout(const std::chrono::duration<_Rep, _Period>& timeout); template AsyncMethodInvoker& withArguments(_Args&&... args); template PendingAsyncCall uponReplyInvoke(_Function&& callback); + template [[nodiscard]] Slot uponReplyInvoke(_Function&& callback, return_slot_t); // Returned future will be std::future for no (void) D-Bus method return value // or std::future for single D-Bus method return value // or std::future> for multiple method return values @@ -140,6 +141,7 @@ namespace sdbus { friend IProxy; AsyncMethodInvoker(IProxy& proxy, const MethodName& methodName); AsyncMethodInvoker(IProxy& proxy, const char* methodName); + template async_reply_handler makeAsyncReplyHandler(_Function&& callback); private: IProxy& proxy_; @@ -190,6 +192,7 @@ namespace sdbus { public: AsyncPropertyGetter& onInterface(std::string_view interfaceName); template PendingAsyncCall uponReplyInvoke(_Function&& callback); + template [[nodiscard]] Slot uponReplyInvoke(_Function&& callback, return_slot_t); std::future getResultAsFuture(); private: @@ -232,6 +235,7 @@ namespace sdbus { template AsyncPropertySetter& toValue(_Value&& value); AsyncPropertySetter& toValue(Variant value); template PendingAsyncCall uponReplyInvoke(_Function&& callback); + template [[nodiscard]] Slot uponReplyInvoke(_Function&& callback, return_slot_t); std::future getResultAsFuture(); private: @@ -267,6 +271,7 @@ namespace sdbus { public: AsyncAllPropertiesGetter& onInterface(std::string_view interfaceName); template PendingAsyncCall uponReplyInvoke(_Function&& callback); + template [[nodiscard]] Slot uponReplyInvoke(_Function&& callback, return_slot_t); std::future> getResultAsFuture(); private: diff --git a/include/sdbus-c++/ConvenienceApiClasses.inl b/include/sdbus-c++/ConvenienceApiClasses.inl index 93ea4a1..4be40a9 100644 --- a/include/sdbus-c++/ConvenienceApiClasses.inl +++ b/include/sdbus-c++/ConvenienceApiClasses.inl @@ -287,7 +287,24 @@ namespace sdbus { { assert(method_.isValid()); // onInterface() must be placed/called prior to this function - auto asyncReplyHandler = [callback = std::forward<_Function>(callback)](MethodReply reply, std::optional error) + return proxy_.callMethodAsync(method_, makeAsyncReplyHandler(std::forward<_Function>(callback)), timeout_); + } + + template + [[nodiscard]] Slot AsyncMethodInvoker::uponReplyInvoke(_Function&& callback, return_slot_t) + { + assert(method_.isValid()); // onInterface() must be placed/called prior to this function + + return proxy_.callMethodAsync( method_ + , makeAsyncReplyHandler(std::forward<_Function>(callback)) + , timeout_ + , return_slot ); + } + + template + inline async_reply_handler AsyncMethodInvoker::makeAsyncReplyHandler(_Function&& callback) + { + return [callback = std::forward<_Function>(callback)](MethodReply reply, std::optional error) { // Create a tuple of callback input arguments' types, which will be used // as a storage for the argument values deserialized from the message. @@ -312,8 +329,6 @@ namespace sdbus { // Invoke callback with input arguments from the tuple. sdbus::apply(callback, std::move(error), args); }; - - return proxy_.callMethodAsync(method_, std::move(asyncReplyHandler), timeout_); } template @@ -474,7 +489,8 @@ namespace sdbus { template PendingAsyncCall AsyncPropertyGetter::uponReplyInvoke(_Function&& callback) { - static_assert(std::is_invocable_r_v, Variant>, "Property get callback function must accept std::optional and property value as Variant"); + static_assert( std::is_invocable_r_v, Variant> + , "Property get callback function must accept std::optional and property value as Variant" ); assert(!interfaceName_.empty()); // onInterface() must be placed/called prior to this function @@ -484,6 +500,20 @@ namespace sdbus { .uponReplyInvoke(std::forward<_Function>(callback)); } + template + [[nodiscard]] Slot AsyncPropertyGetter::uponReplyInvoke(_Function&& callback, return_slot_t) + { + static_assert( std::is_invocable_r_v, Variant> + , "Property get callback function must accept std::optional and property value as Variant" ); + + assert(!interfaceName_.empty()); // onInterface() must be placed/called prior to this function + + return proxy_.callMethodAsync("Get") + .onInterface(DBUS_PROPERTIES_INTERFACE_NAME) + .withArguments(interfaceName_, propertyName_) + .uponReplyInvoke(std::forward<_Function>(callback), return_slot); + } + inline std::future AsyncPropertyGetter::getResultAsFuture() { assert(!interfaceName_.empty()); // onInterface() must be placed/called prior to this function @@ -575,7 +605,8 @@ namespace sdbus { template PendingAsyncCall AsyncPropertySetter::uponReplyInvoke(_Function&& callback) { - static_assert(std::is_invocable_r_v>, "Property set callback function must accept std::optional only"); + static_assert( std::is_invocable_r_v> + , "Property set callback function must accept std::optional only" ); assert(!interfaceName_.empty()); // onInterface() must be placed/called prior to this function @@ -585,6 +616,20 @@ namespace sdbus { .uponReplyInvoke(std::forward<_Function>(callback)); } + template + [[nodiscard]] Slot AsyncPropertySetter::uponReplyInvoke(_Function&& callback, return_slot_t) + { + static_assert( std::is_invocable_r_v> + , "Property set callback function must accept std::optional only" ); + + assert(!interfaceName_.empty()); // onInterface() must be placed/called prior to this function + + return proxy_.callMethodAsync("Set") + .onInterface(DBUS_PROPERTIES_INTERFACE_NAME) + .withArguments(interfaceName_, propertyName_, std::move(value_)) + .uponReplyInvoke(std::forward<_Function>(callback), return_slot); + } + inline std::future AsyncPropertySetter::getResultAsFuture() { assert(!interfaceName_.empty()); // onInterface() must be placed/called prior to this function @@ -644,6 +689,20 @@ namespace sdbus { .uponReplyInvoke(std::forward<_Function>(callback)); } + template + [[nodiscard]] Slot AsyncAllPropertiesGetter::uponReplyInvoke(_Function&& callback, return_slot_t) + { + static_assert( std::is_invocable_r_v, std::map> + , "All properties get callback function must accept std::optional and a map of property names to their values" ); + + assert(!interfaceName_.empty()); // onInterface() must be placed/called prior to this function + + return proxy_.callMethodAsync("GetAll") + .onInterface(DBUS_PROPERTIES_INTERFACE_NAME) + .withArguments(interfaceName_) + .uponReplyInvoke(std::forward<_Function>(callback), return_slot); + } + inline std::future> AsyncAllPropertiesGetter::getResultAsFuture() { assert(!interfaceName_.empty()); // onInterface() must be placed/called prior to this function diff --git a/include/sdbus-c++/IProxy.h b/include/sdbus-c++/IProxy.h index 8f0d0bd..9ea16a6 100644 --- a/include/sdbus-c++/IProxy.h +++ b/include/sdbus-c++/IProxy.h @@ -89,7 +89,6 @@ namespace sdbus { * @brief Calls method on the remote D-Bus object * * @param[in] message Message representing a method call - * @param[in] timeout Timeout for dbus call in microseconds * @return A method reply message * * The call does not block if the method call has dont-expect-reply flag set. In that case, @@ -108,11 +107,44 @@ namespace sdbus { * its own bus connection. So-called light-weight proxies (ones created with `dont_run_event_loop_thread` * tag are designed for exactly that purpose. * + * The default D-Bus method call timeout is used. See IConnection::getMethodCallTimeout(). + * * Note: To avoid messing with messages, use API on a higher level of abstraction defined below. * * @throws sdbus::Error in case of failure (also in case the remote function returned an error) */ - virtual MethodReply callMethod(const MethodCall& message, uint64_t timeout = 0) = 0; + virtual MethodReply callMethod(const MethodCall& message) = 0; + + /*! + * @brief Calls method on the remote D-Bus object + * + * @param[in] message Message representing a method call + * @param[in] timeout Method call timeout (in microseconds) + * @return A method reply message + * + * The call does not block if the method call has dont-expect-reply flag set. In that case, + * the call returns immediately and the return value is an empty, invalid method reply. + * + * The call blocks otherwise, waiting for the remote peer to send back a reply or an error, + * or until the call times out. + * + * While blocking, other concurrent operations (in other threads) on the underlying bus + * connection are stalled until the call returns. This is not an issue in vast majority of + * (simple, single-threaded) applications. In asynchronous, multi-threaded designs involving + * shared bus connections, this may be an issue. It is advised to instead use an asynchronous + * callMethod() function overload, which does not block the bus connection, or do the synchronous + * call from another Proxy instance created just before the call and then destroyed (which is + * anyway quite a typical approach in D-Bus implementations). Such proxy instance must have + * its own bus connection. So-called light-weight proxies (ones created with `dont_run_event_loop_thread` + * tag are designed for exactly that purpose. + * + * If timeout is zero, the default D-Bus method call timeout is used. See IConnection::getMethodCallTimeout(). + * + * Note: To avoid messing with messages, use API on a higher level of abstraction defined below. + * + * @throws sdbus::Error in case of failure (also in case the remote function returned an error) + */ + virtual MethodReply callMethod(const MethodCall& message, uint64_t timeout) = 0; /*! * @copydoc IProxy::callMethod(const MethodCall&,uint64_t) @@ -125,8 +157,7 @@ namespace sdbus { * * @param[in] message Message representing an async method call * @param[in] asyncReplyCallback Handler for the async reply - * @param[in] timeout Timeout for dbus call in microseconds - * @return Cookie for the the pending asynchronous call + * @return Observing handle for the the pending asynchronous call * * This is a callback-based way of asynchronously calling a remote D-Bus method. * @@ -134,13 +165,95 @@ namespace sdbus { * the provided async reply handler will get invoked from the context of the bus * connection I/O event loop thread. * + * An non-owning, observing async call handle is returned that can be used to query call status or cancel the call. + * + * The default D-Bus method call timeout is used. See IConnection::getMethodCallTimeout(). + * + * Note: To avoid messing with messages, use API on a higher level of abstraction defined below. + * + * @throws sdbus::Error in case of failure + */ + virtual PendingAsyncCall callMethodAsync(const MethodCall& message, async_reply_handler asyncReplyCallback) = 0; + + /*! + * @brief Calls method on the D-Bus object asynchronously + * + * @param[in] message Message representing an async method call + * @param[in] asyncReplyCallback Handler for the async reply + * @return RAII-style slot handle representing the ownership of the async call + * + * This is a callback-based way of asynchronously calling a remote D-Bus method. + * + * The call itself is non-blocking. It doesn't wait for the reply. Once the reply arrives, + * the provided async reply handler will get invoked from the context of the bus + * connection I/O event loop thread. + * + * A slot (an owning handle) is returned for the async call. Lifetime of the call is bound to the lifetime of the slot. + * The slot can be used to cancel the method call at a later time by simply destroying it. + * + * The default D-Bus method call timeout is used. See IConnection::getMethodCallTimeout(). + * + * Note: To avoid messing with messages, use API on a higher level of abstraction defined below. + * + * @throws sdbus::Error in case of failure + */ + [[nodiscard]] virtual Slot callMethodAsync( const MethodCall& message + , async_reply_handler asyncReplyCallback + , return_slot_t ) = 0; + + /*! + * @brief Calls method on the D-Bus object asynchronously, with custom timeout + * + * @param[in] message Message representing an async method call + * @param[in] asyncReplyCallback Handler for the async reply + * @param[in] timeout Method call timeout (in microseconds) + * @return Observing handle for the the pending asynchronous call + * + * This is a callback-based way of asynchronously calling a remote D-Bus method. + * + * The call itself is non-blocking. It doesn't wait for the reply. Once the reply arrives, + * the provided async reply handler will get invoked from the context of the bus + * connection I/O event loop thread. + * + * An non-owning, observing async call handle is returned that can be used to query call status or cancel the call. + * + * If timeout is zero, the default D-Bus method call timeout is used. See IConnection::getMethodCallTimeout(). + * * Note: To avoid messing with messages, use API on a higher level of abstraction defined below. * * @throws sdbus::Error in case of failure */ virtual PendingAsyncCall callMethodAsync( const MethodCall& message , async_reply_handler asyncReplyCallback - , uint64_t timeout = 0 ) = 0; + , uint64_t timeout ) = 0; + + /*! + * @brief Calls method on the D-Bus object asynchronously, with custom timeout + * + * @param[in] message Message representing an async method call + * @param[in] asyncReplyCallback Handler for the async reply + * @param[in] timeout Method call timeout (in microseconds) + * @return RAII-style slot handle representing the ownership of the async call + * + * This is a callback-based way of asynchronously calling a remote D-Bus method. + * + * The call itself is non-blocking. It doesn't wait for the reply. Once the reply arrives, + * the provided async reply handler will get invoked from the context of the bus + * connection I/O event loop thread. + * + * A slot (an owning handle) is returned for the async call. Lifetime of the call is bound to the lifetime of the slot. + * The slot can be used to cancel the method call at a later time by simply destroying it. + * + * If timeout is zero, the default D-Bus method call timeout is used. See IConnection::getMethodCallTimeout(). + * + * Note: To avoid messing with messages, use API on a higher level of abstraction defined below. + * + * @throws sdbus::Error in case of failure + */ + [[nodiscard]] virtual Slot callMethodAsync( const MethodCall& message + , async_reply_handler asyncReplyCallback + , uint64_t timeout + , return_slot_t ) = 0; /*! * @copydoc IProxy::callMethod(const MethodCall&,async_reply_handler,uint64_t) @@ -150,6 +263,15 @@ namespace sdbus { , async_reply_handler asyncReplyCallback , const std::chrono::duration<_Rep, _Period>& timeout ); + /*! + * @copydoc IProxy::callMethod(const MethodCall&,async_reply_handler,uint64_t,return_slot_t) + */ + template + [[nodiscard]] Slot callMethodAsync( const MethodCall& message + , async_reply_handler asyncReplyCallback + , const std::chrono::duration<_Rep, _Period>& timeout + , return_slot_t ); + /*! * @brief Calls method on the D-Bus object asynchronously * @@ -163,6 +285,8 @@ namespace sdbus { * the provided future object will be set to contain the reply (or sdbus::Error * in case the remote method threw an exception). * + * The default D-Bus method call timeout is used. See IConnection::getMethodCallTimeout(). + * * Note: To avoid messing with messages, use higher-level API defined below. * * @throws sdbus::Error in case of failure @@ -183,6 +307,8 @@ namespace sdbus { * the provided future object will be set to contain the reply (or sdbus::Error * in case the remote method threw an exception, or the call timed out). * + * If timeout is zero, the default D-Bus method call timeout is used. See IConnection::getMethodCallTimeout(). + * * Note: To avoid messing with messages, use higher-level API defined below. * * @throws sdbus::Error in case of failure @@ -273,8 +399,12 @@ namespace sdbus { * @param[in] signalName Name of the signal * @param[in] signalHandler Callback that implements the body of the signal handler * - * A signal can be subscribed to and unsubscribed from at any time during proxy - * lifetime. The subscription is active immediately after the call. + * A signal can be subscribed to at any time during proxy lifetime. The subscription + * is active immediately after the call, and stays active for the entire lifetime + * of the Proxy object. + * + * To be able to unsubscribe from the signal at a later time, use the registerSignalHandler() + * overload with request_slot tag. * * @throws sdbus::Error in case of failure */ @@ -292,8 +422,9 @@ namespace sdbus { * @return RAII-style slot handle representing the ownership of the subscription * * A signal can be subscribed to and unsubscribed from at any time during proxy - * lifetime. The subscription is active immediately after the call. The subscription - * is unregistered when the client destroys the returned slot object. + * lifetime. The subscription is active immediately after the call. The lifetime + * of the subscription is bound to the lifetime of the slot object. The subscription + * is unregistered by letting go of the slot object. * * @throws sdbus::Error in case of failure */ @@ -566,10 +697,10 @@ namespace sdbus { private: friend internal::Proxy; - PendingAsyncCall(std::weak_ptr callData); + PendingAsyncCall(std::weak_ptr callInfo); private: - std::weak_ptr callData_; + std::weak_ptr callInfo_; }; // Out-of-line member definitions @@ -590,6 +721,16 @@ namespace sdbus { return callMethodAsync(message, std::move(asyncReplyCallback), microsecs.count()); } + template + inline Slot IProxy::callMethodAsync( const MethodCall& message + , async_reply_handler asyncReplyCallback + , const std::chrono::duration<_Rep, _Period>& timeout + , return_slot_t ) + { + auto microsecs = std::chrono::duration_cast(timeout); + return callMethodAsync(message, std::move(asyncReplyCallback), microsecs.count(), return_slot); + } + template inline std::future IProxy::callMethodAsync( const MethodCall& message , const std::chrono::duration<_Rep, _Period>& timeout diff --git a/include/sdbus-c++/StandardInterfaces.h b/include/sdbus-c++/StandardInterfaces.h index 272a724..ad44784 100644 --- a/include/sdbus-c++/StandardInterfaces.h +++ b/include/sdbus-c++/StandardInterfaces.h @@ -162,12 +162,24 @@ namespace sdbus { return proxy_->getPropertyAsync(propertyName).onInterface(interfaceName).uponReplyInvoke(std::forward<_Function>(callback)); } + template + [[nodiscard]] Slot GetAsync(const InterfaceName& interfaceName, const PropertyName& propertyName, _Function&& callback, return_slot_t) + { + return proxy_->getPropertyAsync(propertyName).onInterface(interfaceName).uponReplyInvoke(std::forward<_Function>(callback), return_slot); + } + template PendingAsyncCall GetAsync(std::string_view interfaceName, std::string_view propertyName, _Function&& callback) { return proxy_->getPropertyAsync(propertyName).onInterface(interfaceName).uponReplyInvoke(std::forward<_Function>(callback)); } + template + [[nodiscard]] Slot GetAsync(std::string_view interfaceName, std::string_view propertyName, _Function&& callback, return_slot_t) + { + return proxy_->getPropertyAsync(propertyName).onInterface(interfaceName).uponReplyInvoke(std::forward<_Function>(callback), return_slot); + } + std::future GetAsync(const InterfaceName& interfaceName, const PropertyName& propertyName, with_future_t) { return proxy_->getPropertyAsync(propertyName).onInterface(interfaceName).getResultAsFuture(); @@ -204,12 +216,24 @@ namespace sdbus { return proxy_->setPropertyAsync(propertyName).onInterface(interfaceName).toValue(value).uponReplyInvoke(std::forward<_Function>(callback)); } + template + PendingAsyncCall SetAsync(const InterfaceName& interfaceName, const PropertyName& propertyName, const sdbus::Variant& value, _Function&& callback, return_slot_t) + { + return proxy_->setPropertyAsync(propertyName).onInterface(interfaceName).toValue(value).uponReplyInvoke(std::forward<_Function>(callback), return_slot); + } + template PendingAsyncCall SetAsync(std::string_view interfaceName, std::string_view propertyName, const sdbus::Variant& value, _Function&& callback) { return proxy_->setPropertyAsync(propertyName).onInterface(interfaceName).toValue(value).uponReplyInvoke(std::forward<_Function>(callback)); } + template + PendingAsyncCall SetAsync(std::string_view interfaceName, std::string_view propertyName, const sdbus::Variant& value, _Function&& callback, return_slot_t) + { + return proxy_->setPropertyAsync(propertyName).onInterface(interfaceName).toValue(value).uponReplyInvoke(std::forward<_Function>(callback), return_slot); + } + std::future SetAsync(const InterfaceName& interfaceName, const PropertyName& propertyName, const sdbus::Variant& value, with_future_t) { return proxy_->setPropertyAsync(propertyName).onInterface(interfaceName).toValue(value).getResultAsFuture(); @@ -236,12 +260,24 @@ namespace sdbus { return proxy_->getAllPropertiesAsync().onInterface(interfaceName).uponReplyInvoke(std::forward<_Function>(callback)); } + template + PendingAsyncCall GetAllAsync(const InterfaceName& interfaceName, _Function&& callback, return_slot_t) + { + return proxy_->getAllPropertiesAsync().onInterface(interfaceName).uponReplyInvoke(std::forward<_Function>(callback), return_slot); + } + template PendingAsyncCall GetAllAsync(std::string_view interfaceName, _Function&& callback) { return proxy_->getAllPropertiesAsync().onInterface(interfaceName).uponReplyInvoke(std::forward<_Function>(callback)); } + template + PendingAsyncCall GetAllAsync(std::string_view interfaceName, _Function&& callback, return_slot_t) + { + return proxy_->getAllPropertiesAsync().onInterface(interfaceName).uponReplyInvoke(std::forward<_Function>(callback), return_slot); + } + std::future> GetAllAsync(const InterfaceName& interfaceName, with_future_t) { return proxy_->getAllPropertiesAsync().onInterface(interfaceName).getResultAsFuture(); diff --git a/include/sdbus-c++/TypeTraits.h b/include/sdbus-c++/TypeTraits.h index 7b146c5..164cd15 100644 --- a/include/sdbus-c++/TypeTraits.h +++ b/include/sdbus-c++/TypeTraits.h @@ -84,7 +84,7 @@ namespace sdbus { // Type-erased RAII-style handle to callbacks/subscriptions registered to sdbus-c++ using Slot = std::unique_ptr>; - // Tag specifying that an owning slot handle shall be returned from a registration/subscription function to the caller + // Tag specifying that an owning handle (so-called slot) of the logical resource shall be provided to the client struct return_slot_t { explicit return_slot_t() = default; }; inline constexpr return_slot_t return_slot{}; // Tag specifying that the library shall own the slot resulting from the call of the function (so-called floating slot) diff --git a/src/Proxy.cpp b/src/Proxy.cpp index 135b702..0cb525c 100644 --- a/src/Proxy.cpp +++ b/src/Proxy.cpp @@ -94,6 +94,11 @@ MethodCall Proxy::createMethodCall(const char* interfaceName, const char* method return connection_->createMethodCall(destination_.c_str(), objectPath_.c_str(), interfaceName, methodName); } +MethodReply Proxy::callMethod(const MethodCall& message) +{ + return Proxy::callMethod(message, /*timeout*/ 0); +} + MethodReply Proxy::callMethod(const MethodCall& message, uint64_t timeout) { SDBUS_THROW_ERROR_IF(!message.isValid(), "Invalid method call message provided", EINVAL); @@ -101,19 +106,44 @@ MethodReply Proxy::callMethod(const MethodCall& message, uint64_t timeout) return connection_->callMethod(message, timeout); } +PendingAsyncCall Proxy::callMethodAsync(const MethodCall& message, async_reply_handler asyncReplyCallback) +{ + return Proxy::callMethodAsync(message, std::move(asyncReplyCallback), /*timeout*/ 0); +} + +Slot Proxy::callMethodAsync(const MethodCall& message, async_reply_handler asyncReplyCallback, return_slot_t) +{ + return Proxy::callMethodAsync(message, std::move(asyncReplyCallback), /*timeout*/ 0, return_slot); +} + PendingAsyncCall Proxy::callMethodAsync(const MethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout) { SDBUS_THROW_ERROR_IF(!message.isValid(), "Invalid async method call message provided", EINVAL); - auto callback = (void*)&Proxy::sdbus_async_reply_handler; - auto callData = std::make_shared(AsyncCalls::CallData{*this, std::move(asyncReplyCallback)}); - auto weakData = std::weak_ptr{callData}; + auto asyncCallInfo = std::make_shared(AsyncCallInfo{ .callback = std::move(asyncReplyCallback) + , .proxy = *this + , .floating = false }); - callData->slot = connection_->callMethod(message, callback, callData.get(), timeout); + asyncCallInfo->slot = connection_->callMethod(message, (void*)&Proxy::sdbus_async_reply_handler, asyncCallInfo.get(), timeout); - pendingAsyncCalls_.addCall(std::move(callData)); + auto asyncCallInfoWeakPtr = std::weak_ptr{asyncCallInfo}; - return {weakData}; + floatingAsyncCallSlots_.push_back(std::move(asyncCallInfo)); + + return {asyncCallInfoWeakPtr}; +} + +Slot Proxy::callMethodAsync(const MethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout, return_slot_t) +{ + SDBUS_THROW_ERROR_IF(!message.isValid(), "Invalid async method call message provided", EINVAL); + + auto asyncCallInfo = std::make_unique(AsyncCallInfo{ .callback = std::move(asyncReplyCallback) + , .proxy = *this + , .floating = true }); + + asyncCallInfo->slot = connection_->callMethod(message, (void*)&Proxy::sdbus_async_reply_handler, asyncCallInfo.get(), timeout); + + return {asyncCallInfo.release(), [](void *ptr){ delete static_cast(ptr); }}; } std::future Proxy::callMethodAsync(const MethodCall& message, with_future_t) @@ -186,7 +216,7 @@ Slot Proxy::registerSignalHandler( const char* interfaceName void Proxy::unregister() { - pendingAsyncCalls_.clear(); + floatingAsyncCallSlots_.clear(); floatingSignalSlots_.clear(); } @@ -207,17 +237,17 @@ Message Proxy::getCurrentlyProcessedMessage() const int Proxy::sdbus_async_reply_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError) { - auto* asyncCallData = static_cast(userData); - assert(asyncCallData != nullptr); - assert(asyncCallData->callback); - auto& proxy = asyncCallData->proxy; + auto* asyncCallInfo = static_cast(userData); + assert(asyncCallInfo != nullptr); + assert(asyncCallInfo->callback); + auto& proxy = asyncCallInfo->proxy; // We are removing the CallData item at the complete scope exit, after the callback has been invoked. // We can't do it earlier (before callback invocation for example), because CallBack data (slot release) // is the synchronization point between callback invocation and Proxy::unregister. SCOPE_EXIT { - proxy.pendingAsyncCalls_.removeCall(asyncCallData); + proxy.floatingAsyncCallSlots_.erase(asyncCallInfo); }; auto message = Message::Factory::create(sdbusMessage, &proxy.connection_->getSdBusInterface()); @@ -227,12 +257,12 @@ int Proxy::sdbus_async_reply_handler(sd_bus_message *sdbusMessage, void *userDat const auto* error = sd_bus_message_get_error(sdbusMessage); if (error == nullptr) { - asyncCallData->callback(std::move(message), {}); + asyncCallInfo->callback(std::move(message), {}); } else { Error exception(Error::Name{error->name}, error->message); - asyncCallData->callback(std::move(message), std::move(exception)); + asyncCallInfo->callback(std::move(message), std::move(exception)); } }, retError); @@ -253,21 +283,64 @@ int Proxy::sdbus_signal_handler(sd_bus_message *sdbusMessage, void *userData, sd return ok ? 0 : -1; } +Proxy::FloatingAsyncCallSlots::~FloatingAsyncCallSlots() +{ + clear(); +} + +void Proxy::FloatingAsyncCallSlots::push_back(std::shared_ptr asyncCallInfo) +{ + std::lock_guard lock(mutex_); + if (!asyncCallInfo->finished) // The call may have finished in the meantime + slots_.emplace_back(std::move(asyncCallInfo)); +} + +void Proxy::FloatingAsyncCallSlots::erase(AsyncCallInfo* info) +{ + std::unique_lock lock(mutex_); + info->finished = true; + auto it = std::find_if(slots_.begin(), slots_.end(), [info](auto const& entry){ return entry.get() == info; }); + if (it != slots_.end()) + { + auto callInfo = std::move(*it); + slots_.erase(it); + lock.unlock(); + + // Releasing call slot pointer acquires global sd-bus mutex. We have to perform the release + // out of the `mutex_' critical section here, because if the `removeCall` is called by some + // thread and at the same time Proxy's async reply handler (which already holds global sd-bus + // mutex) is in progress in a different thread, we get double-mutex deadlock. + } +} + +void Proxy::FloatingAsyncCallSlots::clear() +{ + std::unique_lock lock(mutex_); + auto asyncCallSlots = std::move(slots_); + slots_ = {}; + lock.unlock(); + + // Releasing call slot pointer acquires global sd-bus mutex. We have to perform the release + // out of the `mutex_' critical section here, because if the `clear` is called by some thread + // and at the same time Proxy's async reply handler (which already holds global sd-bus + // mutex) is in progress in a different thread, we get double-mutex deadlock. +} + } namespace sdbus { -PendingAsyncCall::PendingAsyncCall(std::weak_ptr callData) - : callData_(std::move(callData)) +PendingAsyncCall::PendingAsyncCall(std::weak_ptr callInfo) + : callInfo_(std::move(callInfo)) { } void PendingAsyncCall::cancel() { - if (auto ptr = callData_.lock(); ptr != nullptr) + if (auto ptr = callInfo_.lock(); ptr != nullptr) { - auto* callData = static_cast(ptr.get()); - callData->proxy.pendingAsyncCalls_.removeCall(callData); + auto* asyncCallInfo = static_cast(ptr.get()); + asyncCallInfo->proxy.floatingAsyncCallSlots_.erase(asyncCallInfo); // At this point, the callData item is being deleted, leading to the release of the // sd-bus slot pointer. This release locks the global sd-bus mutex. If the async @@ -278,7 +351,7 @@ void PendingAsyncCall::cancel() bool PendingAsyncCall::isPending() const { - return !callData_.expired(); + return !callInfo_.expired(); } } diff --git a/src/Proxy.h b/src/Proxy.h index 24906ce..4ee2047 100644 --- a/src/Proxy.h +++ b/src/Proxy.h @@ -58,8 +58,19 @@ namespace sdbus::internal { MethodCall createMethodCall(const InterfaceName& interfaceName, const MethodName& methodName) override; MethodCall createMethodCall(const char* interfaceName, const char* methodName) override; + MethodReply callMethod(const MethodCall& message) override; MethodReply callMethod(const MethodCall& message, uint64_t timeout) override; - PendingAsyncCall callMethodAsync(const MethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout) override; + PendingAsyncCall callMethodAsync(const MethodCall& message, async_reply_handler asyncReplyCallback) override; + Slot callMethodAsync( const MethodCall& message + , async_reply_handler asyncReplyCallback + , return_slot_t ) override; + PendingAsyncCall callMethodAsync( const MethodCall& message + , async_reply_handler asyncReplyCallback + , uint64_t timeout ) override; + Slot callMethodAsync( const MethodCall& message + , async_reply_handler asyncReplyCallback + , uint64_t timeout + , return_slot_t ) override; std::future callMethodAsync(const MethodCall& message, with_future_t) override; std::future callMethodAsync(const MethodCall& message, uint64_t timeout, with_future_t) override; @@ -105,67 +116,30 @@ namespace sdbus::internal { Slot slot; }; - // We need to keep track of pending async calls. When the proxy is being destructed, we must - // remove all slots of these pending calls, otherwise in case when the connection outlives - // the proxy, we might get async reply handlers invoked for pending async calls after the proxy - // has been destroyed, which is a free ticket into the realm of undefined behavior. - class AsyncCalls + struct AsyncCallInfo + { + async_reply_handler callback; + Proxy& proxy; + Slot slot{}; + bool finished{false}; + bool floating; + }; + + // Container keeping track of pending async calls + class FloatingAsyncCallSlots { public: - struct CallData - { - Proxy& proxy; - async_reply_handler callback; - Slot slot{}; - bool finished{false}; - }; - - ~AsyncCalls() - { - clear(); - } - - void addCall(std::shared_ptr asyncCallData) - { - std::lock_guard lock(mutex_); - if (!asyncCallData->finished) // The call may have finished in the meantime - calls_.emplace_back(std::move(asyncCallData)); - } - - void removeCall(CallData* data) - { - std::unique_lock lock(mutex_); - data->finished = true; - if (auto it = std::find_if(calls_.begin(), calls_.end(), [data](auto const& entry){ return entry.get() == data; }); it != calls_.end()) - { - auto callData = std::move(*it); - calls_.erase(it); - lock.unlock(); - - // Releasing call slot pointer acquires global sd-bus mutex. We have to perform the release - // out of the `mutex_' critical section here, because if the `removeCall` is called by some - // thread and at the same time Proxy's async reply handler (which already holds global sd-bus - // mutex) is in progress in a different thread, we get double-mutex deadlock. - } - } - - void clear() - { - std::unique_lock lock(mutex_); - auto asyncCallSlots = std::move(calls_); - calls_ = {}; - lock.unlock(); - - // Releasing call slot pointer acquires global sd-bus mutex. We have to perform the release - // out of the `mutex_' critical section here, because if the `clear` is called by some thread - // and at the same time Proxy's async reply handler (which already holds global sd-bus - // mutex) is in progress in a different thread, we get double-mutex deadlock. - } + ~FloatingAsyncCallSlots(); + void push_back(std::shared_ptr asyncCallInfo); + void erase(AsyncCallInfo* info); + void clear(); private: std::mutex mutex_; - std::deque> calls_; - } pendingAsyncCalls_; + std::deque> slots_; + }; + + FloatingAsyncCallSlots floatingAsyncCallSlots_; }; } diff --git a/tests/integrationtests/DBusAsyncMethodsTests.cpp b/tests/integrationtests/DBusAsyncMethodsTests.cpp index c868649..0012da1 100644 --- a/tests/integrationtests/DBusAsyncMethodsTests.cpp +++ b/tests/integrationtests/DBusAsyncMethodsTests.cpp @@ -198,6 +198,20 @@ TYPED_TEST(AsyncSdbusTestObject, CancelsPendingAsyncCallOnClientSide) ASSERT_THAT(future.wait_for(300ms), Eq(std::future_status::timeout)); } +TYPED_TEST(AsyncSdbusTestObject, CancelsPendingAsyncCallOnClientSideByDestroyingOwningSlot) +{ + std::promise promise; + auto future = promise.get_future(); + this->m_proxy->installDoOperationClientSideAsyncReplyHandler([&](uint32_t /*res*/, std::optional /*err*/){ promise.set_value(1); }); + + { + auto slot = this->m_proxy->doOperationClientSideAsync(100, sdbus::return_slot); + // Now the slot is destroyed, cancelling the async call + } + + ASSERT_THAT(future.wait_for(300ms), Eq(std::future_status::timeout)); +} + TYPED_TEST(AsyncSdbusTestObject, AnswersThatAsyncCallIsNotPendingAfterItHasBeenCancelled) { std::promise promise; diff --git a/tests/integrationtests/TestProxy.cpp b/tests/integrationtests/TestProxy.cpp index 2eddffa..9248c79 100644 --- a/tests/integrationtests/TestProxy.cpp +++ b/tests/integrationtests/TestProxy.cpp @@ -124,6 +124,17 @@ sdbus::PendingAsyncCall TestProxy::doOperationClientSideAsync(uint32_t param) }); } +Slot TestProxy::doOperationClientSideAsync(uint32_t param, sdbus::return_slot_t) +{ + return getProxy().callMethodAsync("doOperation") + .onInterface(sdbus::test::INTERFACE_NAME) + .withArguments(param) + .uponReplyInvoke([this](std::optional error, uint32_t returnValue) + { + this->onDoOperationReply(returnValue, std::move(error)); + }, sdbus::return_slot); +} + std::future TestProxy::doOperationClientSideAsync(uint32_t param, with_future_t) { return getProxy().callMethodAsync("doOperation") diff --git a/tests/integrationtests/TestProxy.h b/tests/integrationtests/TestProxy.h index 45797fc..cd1cfca 100644 --- a/tests/integrationtests/TestProxy.h +++ b/tests/integrationtests/TestProxy.h @@ -96,6 +96,7 @@ public: void installDoOperationClientSideAsyncReplyHandler(std::function err)> handler); uint32_t doOperationWithTimeout(const std::chrono::microseconds &timeout, uint32_t param); sdbus::PendingAsyncCall doOperationClientSideAsync(uint32_t param); + [[nodiscard]] sdbus::Slot doOperationClientSideAsync(uint32_t param, sdbus::return_slot_t); std::future doOperationClientSideAsync(uint32_t param, with_future_t); std::future doOperationClientSideAsyncOnBasicAPILevel(uint32_t param); std::future doErroneousOperationClientSideAsync(with_future_t);