Emulate sync D-Bus method call with async implementation underneath to avoid poll race condition

src/Connection.cpp

@@ -34,6 +34,7 @@
 #include <unistd.h>
 #include <poll.h>
 #include <sys/eventfd.h>
+#include <set>
 
 #include <unistd.h>
 #include <sys/syscall.h>
@@ -51,7 +52,6 @@ Connection::Connection(std::unique_ptr<ISdBus>&& interface, const BusFactory& bu
 Connection::Connection(std::unique_ptr<ISdBus>&& interface, system_bus_t)
     : Connection(std::move(interface), [this](sd_bus** bus){ return iface_->sd_bus_open_system(bus); })
 {
-printf("Here: %p\n", this);
 }
 
 Connection::Connection(std::unique_ptr<ISdBus>&& interface, session_bus_t)
@@ -91,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();
@@ -101,46 +105,14 @@ void Connection::enterProcessingLoop()
         if (!success)
             break; // Exit processing loop
     }
+
+    loopThreadId_ = std::thread::id{};
 }
 
 void Connection::enterProcessingLoopAsync()
 {
     if (!asyncLoopThread_.joinable())
         asyncLoopThread_ = std::thread([this](){ enterProcessingLoop(); });
-//    if (!asyncLoopThread2_.joinable())
-//        asyncLoopThread2_ = std::thread([this](){
-//            enterProcessingLoop();
-////            while (true)
-////            {
-////                auto bus = bus_.get();
-
-////                assert(bus != nullptr);
-////                assert(loopExitFd_.fd != 0);
-
-////                auto sdbusPollData = getProcessLoopPollData();
-////                struct pollfd fds[] = {{sdbusPollData.fd, sdbusPollData.events, 0}, {loopExitFd_.fd, POLLIN, 0}};
-////                auto fdsCount = sizeof(fds)/sizeof(fds[0]);
-
-////                printf("Thread 2: Going to poll %p\n", this);
-
-////                auto timeout = sdbusPollData.timeout_usec == (uint64_t) -1 ? (uint64_t)-1 : (sdbusPollData.timeout_usec+999)/1000;
-////                auto r = poll(fds, fdsCount, timeout);
-
-////                printf("Thread 2: Poll woken up %p\n", this);
-
-////                if (r < 0 && errno == EINTR)
-////                    continue;
-
-////                SDBUS_THROW_ERROR_IF(r < 0, "Failed to wait on the bus", -errno);
-
-////                if (fds[1].revents & POLLIN)
-////                {
-////                    clearExitNotification();
-////                    printf("Thread 2: Exiting %p\n", this);
-////                    break;
-////                }
-////            }
-//        });
 }
 
 void Connection::leaveProcessingLoop()
@@ -327,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{};
@@ -397,13 +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);
+    //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);
+    //printf("Thread %d: Poll woken up %p\n", gettid(), this);
 
     if (r < 0 && errno == EINTR)
         return true; // Try again
@@ -413,7 +414,7 @@ bool Connection::waitForNextRequest()
     if (fds[1].revents & POLLIN)
     {
         clearExitNotification();
-        printf("Thread %d: Exiting %p\n", gettid(), this);
+        //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,7 +134,8 @@ namespace sdbus { namespace internal {
         std::unique_ptr<ISdBus> iface_;
         BusPtr bus_;
         std::thread asyncLoopThread_;
-        std::thread asyncLoopThread2_;
+        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,35 @@ 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);
+
+    /*
+    // 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 +121,35 @@ 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)
+{
+    // TODO: use thread_local data exchange facility (OPTIMIZE)
+    std::promise<MethodReply> result;
+    auto future = result.get_future();
+
+    auto callback = (void*)&Proxy::sdbus_quasi_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;
+
+    //    // TODO: Switch to thread_local once we have re-usable thread_local data exchange facility
+    //    /*thread_local*/ async_reply_handler asyncReplyCallback = [&result](MethodReply& reply, const Error* error)
+    //    {
+    //        if (error == nullptr)
+    //            result.set_value(std::move(reply));
+    //        else
+    //            result.set_exception(std::make_exception_ptr(error));
+    //    };
+
+    //    auto callback = (void*)&Proxy::sdbus_async_reply_handler;
+    //    AsyncCalls::CallData callData{*this, std::move(asyncReplyCallback), {}};
+    //    message.sendWithAsyncReply((void*)&Proxy::sdbus_async_reply_handler, &data, timeout);
+}
+
 void Proxy::registerSignalHandler( const std::string& interfaceName
                                  , const std::string& signalName
                                  , signal_handler signalHandler )
@@ -122,7 +184,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 +196,7 @@ void Proxy::unregister()
     interfaces_.clear();
 }
 
+// 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);
@@ -159,7 +222,35 @@ 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_quasi_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());
+
+    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

@@ -62,9 +62,11 @@ namespace internal {
         void unregister() override;
 
     private:
+        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_quasi_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