Fix issue of event loop thread and synchronous method call thread polling on the same D-Bus connection

Synchronous D-Bus method calls are now done in terms of blocking asynchronous calls.

src/Connection.cpp

@@ -86,6 +86,10 @@ std::string Connection::getUniqueName() const
 
 void Connection::enterProcessingLoop()
 {
+    loopThreadId_ = std::this_thread::get_id();
+
+    std::lock_guard guard(loopMutex_);
+
     while (true)
     {
         auto processed = processPendingRequest();
@@ -96,6 +100,8 @@ void Connection::enterProcessingLoop()
         if (!success)
             break; // Exit processing loop
     }
+
+    loopThreadId_ = std::thread::id{};
 }
 
 void Connection::enterProcessingLoopAsync()
@@ -288,6 +294,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 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 {};
+}
+
 Connection::BusPtr Connection::openBus(const BusFactory& busFactory)
 {
     sd_bus* bus{};

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

@@ -36,7 +36,6 @@
 // Forward declaration
 namespace sdbus {
     class MethodCall;
-    class AsyncMethodCall;
     class MethodReply;
     class Signal;
     namespace internal {
@@ -91,6 +90,8 @@ namespace internal {
 
         virtual void enterProcessingLoopAsync() = 0;
         virtual void leaveProcessingLoop() = 0;
+
+        virtual MethodReply tryCallMethodSynchronously(const MethodCall& message, uint64_t timeout) = 0;
     };
 
 }

src/Message.cpp

@@ -660,6 +660,22 @@ MethodReply MethodCall::sendWithNoReply() const
     return Factory::create<MethodReply>(); // No reply
 }
 
+void MethodCall::send(void* callback, void* userData, uint64_t timeout, dont_request_slot_t) 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", -r);
+}
+
+MethodCall::Slot MethodCall::send(void* callback, void* userData, uint64_t timeout) const
+{
+    sd_bus_slot* slot;
+
+    auto r = sdbus_->sd_bus_call_async(nullptr, &slot, (sd_bus_message*)msg_, (sd_bus_message_handler_t)callback, userData, timeout);
+    SDBUS_THROW_ERROR_IF(r < 0, "Failed to call method asynchronously", -r);
+
+    return Slot{slot, [sdbus_ = sdbus_](void *slot){ sdbus_->sd_bus_slot_unref((sd_bus_slot*)slot); }};
+}
+
 MethodReply MethodCall::createReply() const
 {
     sd_bus_message* sdbusReply{};
@@ -682,21 +698,6 @@ MethodReply MethodCall::createErrorReply(const Error& error) const
     return Factory::create<MethodReply>(sdbusErrorReply, sdbus_, adopt_message);
 }
 
-AsyncMethodCall::AsyncMethodCall(MethodCall&& call) noexcept
-    : Message(std::move(call))
-{
-}
-
-AsyncMethodCall::Slot AsyncMethodCall::send(void* callback, void* userData, uint64_t timeout) const
-{
-    sd_bus_slot* slot;
-
-    auto r = sdbus_->sd_bus_call_async(nullptr, &slot, (sd_bus_message*)msg_, (sd_bus_message_handler_t)callback, userData, timeout);
-    SDBUS_THROW_ERROR_IF(r < 0, "Failed to call method asynchronously", -r);
-
-    return Slot{slot, [sdbus_ = sdbus_](void *slot){ sdbus_->sd_bus_slot_unref((sd_bus_slot*)slot); }};
-}
-
 void MethodReply::send() const
 {
     auto r = sdbus_->sd_bus_send(nullptr, (sd_bus_message*)msg_, nullptr);

src/Proxy.cpp

@@ -64,19 +64,36 @@ MethodCall Proxy::createMethodCall(const std::string& interfaceName, const std::
     return connection_->createMethodCall(destination_, objectPath_, interfaceName, methodName);
 }
 
-AsyncMethodCall Proxy::createAsyncMethodCall(const std::string& interfaceName, const std::string& methodName)
-{
-    return AsyncMethodCall{Proxy::createMethodCall(interfaceName, methodName)};
-}
-
 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 via sd_bus_call, which blocks. 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 other sd-bus API accesses), but the incoming reply we have to get through the event
+    // loop thread, because this is 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);
+    // 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.send(timeout);
+
+    // 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 sendMethodCallMessageAndWaitForReply(message, timeout);
 }
 
