Poll tests

src/Connection.cpp

@@ -35,6 +35,10 @@
 #include <poll.h>
 #include <sys/eventfd.h>
 
+#include <unistd.h>
+#include <sys/syscall.h>
+#define gettid() syscall(SYS_gettid)
+
 namespace sdbus { namespace internal {
 
 Connection::Connection(std::unique_ptr<ISdBus>&& interface, const BusFactory& busFactory)
@@ -47,6 +51,7 @@ 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)
@@ -102,6 +107,40 @@ 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()
@@ -332,6 +371,8 @@ void Connection::joinWithProcessingLoop()
 {
     if (asyncLoopThread_.joinable())
         asyncLoopThread_.join();
+//    if (asyncLoopThread2_.joinable())
+//        asyncLoopThread2_.join();
 }
 
 bool Connection::processPendingRequest()
@@ -356,8 +397,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 +413,7 @@ bool Connection::waitForNextRequest()
     if (fds[1].revents & POLLIN)
     {
         clearExitNotification();
+        printf("Thread %d: Exiting %p\n", gettid(), this);
         return false;
     }
 

src/Connection.h

@@ -130,6 +130,7 @@ namespace sdbus { namespace internal {
         std::unique_ptr<ISdBus> iface_;
         BusPtr bus_;
         std::thread asyncLoopThread_;
+        std::thread asyncLoopThread2_;
         LoopExitEventFd loopExitFd_;
     };
 

tests/integrationtests/TestingAdaptor.h

@@ -50,6 +50,20 @@ public:
 
 protected:
 
+    int32_t doSignalEmission() override
+    {
+        static int32_t counter = 0;
+        emitSimpleSignal();
+        printf("First simple signal thrown\n");
+        counter++;
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+        //emitSimpleSignal();
+        //printf("Second simple signal thrown\n");
+        counter++;
+
+        return counter;
+    }
+
     void noArgNoReturn() const
     {
     }

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
@@ -57,6 +61,7 @@ protected:
     void onSimpleSignal() override
     {
         m_gotSimpleSignal = true;
+        printf("Thread %d: Got simple signal\n", gettid());
     }
 
     void onSignalWithMap(const std::map<int32_t, std::string>& m) override

tests/integrationtests/adaptor-glue.h

@@ -59,6 +59,8 @@ protected:
     {
         object_.setInterfaceFlags(INTERFACE_NAME).markAsDeprecated().withPropertyUpdateBehavior(sdbus::Flags::EMITS_NO_SIGNAL);
 
+        object_.registerMethod("doSignalEmission").onInterface(INTERFACE_NAME).implementedAs([this](){ return this->doSignalEmission(); });
+
         object_.registerMethod("noArgNoReturn").onInterface(INTERFACE_NAME).implementedAs([this](){ return this->noArgNoReturn(); });
         object_.registerMethod("getInt").onInterface(INTERFACE_NAME).implementedAs([this](){ return this->getInt(); });
         object_.registerMethod("getTuple").onInterface(INTERFACE_NAME).implementedAs([this](){ return this->getTuple(); });
@@ -150,6 +152,7 @@ private:
 
 protected:
 
+    virtual int32_t doSignalEmission() = 0;
     virtual void noArgNoReturn() const  = 0;
     virtual int32_t getInt() const = 0;
     virtual std::tuple<uint32_t, std::string> getTuple() const = 0;

tests/integrationtests/proxy-glue.h

@@ -56,6 +56,14 @@ protected:
     virtual void onDoOperationReply(uint32_t returnValue, const sdbus::Error* error) = 0;
 
 public:
+    int32_t doSignalEmission()
+    {
+        int32_t result;
+        object_.callMethod("doSignalEmission").onInterface(INTERFACE_NAME).storeResultsTo(result);
+        //object_.callMethodAsync("doSignalEmission").onInterface(INTERFACE_NAME).uponReplyInvoke([this](const sdbus::Error* error, int32_t returnValue){ printf("Async reply handler\n"); });
+        return result;
+    }
+
     void noArgNoReturn()
     {
         object_.callMethod("noArgNoReturn").onInterface(INTERFACE_NAME);