Play with sync call data

docs/using-sdbus-c++.md

@@ -167,13 +167,17 @@ The following diagram illustrates the major entities in sdbus-c++.
 
 ### Thread safety in sdbus-c++
 
-sdbus-c++ is thread-aware by design. But, in general, it's not thread-safe. At least not in all places. There are situations where sdbus-c++ provides and guarantees API-level thread safety by design. It is safe to do these operations from multiple threads at the same time:
+sdbus-c++ is completely thread-aware by design. Though sdbus-c++ is not thread-safe in general, there are situations where sdbus-c++ provides and guarantees API-level thread safety by design. It is safe to do these operations (operations within the bullet points, not across them) from multiple threads at the same time:
 
-  * Making and destroying `Object`s and `Proxy`s, even on a shared connection that is already running an event loop. Under *making* here is meant a complete atomic sequence of construction, registration of method/signal/property callbacks and export of the `Object`/`Proxy` so it is ready to issue/receive messages. This sequence must be done in one thread.
+  * Making or destroying distinct `Object`/`Proxy` instances simultaneously (even on a shared connection that is running an event loop already, see below). Under *making* here is meant a complete sequence of construction, registration of method/signal/property callbacks and export of the `Object`/`Proxy` so it is ready to issue/receive messages. This sequence must be completely done within the context of one thread.
   * Creating and sending asynchronous method replies on an `Object` instance.
   * Creating and emitting signals on an `Object` instance.
   * Creating and sending method calls (both synchronously and asynchronously) on an `Proxy` instance. (But it's generally better that our threads use their own exclusive instances of proxy, to minimize shared state and contention.)
 
+sdbus-c++ is designed such that all the above operations are thread-safe also on a connection that is running an event loop (usually in a separate thread) at that time. It's an internal thread safety. For example, a signal arrives and is processed by sdbus-c++ even loop at an appropriate `Proxy` instance, while the user is going to destroy that instance in their application thread. The user cannot explicitly control these situations (or they could, but that would be very limiting and cubersome on the API level).
+
+However, other combinations, that the user invokes explicitly from within more threads are NOT thread-safe in sdbus-c++ by design, and the user should make sure by their design that these cases never occur. For example, destroying an `Object` instance in one thread while emitting a signal on it in another thread is not thread-safe. In this specific case, the user should make sure in their application that all threads stop working with a specific instance before a thread proceeds with deleting an instance.
+
 Multiple layers of sdbus-c++ API
 -------------------------------
 

include/sdbus-c++/Message.h

@@ -172,15 +172,17 @@ namespace sdbus {
 
     public:
         MethodCall() = default;
+
         MethodReply send(uint64_t timeout = 0) const;
+        MethodReply sendWithReply(uint64_t timeout = 0) const;
+        void sendWithAsyncReply(void* callback, void* userData, uint64_t timeout = 0) const;
+        MethodReply sendWithNoReply() const;
+
         MethodReply createReply() const;
         MethodReply createErrorReply(const sdbus::Error& error) const;
+
         void dontExpectReply();
         bool doesntExpectReply() const;
-
-    private:
-        MethodReply sendWithReply(uint64_t timeout) const;
-        MethodReply sendWithNoReply() const;
     };
 
     class AsyncMethodCall : public Message

src/Connection.cpp

@@ -34,6 +34,11 @@
 #include <unistd.h>
 #include <poll.h>
 #include <sys/eventfd.h>
+#include <set>
+
+#include <unistd.h>
+#include <sys/syscall.h>
+#define gettid() syscall(SYS_gettid)
 
 namespace sdbus { namespace internal {
 
@@ -86,6 +91,10 @@ std::string Connection::getUniqueName() const
 
 void Connection::enterProcessingLoop()
 {
+    loopThreadId_ = std::this_thread::get_id();
+
+    std::lock_guard<std::mutex> guard(loopMutex_);
+
     while (true)
     {
         auto processed = processPendingRequest();
@@ -96,6 +105,8 @@ void Connection::enterProcessingLoop()
         if (!success)
             break; // Exit processing loop
     }
+
+    loopThreadId_ = std::thread::id{};
 }
 
 void Connection::enterProcessingLoopAsync()
@@ -288,6 +299,35 @@ SlotPtr Connection::registerSignalHandler( const std::string& objectPath
     return {slot, [this](void *slot){ iface_->sd_bus_slot_unref((sd_bus_slot*)slot); }};
 }
 
+MethodReply Connection::tryCallMethodSynchronously(const MethodCall& message, uint64_t timeout)
+{
+    auto loopThreadId = loopThreadId_.load(std::memory_order_relaxed);
+
+    // Is the loop not yet on? => Go make synchronous call
+    while (loopThreadId == std::thread::id{})
+    {
+        // Did the loop begin in the meantime? Or try_lock() failed spuriously?
+        if (!loopMutex_.try_lock())
+        {
+            loopThreadId = loopThreadId_.load(std::memory_order_relaxed);
+            continue;
+        }
+
+        // Synchronous D-Bus call
+        std::lock_guard<std::mutex> guard(loopMutex_, std::adopt_lock);
+        return message.send(timeout);
+    }
+
+    // Is the loop on and we are in the same thread? => Go for synchronous call
+    if (loopThreadId == std::this_thread::get_id())
+    {
+        assert(!loopMutex_.try_lock());
+        return message.send(timeout);
+    }
+
+    return MethodReply{};
+}
+
 Connection::BusPtr Connection::openBus(const BusFactory& busFactory)
 {
     sd_bus* bus{};
@@ -332,6 +372,8 @@ void Connection::joinWithProcessingLoop()
 {
     if (asyncLoopThread_.joinable())
         asyncLoopThread_.join();
+//    if (asyncLoopThread2_.joinable())
+//        asyncLoopThread2_.join();
 }
 
 bool Connection::processPendingRequest()
@@ -356,8 +398,13 @@ bool Connection::waitForNextRequest()
     struct pollfd fds[] = {{sdbusPollData.fd, sdbusPollData.events, 0}, {loopExitFd_.fd, POLLIN, 0}};
     auto fdsCount = sizeof(fds)/sizeof(fds[0]);
 
+    //printf("Thread %d: Going to poll %p\n", gettid(), this);
+
     auto timeout = sdbusPollData.timeout_usec == (uint64_t) -1 ? (uint64_t)-1 : (sdbusPollData.timeout_usec+999)/1000;
     auto r = poll(fds, fdsCount, timeout);
+    //auto r = ppoll(fds, fdsCount, nullptr, nullptr);
+
+    //printf("Thread %d: Poll woken up %p\n", gettid(), this);
 
     if (r < 0 && errno == EINTR)
         return true; // Try again
@@ -367,6 +414,7 @@ bool Connection::waitForNextRequest()
     if (fds[1].revents & POLLIN)
     {
         clearExitNotification();
+        //printf("Thread %d: Exiting %p\n", gettid(), this);
         return false;
     }
 

src/Connection.h

@@ -37,6 +37,8 @@
 #include <thread>
 #include <string>
 #include <vector>
+#include <atomic>
+#include <mutex>
 
 namespace sdbus { namespace internal {
 
@@ -103,6 +105,8 @@ namespace sdbus { namespace internal {
                                      , sd_bus_message_handler_t callback
                                      , void* userData ) override;
 
+        MethodReply tryCallMethodSynchronously(const MethodCall& message, uint64_t timeout) override;
+
     private:
         using BusFactory = std::function<int(sd_bus**)>;
         using BusPtr = std::unique_ptr<sd_bus, std::function<sd_bus*(sd_bus*)>>;
@@ -130,6 +134,8 @@ namespace sdbus { namespace internal {
         std::unique_ptr<ISdBus> iface_;
         BusPtr bus_;
         std::thread asyncLoopThread_;
+        std::atomic<std::thread::id> loopThreadId_;
+        std::mutex loopMutex_;
         LoopExitEventFd loopExitFd_;
     };
 

src/IConnection.h

@@ -91,6 +91,8 @@ namespace internal {
 
         virtual void enterProcessingLoopAsync() = 0;
         virtual void leaveProcessingLoop() = 0;
+
+        virtual MethodReply tryCallMethodSynchronously(const MethodCall& message, uint64_t timeout) = 0;
     };
 
 }

src/Message.cpp

@@ -652,6 +652,14 @@ MethodReply MethodCall::sendWithReply(uint64_t timeout) const
     return Factory::create<MethodReply>(sdbusReply, sdbus_, adopt_message);
 }
 
+// TODO: Consider merging MethodCall with AsyncMethodCall  (provide owning boolean parameter for no Slot),
+// return sendWithReply and sendWithNoReply back as private methods.
+void MethodCall::sendWithAsyncReply(void* callback, void* userData, uint64_t timeout) const
+{
+    auto r = sdbus_->sd_bus_call_async(nullptr, nullptr, (sd_bus_message*)msg_, (sd_bus_message_handler_t)callback, userData, timeout);
+    SDBUS_THROW_ERROR_IF(r < 0, "Failed to call method with asynchronous reply", -r);
+}
+
 MethodReply MethodCall::sendWithNoReply() const
 {
     auto r = sdbus_->sd_bus_send(nullptr, (sd_bus_message*)msg_, nullptr);

src/Proxy.cpp

@@ -26,6 +26,7 @@
 
 #include "Proxy.h"
 #include "IConnection.h"
+#include "ISdBus.h"
 #include "MessageUtils.h"
 #include "sdbus-c++/Message.h"
 #include "sdbus-c++/IConnection.h"
@@ -35,6 +36,12 @@
 #include <cassert>
 #include <chrono>
 #include <thread>
+#include <future>
+#include <utility>
+
+#include <unistd.h>
+#include <sys/syscall.h>
+#define gettid() syscall(SYS_gettid)
 
 namespace sdbus { namespace internal {
 
@@ -71,9 +78,33 @@ AsyncMethodCall Proxy::createAsyncMethodCall(const std::string& interfaceName, c
 
 MethodReply Proxy::callMethod(const MethodCall& message, uint64_t timeout)
 {
+    // Sending method call synchronously is the only operation that blocks, waiting for the method
+    // reply message among the incoming message on the sd-bus connection socket. But typically there
+    // already is somebody that generally handles incoming D-Bus messages -- the connection event loop
+    // running typically in its own thread. We have to avoid polling on socket from several threads.
+    // So we have to branch here: either we are within the context of the event loop thread, then we
+    // can send the message simply, and blockingly, via sd_bus_call. Or we are in another thread, then
+    // we can perform the send operation of the method call message from here (because that is thread-
+    // safe like all other sd-bus API accesses), but the incoming reply we have to get through the event
+    // loop thread, because this should be the only rightful listener on the sd-bus connection socket.
+    // So, technically, we use async means to wait here for reply received by the event loop thread.
+
     SDBUS_THROW_ERROR_IF(!message.isValid(), "Invalid method call message provided", EINVAL);
 
-    return message.send(timeout);
+    //return message.send(timeout);
+
+    // If we don't need to wait for any reply, we can send the message now irrespective of the context
+    if (message.doesntExpectReply())
+        return message.sendWithNoReply();
+
+    // If we are in the context of event loop thread, we can send the D-Bus call synchronously
+    // and wait blockingly for the reply, because we are the exclusive listeners on the socket
+    auto reply = connection_->tryCallMethodSynchronously(message, timeout);
+    if (reply.isValid())
+        return reply;
+
+    // Otherwise we send the call asynchronously and do blocking wait for the reply from the event loop thread
+    return callMethodWithAsyncReplyBlocking(message, timeout);
 }
 
 void Proxy::callMethod(const AsyncMethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout)
@@ -88,6 +119,66 @@ void Proxy::callMethod(const AsyncMethodCall& message, async_reply_handler async
     pendingAsyncCalls_.addCall(callData->slot.get(), std::move(callData));
 }
 
+MethodReply Proxy::callMethodWithAsyncReplyBlocking(const MethodCall& message, uint64_t timeout)
+{
+    // VARIANT 1: THREAD LOCAL COND VAR
+    message.sendWithAsyncReply((void*)&Proxy::sdbus_sync_reply_handler, this, timeout);
+
+    auto& syncCallReplyData = getSyncCallReplyData();
+    std::unique_lock<std::mutex> lock(syncCallReplyData.mutex);
+    syncCallReplyData.cond.wait(lock, [&syncCallReplyData](){ return syncCallReplyData.arrived; });
+
+    syncCallReplyData.arrived = false;
+    if (syncCallReplyData.error)
+        throw *syncCallReplyData.error;
+
+    return std::move(syncCallReplyData.reply);
+
+    // VARIANT 2: USING SPECIAL APPROACH, A. PROMISE/FUTURE
+//    std::promise<MethodReply> result;
+//    auto future = result.get_future();
+
+//    auto callback = (void*)&Proxy::sdbus_sync_reply_handler;
+//    auto data = std::make_pair(std::ref(result), std::ref(connection_->getSdBusInterface()));
+//    message.sendWithAsyncReply(callback, &data, timeout);
+
+//    //printf("Thread %d: Proxy going to wait on future\n", gettid());
+//    MethodReply r = future.get();
+//    //printf("Thread %d: Proxy woken up on future\n", gettid());
+//    return r;
+
+    // VARIANT 3: USING CLASSIC ASYNC APPROACH
+    // TODO: Try with thread local std::function
+//    /*thread_local*/ SyncCallReplyData syncCallReplyData;
+
+//    async_reply_handler asyncReplyCallback = [&syncCallReplyData](MethodReply& reply, const Error* error)
+//    {
+//        std::unique_lock<std::mutex> lock(syncCallReplyData.mutex);
+
+//        //syncCallReplyData.error = nullptr;
+//        if (error == nullptr)
+//            syncCallReplyData.reply = std::move(reply);
+//        else
+//            syncCallReplyData.error = std::make_unique<Error>(*error);
+//        syncCallReplyData.arrived = true;
+
+//        lock.unlock();
+//        syncCallReplyData.cond.notify_one();
+//    };
+
+//    AsyncCalls::CallData callData{*this, std::move(asyncReplyCallback), {}};
+//    message.sendWithAsyncReply((void*)&Proxy::sdbus_async_reply_handler, &callData, timeout);
+
+//    std::unique_lock<std::mutex> lock(syncCallReplyData.mutex);
+//    syncCallReplyData.cond.wait(lock, [&syncCallReplyData](){ return syncCallReplyData.arrived; });
+
+//    //syncCallReplyData.arrived = false;
+//    if (syncCallReplyData.error)
+//        throw *syncCallReplyData.error;
+
+//    return std::move(syncCallReplyData.reply);
+}
+
 void Proxy::registerSignalHandler( const std::string& interfaceName
                                  , const std::string& signalName
                                  , signal_handler signalHandler )
@@ -122,7 +213,7 @@ void Proxy::registerSignalHandlers(sdbus::internal::IConnection& connection)
             slot = connection.registerSignalHandler( objectPath_
                                                    , interfaceName
                                                    , signalName
-                                                   , &Proxy::sdbus_signal_callback
+                                                   , &Proxy::sdbus_signal_handler
                                                    , this );
         }
     }
@@ -134,6 +225,13 @@ void Proxy::unregister()
     interfaces_.clear();
 }
 
+Proxy::SyncCallReplyData& Proxy::getSyncCallReplyData()
+{
+    thread_local SyncCallReplyData syncCallReplyData;
+    return syncCallReplyData;
+}
+
+// Handler for D-Bus method replies of fully asynchronous D-Bus method calls
 int Proxy::sdbus_async_reply_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error */*retError*/)
 {
     auto* asyncCallData = static_cast<AsyncCalls::CallData*>(userData);
@@ -141,7 +239,7 @@ int Proxy::sdbus_async_reply_handler(sd_bus_message *sdbusMessage, void *userDat
     assert(asyncCallData->callback);
     auto& proxy = asyncCallData->proxy;
 
-    SCOPE_EXIT{ proxy.pendingAsyncCalls_.removeCall(asyncCallData->slot.get()); };
+    SCOPE_EXIT{ if (asyncCallData->slot) proxy.pendingAsyncCalls_.removeCall(asyncCallData->slot.get()); };
 
     auto message = Message::Factory::create<MethodReply>(sdbusMessage, &proxy.connection_->getSdBusInterface());
 
@@ -159,7 +257,60 @@ int Proxy::sdbus_async_reply_handler(sd_bus_message *sdbusMessage, void *userDat
     return 1;
 }
 
-int Proxy::sdbus_signal_callback(sd_bus_message *sdbusMessage, void *userData, sd_bus_error */*retError*/)
+// Handler for D-Bus method replies of synchronous D-Bus method calls done out of event loop thread context
+int Proxy::sdbus_sync_reply_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error */*retError*/)
+{
+    //printf("Thread %d: Proxy::sdbus_quasi_sync_reply_handler 1\n", gettid());
+
+    // VARIANT 1: THREAD LOCAL COND VAR
+    assert(userData != nullptr);
+
+    auto& syncCallReplyData = getSyncCallReplyData();
+    std::unique_lock<std::mutex> lock(syncCallReplyData.mutex);
+
+    const auto* error = sd_bus_message_get_error(sdbusMessage);
+    if (error == nullptr)
+    {
+        auto* proxy = static_cast<Proxy*>(userData);
+        auto message = Message::Factory::create<MethodReply>(sdbusMessage, &proxy->connection_->getSdBusInterface());
+        syncCallReplyData.reply = std::move(message);
+        syncCallReplyData.error = nullptr;
+    }
+    else
+    {
+        auto exception = std::make_unique<sdbus::Error>(error->name, error->message);
+        syncCallReplyData.error = std::move(exception);
+    }
+    syncCallReplyData.arrived = true;
+
+    lock.unlock();
+    syncCallReplyData.cond.notify_one();
+
+    // VARIANT 2: Promise/Future
+//    assert(userData != nullptr);
+//    auto* data = static_cast<std::pair<std::promise<MethodReply>&, ISdBus&>*>(userData);
+//    auto& promise = data->first;
+//    auto& sdBus = data->second;
+
+//    auto message = Message::Factory::create<MethodReply>(sdbusMessage, &sdBus);
+
+//    const auto* error = sd_bus_message_get_error(sdbusMessage);
+//    if (error == nullptr)
+//    {
+//        //printf("Thread %d: Proxy::sdbus_quasi_sync_reply_handler 2\n", gettid());
+//        promise.set_value(std::move(message));
+//    }
+//    else
+//    {
+//        sdbus::Error exception(error->name, error->message);
+//        promise.set_exception(std::make_exception_ptr(exception));
+//    }
+
+    return 1;
+}
+
+// Handler for signals coming from the D-Bus object
+int Proxy::sdbus_signal_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error */*retError*/)
 {
     auto* proxy = static_cast<Proxy*>(userData);
     assert(proxy != nullptr);

src/Proxy.h

@@ -35,6 +35,7 @@
 #include <map>
 #include <unordered_map>
 #include <mutex>
+#include <condition_variable>
 
 namespace sdbus {
 namespace internal {
@@ -62,9 +63,13 @@ namespace internal {
         void unregister() override;
 
     private:
+        struct SyncCallReplyData;
+        static SyncCallReplyData& getSyncCallReplyData();
+        MethodReply callMethodWithAsyncReplyBlocking(const MethodCall& message, uint64_t timeout);
         void registerSignalHandlers(sdbus::internal::IConnection& connection);
         static int sdbus_async_reply_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError);
-        static int sdbus_signal_callback(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError);
+        static int sdbus_sync_reply_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError);
+        static int sdbus_signal_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError);
 
     private:
         std::unique_ptr< sdbus::internal::IConnection
@@ -130,6 +135,15 @@ namespace internal {
             std::unordered_map<void*, std::unique_ptr<CallData>> calls_;
             std::mutex mutex_;
         } pendingAsyncCalls_;
+
+        struct SyncCallReplyData
+        {
+            std::mutex mutex;
+            std::condition_variable cond;
+            bool arrived{};
+            MethodReply reply;
+            std::unique_ptr<Error> error;
+        };
     };
 
 }}

src/SdBus.cpp

@@ -27,9 +27,58 @@
 
 #include "SdBus.h"
 #include <sdbus-c++/Error.h>
+#include "ScopeGuard.h"
 
 namespace sdbus { namespace internal {
 
+sd_bus_message* createPlainMessage()
+{
+    int r;
+
+    // All references to the bus (like created messages) must not outlive this thread (because messages refer to sdbus
+    // which is thread-local, and because BusReferenceKeeper below destroys the bus at thread exit).
+    // A more flexible solution would be that the caller would already provide an ISdBus reference as a parameter.
+    // Variant is one of the callers. This means Variant could no more be created in a stand-alone way, but
+    // through a factory of some existing facility (Object, Proxy, Connection...).
+    // TODO: Consider this alternative of creating Variant, it may live next to the current one. This function would
+    // get IConnection* parameter and IConnection would provide createPlainMessage factory (just like it already
+    // provides e.g. createMethodCall). If this parameter were null, the current mechanism would be used.
+
+    thread_local internal::SdBus sdbus;
+
+    sd_bus* bus{};
+    SCOPE_EXIT{ sd_bus_unref(bus); };
+    r = sd_bus_default_system(&bus);
+    SDBUS_THROW_ERROR_IF(r < 0, "Failed to get default system bus", -r);
+
+    thread_local struct BusReferenceKeeper
+    {
+        explicit BusReferenceKeeper(sd_bus* bus) : bus_(sd_bus_ref(bus)) { sd_bus_flush(bus_); }
+        ~BusReferenceKeeper() { sd_bus_flush_close_unref(bus_); }
+        sd_bus* bus_{};
+    } busReferenceKeeper{bus};
+
+    // Shelved here as handy thing for potential future tracing purposes:
+    //#include <unistd.h>
+    //#include <sys/syscall.h>
+    //#define gettid() syscall(SYS_gettid)
+    //printf("createPlainMessage: sd_bus*=[%p], n_ref=[%d], TID=[%d]\n", bus, *(unsigned*)bus, gettid());
+
+    sd_bus_message* sdbusMsg{};
+    r = sd_bus_message_new(bus, &sdbusMsg, _SD_BUS_MESSAGE_TYPE_INVALID);
+    SDBUS_THROW_ERROR_IF(r < 0, "Failed to create a new message", -r);
+
+    r = sd_bus_message_append_basic(sdbusMsg, SD_BUS_TYPE_STRING, "This is item.c_str()");
+    SDBUS_THROW_ERROR_IF(r < 0, "Failed to serialize a string value", -r);
+
+    r = ::sd_bus_message_seal(sdbusMsg, 1, 0);
+    SDBUS_THROW_ERROR_IF(r < 0, "Failed to seal the reply", -r);
+
+    return sdbusMsg;
+}
+
+static auto g_sdbusMessage = createPlainMessage();
+
 sd_bus_message* SdBus::sd_bus_message_ref(sd_bus_message *m)
 {
     std::unique_lock<std::recursive_mutex> lock(sdbusMutex_);
@@ -55,14 +104,38 @@ int SdBus::sd_bus_call(sd_bus *bus, sd_bus_message *m, uint64_t usec, sd_bus_err
 {
     std::unique_lock<std::recursive_mutex> lock(sdbusMutex_);
 
-    return ::sd_bus_call(bus, m, usec, ret_error, reply);
+    //return ::sd_bus_call(bus, m, usec, ret_error, reply);
+    ::sd_bus_message_ref(g_sdbusMessage);
+
+    *reply = g_sdbusMessage;
+
+    return 1;
 }
 
 int SdBus::sd_bus_call_async(sd_bus *bus, sd_bus_slot **slot, sd_bus_message *m, sd_bus_message_handler_t callback, void *userdata, uint64_t usec)
 {
     std::unique_lock<std::recursive_mutex> lock(sdbusMutex_);
 
-    return ::sd_bus_call_async(bus, slot, m, callback, userdata, usec);
+    //return ::sd_bus_call_async(bus, slot, m, callback, userdata, usec);
+
+//    auto r = ::sd_bus_message_seal(m, 1, 0);
+//    SDBUS_THROW_ERROR_IF(r < 0, "Failed to seal the message", -r);
+
+//    sd_bus_message* sdbusReply{};
+//    r = this->sd_bus_message_new_method_return(m, &sdbusReply);
+//    SDBUS_THROW_ERROR_IF(r < 0, "Failed to create method reply", -r);
+
+//    r = sd_bus_message_append_basic(sdbusReply, SD_BUS_TYPE_STRING, "This is item.c_str()");
+//    SDBUS_THROW_ERROR_IF(r < 0, "Failed to serialize a string value", -r);
+
+//    r = ::sd_bus_message_seal(sdbusReply, 1, 0);
+//    SDBUS_THROW_ERROR_IF(r < 0, "Failed to seal the reply", -r);
+
+    ::sd_bus_message_ref(g_sdbusMessage);
+
+    callback(g_sdbusMessage, userdata, nullptr);
+
+    return 1;
 }
 
 int SdBus::sd_bus_message_new_method_call(sd_bus *bus, sd_bus_message **m, const char *destination, const char *path, const char *interface, const char *member)

src/SdBus.h

@@ -33,7 +33,7 @@
 
 namespace sdbus { namespace internal {
 
-class SdBus final : public ISdBus
+class SdBus /*final*/ : public ISdBus
 {
 public:
     virtual sd_bus_message* sd_bus_message_ref(sd_bus_message *m) override;

tests/integrationtests/AdaptorAndProxy_test.cpp

@@ -48,6 +48,7 @@
 using ::testing::Eq;
 using ::testing::DoubleEq;
 using ::testing::Gt;
+using ::testing::AnyOf;
 using ::testing::ElementsAre;
 using ::testing::SizeIs;
 using namespace std::chrono_literals;
@@ -61,12 +62,13 @@ public:
     static void SetUpTestCase()
     {
         s_connection->requestName(INTERFACE_NAME);
-        s_connection->enterProcessingLoopAsync();
+        m_t = std::thread([](){s_connection->enterProcessingLoop();});
     }
 
     static void TearDownTestCase()
     {
         s_connection->leaveProcessingLoop();
+        m_t.join();
         s_connection->releaseName(INTERFACE_NAME);
     }
 
@@ -88,7 +90,9 @@ private:
     void SetUp() override
     {
         m_adaptor = std::make_unique<TestingAdaptor>(*s_connection);
+        //printf("PROXY BEGIN!\n");
         m_proxy = std::make_unique<TestingProxy>(INTERFACE_NAME, OBJECT_PATH);
+        //printf("PROXY END!\n");
         std::this_thread::sleep_for(50ms); // Give time for the proxy to start listening to signals
     }
 
@@ -103,9 +107,13 @@ public:
 
     std::unique_ptr<TestingAdaptor> m_adaptor;
     std::unique_ptr<TestingProxy> m_proxy;
+
+    static std::thread m_t;
 };
 
 std::unique_ptr<sdbus::IConnection> AdaptorAndProxyFixture::s_connection = sdbus::createSystemBusConnection();
+std::thread AdaptorAndProxyFixture::m_t;
+
 }
 
 /*-------------------------------------*/
@@ -247,8 +255,38 @@ TEST_F(SdbusTestObject, ThrowsTimeoutErrorWhenMethodTimesOut)
     }
     catch (const sdbus::Error& e)
     {
-        ASSERT_THAT(e.getName(), Eq("org.freedesktop.DBus.Error.Timeout"));
-        ASSERT_THAT(e.getMessage(), Eq("Connection timed out"));
+        ASSERT_THAT(e.getName(), AnyOf("org.freedesktop.DBus.Error.Timeout", "org.freedesktop.DBus.Error.NoReply"));
+        ASSERT_THAT(e.getMessage(), AnyOf("Connection timed out", "Method call timed out"));
+    }
+    catch(...)
+    {
+        FAIL() << "Expected sdbus::Error exception";
+    }
+}
+
+TEST_F(SdbusTestObject, ThrowsTimeoutErrorWhenClientSideAsyncMethodTimesOut)
+{
+    try
+    {
+        std::promise<uint32_t> promise;
+        auto future = promise.get_future();
+        m_proxy->installDoOperationClientSideAsyncReplyHandler([&](uint32_t res, const sdbus::Error* err)
+        {
+            if (err == nullptr)
+                promise.set_value(res);
+            else
+                promise.set_exception(std::make_exception_ptr(*err));
+        });
+
+        m_proxy->doOperationClientSideAsyncWith500msTimeout(1000); // The operation will take 1s, but the timeout is 500ms, so we should time out
+        future.get(), Eq(100);
+
+        FAIL() << "Expected sdbus::Error exception";
+    }
+    catch (const sdbus::Error& e)
+    {
+        ASSERT_THAT(e.getName(), AnyOf("org.freedesktop.DBus.Error.Timeout", "org.freedesktop.DBus.Error.NoReply"));
+        ASSERT_THAT(e.getMessage(), AnyOf("Connection timed out", "Method call timed out"));
     }
     catch(...)
     {
@@ -392,6 +430,14 @@ TEST_F(SdbusTestObject, FailsCallingMethodOnNonexistentObject)
     ASSERT_THROW(proxy.getInt(), sdbus::Error);
 }
 
+TEST_F(SdbusTestObject, ReceivesTwoSignalsWhileMakingMethodCall)
+{
+    m_proxy->emitTwoSimpleSignals();
+
+    ASSERT_TRUE(waitUntil(m_proxy->m_gotSimpleSignal));
+    ASSERT_TRUE(waitUntil(m_proxy->m_gotSignalWithMap));
+}
+
 #if LIBSYSTEMD_VERSION>=240
 TEST_F(SdbusTestObject, CanSetGeneralMethodTimeoutWithLibsystemdVersionGreaterThan239)
 {

tests/integrationtests/TestingAdaptor.h

@@ -197,6 +197,13 @@ protected:
         throw sdbus::createError(1, "A test error occurred");
     }
 
+
+    void emitTwoSimpleSignals() override
+    {
+        emitSimpleSignal();
+        emitSignalWithMap({});
+    }
+
     std::string state()
     {
         return m_state;

tests/integrationtests/TestingProxy.h

@@ -30,6 +30,10 @@
 #include "proxy-glue.h"
 #include <atomic>
 
+#include <unistd.h>
+#include <sys/syscall.h>
+#define gettid() syscall(SYS_gettid)
+
 class TestingProxy : public sdbus::ProxyInterfaces< ::testing_proxy
                                                   , sdbus::Peer_proxy
                                                   , sdbus::Introspectable_proxy
@@ -108,6 +112,7 @@ protected:
 
 //private:
 public: // for tests
+    int m_SimpleSignals = 0;
     std::atomic<bool> m_gotSimpleSignal{false};
     std::atomic<bool> m_gotSignalWithMap{false};
     std::map<int32_t, std::string> m_mapFromSignal;

tests/integrationtests/adaptor-glue.h

@@ -106,6 +106,8 @@ protected:
 
         object_.registerMethod("doPrivilegedStuff").onInterface(INTERFACE_NAME).implementedAs([](){}).markAsPrivileged();
 
+        object_.registerMethod("emitTwoSimpleSignals").onInterface(INTERFACE_NAME).implementedAs([this](){ this->emitTwoSimpleSignals(); });
+
         // registration of signals is optional, it is useful because of introspection
         object_.registerSignal("simpleSignal").onInterface(INTERFACE_NAME).markAsDeprecated();
         object_.registerSignal("signalWithMap").onInterface(INTERFACE_NAME).withParameters<std::map<int32_t, std::string>>();
@@ -168,6 +170,7 @@ protected:
     virtual sdbus::UnixFd getUnixFd() const  = 0;
     virtual ComplexType getComplex() const = 0;
     virtual void throwError() const = 0;
+    virtual void emitTwoSimpleSignals() = 0;
 
     virtual std::string state() = 0;
     virtual uint32_t action() = 0;

tests/integrationtests/proxy-glue.h

@@ -56,6 +56,11 @@ protected:
     virtual void onDoOperationReply(uint32_t returnValue, const sdbus::Error* error) = 0;
 
 public:
+    void emitTwoSimpleSignals()
+    {
+        object_.callMethod("emitTwoSimpleSignals").onInterface(INTERFACE_NAME);
+    }
+
     void noArgNoReturn()
     {
         object_.callMethod("noArgNoReturn").onInterface(INTERFACE_NAME);
@@ -172,6 +177,19 @@ public:
                                 });
     }
 
+    void doOperationClientSideAsyncWith500msTimeout(uint32_t param)
+    {
+        using namespace std::chrono_literals;
+        object_.callMethodAsync("doOperation")
+               .onInterface(INTERFACE_NAME)
+               .withTimeout(500000us)
+               .withArguments(param)
+               .uponReplyInvoke([this](const sdbus::Error* error, uint32_t returnValue)
+                                {
+                                    this->onDoOperationReply(returnValue, error);
+                                });
+    }
+
     sdbus::Signature getSignature()
     {
         sdbus::Signature result;

tests/perftests/client.cpp

@@ -34,9 +34,32 @@
 #include <cassert>
 #include <algorithm>
 #include <iostream>
+//#include "SdBus.h"
 
 using namespace std::chrono_literals;
 
+//class MySdBus : public sdbus::internal::SdBus
+//{
+//public:
+//    virtual int sd_bus_call_async(sd_bus *bus, sd_bus_slot **slot, sd_bus_message *m, sd_bus_message_handler_t callback, void *userdata, uint64_t usec) override
+//    {
+//        sd_bus_message* sdbusReply{};
+//        auto r = this->sd_bus_message_new_method_return(m, &sdbusReply);
+//        SDBUS_THROW_ERROR_IF(r < 0, "Failed to create method reply", -r);
+
+//        callback(sdbusReply, userdata, nullptr);
+
+//        return 1;
+//    }
+//};
+
+namespace sdbus
+{
+    PlainMessage createPlainMessage();
+}
+
+uint64_t totalDuration = 0;
+
 class PerftestProxy : public sdbus::ProxyInterfaces<org::sdbuscpp::perftests_proxy>
 {
 public:
@@ -66,7 +89,9 @@ protected:
         else if (counter == m_msgCount)
         {
             auto stopTime = std::chrono::steady_clock::now();
-            std::cout << "Received " << m_msgCount << " signals in: " << std::chrono::duration_cast<std::chrono::milliseconds>(stopTime - startTime).count() << " ms" << std::endl;
+            auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(stopTime - startTime).count();
+            totalDuration += duration;
+            std::cout << "Received " << m_msgCount << " signals in: " << duration << " ms" << std::endl;
             counter = 0;
         }
     }
@@ -98,12 +123,13 @@ int main(int /*argc*/, char */*argv*/[])
 {
     const char* destinationName = "org.sdbuscpp.perftests";
     const char* objectPath = "/org/sdbuscpp/perftests";
-    PerftestProxy client(destinationName, objectPath);
+    //PerftestProxy client(destinationName, objectPath);
 
     const unsigned int repetitions{20};
-    unsigned int msgCount = 1000;
+    unsigned int msgCount = 100000;
     unsigned int msgSize{};
 
+    /*
     msgSize = 20;
     std::cout << "** Measuring signals of size " << msgSize << " bytes (" << repetitions << " repetitions)..." << std::endl << std::endl;
     client.m_msgCount = msgCount; client.m_msgSize = msgSize;
@@ -114,6 +140,9 @@ int main(int /*argc*/, char */*argv*/[])
         std::this_thread::sleep_for(1000ms);
     }
 
+    std::cout << "AVERAGE: " << (totalDuration/repetitions) << " ms" << std::endl;
+    totalDuration = 0;
+
     msgSize = 1000;
     std::cout << std::endl << "** Measuring signals of size " << msgSize << " bytes (" << repetitions << " repetitions)..." << std::endl << std::endl;
     client.m_msgCount = msgCount; client.m_msgSize = msgSize;
@@ -124,47 +153,116 @@ int main(int /*argc*/, char */*argv*/[])
         std::this_thread::sleep_for(1000ms);
     }
 
+    std::cout << "AVERAGE: " << (totalDuration/repetitions) << " ms" << std::endl;
+    totalDuration = 0;
+    */
+
+//    msgSize = 20;
+//    std::cout << std::endl << "** Measuring method calls of size " << msgSize << " bytes (" << repetitions << " repetitions)..." << std::endl << std::endl;
+//    for (unsigned int r = 0; r < repetitions; ++r)
+//    {
+//        auto str1 = createRandomString(msgSize/2);
+//        auto str2 = createRandomString(msgSize/2);
+
+//        auto startTime = std::chrono::steady_clock::now();
+//        for (unsigned int i = 0; i < msgCount; i++)
+//        {
+//            auto result = client.concatenateTwoStrings(str1, str2);
+
+//            assert(result.size() == str1.size() + str2.size());
+//            assert(result.size() == msgSize);
+//        }
+//        auto stopTime = std::chrono::steady_clock::now();
+//        auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(stopTime - startTime).count();
+//        totalDuration += duration;
+//        std::cout << "Called " << msgCount << " methods in: " << duration << " ms" << std::endl;
+
+//        std::this_thread::sleep_for(1000ms);
+//    }
+
+//    std::cout << "AVERAGE: " << (totalDuration/repetitions) << " ms" << std::endl;
+//    totalDuration = 0;
+
+//    msgSize = 1000;
+//    std::cout << std::endl << "** Measuring method calls of size " << msgSize << " bytes (" << repetitions << " repetitions)..." << std::endl << std::endl;
+//    for (unsigned int r = 0; r < repetitions; ++r)
+//    {
+//        auto str1 = createRandomString(msgSize/2);
+//        auto str2 = createRandomString(msgSize/2);
+
+//        auto startTime = std::chrono::steady_clock::now();
+//        for (unsigned int i = 0; i < msgCount; i++)
+//        {
+//            auto result = client.concatenateTwoStrings(str1, str2);
+
+//            assert(result.size() == str1.size() + str2.size());
+//            assert(result.size() == msgSize);
+//        }
+//        auto stopTime = std::chrono::steady_clock::now();
+//        auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(stopTime - startTime).count();
+//        totalDuration += duration;
+//        std::cout << "Called " << msgCount << " methods in: " << duration << " ms" << std::endl;
+
+//        std::this_thread::sleep_for(1000ms);
+//    }
+
+//    std::cout << "AVERAGE: " << (totalDuration/repetitions) << " ms" << std::endl;
+//    totalDuration = 0;
+
+    auto proxy = sdbus::createProxy(destinationName, objectPath);
+    auto msg = proxy->createMethodCall("org.sdbuscpp.perftests", "concatenateTwoStrings");
+    //auto msg = sdbus::createPlainMessage();
+    msg.seal();
+
     msgSize = 20;
     std::cout << std::endl << "** Measuring method calls of size " << msgSize << " bytes (" << repetitions << " repetitions)..." << std::endl << std::endl;
     for (unsigned int r = 0; r < repetitions; ++r)
     {
-        auto str1 = createRandomString(msgSize/2);
-        auto str2 = createRandomString(msgSize/2);
-
         auto startTime = std::chrono::steady_clock::now();
         for (unsigned int i = 0; i < msgCount; i++)
         {
-            auto result = client.concatenateTwoStrings(str1, str2);
+            //auto result = client.concatenateTwoStrings(str1, str2);
+            proxy->callMethod(msg);
 
-            assert(result.size() == str1.size() + str2.size());
-            assert(result.size() == msgSize);
+            //assert(result.size() == str1.size() + str2.size());
+            //assert(result.size() == msgSize);
         }
         auto stopTime = std::chrono::steady_clock::now();
-        std::cout << "Called " << msgCount << " methods in: " << std::chrono::duration_cast<std::chrono::milliseconds>(stopTime - startTime).count() << " ms" << std::endl;
+        auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(stopTime - startTime).count();
+        totalDuration += duration;
+        std::cout << "Called " << msgCount << " methods in: " << duration << " ms" << std::endl;
 
-        std::this_thread::sleep_for(1000ms);
+        std::this_thread::sleep_for(100ms);
     }
 
-    msgSize = 1000;
-    std::cout << std::endl << "** Measuring method calls of size " << msgSize << " bytes (" << repetitions << " repetitions)..." << std::endl << std::endl;
-    for (unsigned int r = 0; r < repetitions; ++r)
-    {
-        auto str1 = createRandomString(msgSize/2);
-        auto str2 = createRandomString(msgSize/2);
+    std::cout << "AVERAGE: " << (totalDuration/repetitions) << " ms" << std::endl;
+    totalDuration = 0;
 
-        auto startTime = std::chrono::steady_clock::now();
-        for (unsigned int i = 0; i < msgCount; i++)
-        {
-            auto result = client.concatenateTwoStrings(str1, str2);
+//    msgSize = 1000;
+//    std::cout << std::endl << "** Measuring method calls of size " << msgSize << " bytes (" << repetitions << " repetitions)..." << std::endl << std::endl;
+//    for (unsigned int r = 0; r < repetitions; ++r)
+//    {
+//        auto str1 = createRandomString(msgSize/2);
+//        auto str2 = createRandomString(msgSize/2);
 
-            assert(result.size() == str1.size() + str2.size());
-            assert(result.size() == msgSize);
-        }
-        auto stopTime = std::chrono::steady_clock::now();
-        std::cout << "Called " << msgCount << " methods in: " << std::chrono::duration_cast<std::chrono::milliseconds>(stopTime - startTime).count() << " ms" << std::endl;
+//        auto startTime = std::chrono::steady_clock::now();
+//        for (unsigned int i = 0; i < msgCount; i++)
+//        {
+//            auto result = client.concatenateTwoStrings(str1, str2);
 
-        std::this_thread::sleep_for(1000ms);
-    }
+//            assert(result.size() == str1.size() + str2.size());
+//            assert(result.size() == msgSize);
+//        }
+//        auto stopTime = std::chrono::steady_clock::now();
+//        auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(stopTime - startTime).count();
+//        totalDuration += duration;
+//        std::cout << "Called " << msgCount << " methods in: " << duration << " ms" << std::endl;
+
+//        std::this_thread::sleep_for(1000ms);
+//    }
+
+    std::cout << "AVERAGE: " << (totalDuration/repetitions) << " ms" << std::endl;
+    totalDuration = 0;
 
     return 0;
 }