-void Proxy::callMethod(const AsyncMethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout)
+void Proxy::callMethod(const MethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout)
 {
     SDBUS_THROW_ERROR_IF(!message.isValid(), "Invalid async method call message provided", EINVAL);
 
@@ -88,6 +105,49 @@ void Proxy::callMethod(const AsyncMethodCall& message, async_reply_handler async
     pendingAsyncCalls_.addCall(callData->slot.get(), std::move(callData));
 }
 
+MethodReply Proxy::sendMethodCallMessageAndWaitForReply(const MethodCall& message, uint64_t timeout)
+{
+    /*thread_local*/ SyncCallReplyData syncCallReplyData;
+
+    async_reply_handler asyncReplyCallback = [&syncCallReplyData](MethodReply& reply, const Error* error)
+    {
+        syncCallReplyData.sendMethodReplyToWaitingThread(reply, error);
+    };
+    auto callback = (void*)&Proxy::sdbus_async_reply_handler;
+    AsyncCalls::CallData callData{*this, std::move(asyncReplyCallback), {}};
+
+    message.send(callback, &callData, timeout, dont_request_slot);
+
+    return syncCallReplyData.waitForMethodReply();
+}
+
+void Proxy::SyncCallReplyData::sendMethodReplyToWaitingThread(MethodReply& reply, const Error* error)
+{
+    SCOPE_EXIT{ cond_.notify_one(); };
+    std::unique_lock lock{mutex_};
+    SCOPE_EXIT{ arrived_ = true; };
+
+    //error_ = nullptr; // Necessary if SyncCallReplyData instance is thread_local
+
+    if (error == nullptr)
+        reply_ = std::move(reply);
+    else
+        error_ = std::make_unique<Error>(*error);
+}
+
+MethodReply Proxy::SyncCallReplyData::waitForMethodReply()
+{
+    std::unique_lock lock{mutex_};
+    cond_.wait(lock, [this](){ return arrived_; });
+
+    //arrived_ = false; // Necessary if SyncCallReplyData instance is thread_local
+
+    if (error_)
+        throw *error_;
+
+    return std::move(reply_);
+}
+
 void Proxy::registerSignalHandler( const std::string& interfaceName
                                  , const std::string& signalName
                                  , signal_handler signalHandler )
@@ -122,7 +182,7 @@ void Proxy::registerSignalHandlers(sdbus::internal::IConnection& connection)
             slot = connection.registerSignalHandler( objectPath_
                                                    , interfaceName
                                                    , signalName
-                                                   , &Proxy::sdbus_signal_callback
+                                                   , &Proxy::sdbus_signal_handler
                                                    , this );
         }
     }
@@ -141,7 +201,13 @@ 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
+    {
+        // Slot may be null if we're doing blocking synchronous call implemented by means of asynchronous call,
+        // because in that case the call data is still alive on the stack, we don't need to manage it separately.
+        if (asyncCallData->slot)
+            proxy.pendingAsyncCalls_.removeCall(asyncCallData->slot.get());
+    };
 
     auto message = Message::Factory::create<MethodReply>(sdbusMessage, &proxy.connection_->getSdBusInterface());
 
@@ -159,7 +225,7 @@ 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*/)
+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 {
@@ -51,9 +52,8 @@ namespace internal {
              , std::string objectPath );
 
         MethodCall createMethodCall(const std::string& interfaceName, const std::string& methodName) override;
-        AsyncMethodCall createAsyncMethodCall(const std::string& interfaceName, const std::string& methodName) override;
         MethodReply callMethod(const MethodCall& message, uint64_t timeout) override;
-        void callMethod(const AsyncMethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout) override;
+        void callMethod(const MethodCall& message, async_reply_handler asyncReplyCallback, uint64_t timeout) override;
 
         void registerSignalHandler( const std::string& interfaceName
                                   , const std::string& signalName
@@ -62,9 +62,24 @@ namespace internal {
         void unregister() override;
 
     private:
+        class SyncCallReplyData
+        {
+        public:
+            void sendMethodReplyToWaitingThread(MethodReply& reply, const Error* error);
+            MethodReply waitForMethodReply();
+
+        private:
+            std::mutex mutex_;
+            std::condition_variable cond_;
+            bool arrived_{};
+            MethodReply reply_;
+            std::unique_ptr<Error> error_;
+        };
+
+        MethodReply sendMethodCallMessageAndWaitForReply(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_signal_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError);
 
     private:
         std::unique_ptr< sdbus::internal::IConnection
@@ -97,7 +112,7 @@ namespace internal {
             {
                 Proxy& proxy;
                 async_reply_handler callback;
-                AsyncMethodCall::Slot slot;
+                MethodCall::Slot slot;
             };
 
             ~AsyncCalls()