Merge lp:~thomas-voss/dbus-cpp/fix-invalid-reads-in-executor into lp:dbus-cpp

[Thomas Voß]

Always unref pending calls, add load test to ensure correct behavior.
Make sure that timeouts are never notified after destruction.

Always unref pending calls, add load test to ensure correct behavior.
Make sure that timeouts are never notified after destruction.

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:78
http://jenkins.qa.ubuntu.com/job/dbus-cpp-ci/199/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-amd64-ci/39/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-armhf-ci/39/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-i386-ci/39/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/dbus-cpp-ci/199/rebuild

review: Needs Fixing (continuous-integration)

[lp:~thomas-voss/dbus-cpp/fix-invalid-reads-in-executor]

79. By Thomas Voß on 2014-09-29

Further harden PendingCallImpl.

80. By Thomas Voß on 2014-09-29

Make sure that the ExecutionCounter does not go out of scope.

81. By Thomas Voß on 2014-09-29

Pass by value, not by reference.

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:79
http://jenkins.qa.ubuntu.com/job/dbus-cpp-ci/200/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-amd64-ci/40/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-armhf-ci/40/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-i386-ci/40/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/dbus-cpp-ci/200/rebuild

review: Needs Fixing (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:81
http://jenkins.qa.ubuntu.com/job/dbus-cpp-ci/201/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-amd64-ci/41/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-armhf-ci/41/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-i386-ci/41/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/dbus-cpp-ci/201/rebuild

review: Needs Fixing (continuous-integration)

[Alberto Aguirre (albaguirre)]

Looks good.

135 + std::lock_guard<std::mutex> lg{wrapper->pending_call->guard};

One small nit:

148 + void notify_locked(const Message::Ptr& msg)

Maybe just take the a std::lock_guard const& to enforce the locked constraint?

466 + auto ec = std::make_shared<CountingEventCollector>(5000);
467 +
468 + std::thread t1{[dbus, ec]() {invoke_list_names_n_times_and_update_event_collector(dbus, 1000, ec);}};
469 + std::thread t2{[dbus, ec]() {invoke_list_names_n_times_and_update_event_collector(dbus, 1000, ec);}};
470 + std::thread t3{[dbus, ec]() {invoke_list_names_n_times_and_update_event_collector(dbus, 1000, ec);}};
471 + std::thread t4{[dbus, ec]() {invoke_list_names_n_times_and_update_event_collector(dbus, 1000, ec);}};
472 + std::thread t5{[dbus, ec]() {invoke_list_names_n_times_and_update_event_collector(dbus, 1000, ec);}};

I guess CI is too slow...maybe 100 per thread should be enough?

[Alberto Aguirre (albaguirre)]

review: Approve

[lp:~thomas-voss/dbus-cpp/fix-invalid-reads-in-executor]

82. By Thomas Voß on 2014-10-01

Adjust test timeout.

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:82
http://jenkins.qa.ubuntu.com/job/dbus-cpp-ci/203/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-amd64-ci/43/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-armhf-ci/43/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-i386-ci/43/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/dbus-cpp-ci/203/rebuild

review: Needs Fixing (continuous-integration)

[lp:~thomas-voss/dbus-cpp/fix-invalid-reads-in-executor]

83. By Thomas Voß on 2014-10-02

Lower the default number of remote queries to make Jenkins happy.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:83
http://jenkins.qa.ubuntu.com/job/dbus-cpp-ci/204/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-amd64-ci/44
    SUCCESS: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-armhf-ci/44
        deb: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-armhf-ci/44/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/dbus-cpp-utopic-i386-ci/44

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/dbus-cpp-ci/204/rebuild

review: Approve (continuous-integration)

[Alberto Aguirre (albaguirre)]

428 + std::uint64_t expected;
429 + std::atomic<std::uint64_t> counter;

Can you make these int (mainly the std::atomic<std::uint64_t> to std::atomic<int> ) so that it compiles in ppc?

[lp:~thomas-voss/dbus-cpp/fix-invalid-reads-in-executor]

84. By Thomas Voß on 2014-10-02

std::*64 -> int for powerpc compatibility.

=== modified file 'src/core/dbus/asio/executor.cpp'
--- src/core/dbus/asio/executor.cpp	2014-06-23 10:19:48 +0000
+++ src/core/dbus/asio/executor.cpp	2014-10-02 20:47:37 +0000
@@ -122,14 +122,20 @@
                 return;
             }
 
+            // We do not keep ourselves alive to prevent from races during destruction.
+            std::weak_ptr<Timeout<UnderlyingTimeoutType>> wp{this->shared_from_this()};
+
             timer.expires_from_now(
                         boost::posix_time::milliseconds(
                             traits::Timeout<UnderlyingTimeoutType>::get_timeout_interval(
                                 timeout)));
-            timer.async_wait(
-                        std::bind(&Timeout::on_timeout,
-                                  Timeout<UnderlyingTimeoutType>::shared_from_this(),
-                                  std::placeholders::_1));
+            timer.async_wait([wp](const boost::system::error_code& ec)
+            {
+                auto sp = wp.lock();
+
+                if (sp)
+                    sp->on_timeout(ec);
+            });
         }
 
         void cancel()
@@ -184,27 +190,35 @@
 
         void restart()
         {
+            // We do not keep ourselves alive to prevent from races during destruction.
+            std::weak_ptr<Watch<UnderlyingWatchType>> wp{this->shared_from_this()};
+
             if (traits::Watch<UnderlyingWatchType>::is_watch_monitoring_fd_for_readable(watch))
             {
-                stream_descriptor.async_read_some(
-                            boost::asio::null_buffers(),
-                            std::bind(
-                                &Watch::on_stream_descriptor_event,
-                                Watch<UnderlyingWatchType>::shared_from_this(),
-                                traits::Watch<UnderlyingWatchType>::readable_event(),
-                                std::placeholders::_1,
-                                std::placeholders::_2));
+                stream_descriptor.async_read_some(boost::asio::null_buffers(), [wp](boost::system::error_code ec, std::size_t bytes_transferred)
+                {
+                    auto sp = wp.lock();
+
+                    if (sp)
+                        sp->on_stream_descriptor_event(
+                                    traits::Watch<UnderlyingWatchType>::readable_event(),
+                                    ec,
+                                    bytes_transferred);
+                });
             }
+
             if (traits::Watch<UnderlyingWatchType>::is_watch_monitoring_fd_for_writable(watch))
             {
-                stream_descriptor.async_write_some(
-                            boost::asio::null_buffers(),
-                            std::bind(
-                                &Watch::on_stream_descriptor_event,
-                                Watch<UnderlyingWatchType>::shared_from_this(),
-                                traits::Watch<UnderlyingWatchType>::writeable_event(),
-                                std::placeholders::_1,
-                                std::placeholders::_2));
+                stream_descriptor.async_write_some(boost::asio::null_buffers(), [wp](boost::system::error_code ec, std::size_t bytes_transferred)
+                {
+                    auto sp = wp.lock();
+
+                    if (sp)
+                        sp->on_stream_descriptor_event(
+                                    traits::Watch<UnderlyingWatchType>::writeable_event(),
+                                    ec,
+                                    bytes_transferred);
+                });
             }
         }
 

=== modified file 'src/core/dbus/pending_call_impl.h'
--- src/core/dbus/pending_call_impl.h	2014-03-05 09:08:15 +0000
+++ src/core/dbus/pending_call_impl.h	2014-10-02 20:47:37 +0000
@@ -26,6 +26,17 @@
 
 #include <mutex>
 
+namespace
+{
+// Better save than sorry. We wrap common error handling here.
+bool is_pending_call_completed(DBusPendingCall* pending_call)
+{
+    if (not pending_call)
+        return false;
+
+    return dbus_pending_call_get_completed(pending_call) == TRUE;
+}
+}
 namespace core
 {
 namespace dbus
@@ -45,18 +56,44 @@
     {
         auto wrapper = static_cast<Wrapper*>(cookie);
 
-        auto message = dbus_pending_call_steal_reply(call);
+        if (not wrapper)
+            return;
 
-        if (message)
+        // We tie cleanup of the wrapper to the scope of the callback.
+        // With that, even an exception being thrown would _not_ result
+        // in an instance being leaked.
+        struct Scope
         {
-            wrapper->pending_call->notify(Message::from_raw_message(message));
-        }
+            ~Scope()
+            {
+                // We take over ownership of the wrapper and destroy on scope exit.
+                delete wrapper;
+                // We always unref this message, even if notify_locked or
+                // from_raw_message throws.
+                if (message) dbus_message_unref(message);
+            }
+
+            Wrapper* wrapper;
+            // We only need this later on, when we have stolen the reply
+            // from the pending call.
+            DBusMessage* message;
+        } scope{wrapper, nullptr};
+
+        // We synchronize to avoid races on construction.
+        std::lock_guard<std::mutex> lg{wrapper->pending_call->guard};
+        // And we only steal the reply if the call actually completed.
+        if (not is_pending_call_completed(call))
+            return;
+
+        scope.message = dbus_pending_call_steal_reply(call);
+
+        if (scope.message)
+            wrapper->pending_call->notify_locked(Message::from_raw_message(scope.message));
     }
 
-    void notify(const Message::Ptr& msg)
+    // Announce incoming reply and invoke the callback if set.
+    void notify_locked(const Message::Ptr& msg)
     {
-        std::lock_guard<std::mutex> lg(guard);
-
         message = msg;
 
         if (callback)
@@ -71,36 +108,93 @@
 public:
     inline static core::dbus::PendingCall::Ptr create(DBusPendingCall* call)
     {
+        if (not call) throw std::runtime_error
+        {
+            "core::dbus::PendingCall cannot be constructed for null object."
+        };
+
         auto result = std::shared_ptr<core::dbus::impl::PendingCall>
         {
             new core::dbus::impl::PendingCall{call}
         };
 
+        // Our scope contains two objects that are dynamically created:
+        //   * The actual PendingCall implementation (managed by a shared pointer)
+        //   * A helper object of type Wrapper for passing around the pending call instance.
+        // The latter one needs some manual memory mgmt. until we are sure that it got handed
+        // over to libdbus correctly.
+        struct Scope
+        {
+            // Whenever we go out of scope, we unref the call (we do not need it anymore)
+            // and we potentially (if armed) delete the wrapper.
+            ~Scope()
+            {
+                dbus_pending_call_unref(call);
+                if (armed) delete wrapper;
+            }
+
+            // Disarms the scope for handling the wrapper instance.
+            // Think about it like giving up ownership as we handed over
+            // to libdbus.
+            void disarm_for_wrapper()
+            {
+                armed = false;
+            }
+
+            // The raw call instance.
+            DBusPendingCall* call;
+            // True if the wrapper instance is still owned by the scope.
+            bool armed;
+            // The wrapper instance.
+            Wrapper* wrapper;
+        } scope
+        {
+            // The raw call that we got handed from the caller.
+            call,
+            // Yes, we still own the Wrapper instance.
+            true, new Wrapper{result}
+        };
+
+        // We synchronize to avoid races on construction.
+        std::lock_guard<std::mutex> lg{result->guard};
+
         if (FALSE == dbus_pending_call_set_notify(
-                result->pending_call,
-                PendingCall::on_pending_call_completed,
-                new Wrapper{result},
-                [](void* data)
-                {
-                    delete static_cast<Wrapper*>(data);
-                }))
+                // We dispatch to the static on_pending_call_completed when
+                // the call completes.
+                result->pending_call, PendingCall::on_pending_call_completed,
+                // We pass in our helper object and do not specify a deleter.
+                // Please refer to the source-code comments in on_pending_call_completed.
+                scope.wrapper, nullptr))
         {
             throw std::runtime_error("Error setting up pending call notification.");
         }
 
         // And here comes the beauty of libdbus, and its racy architecture:
-        {
-            std::lock_guard<std::mutex> lg(result->guard);
-            if (TRUE == dbus_pending_call_get_completed(call))
+        {            
+            if (is_pending_call_completed(call))
             {
                 // We took too long while setting up the pending call notification.
                 // For that we now have to inject the message here.
                 auto msg = dbus_pending_call_steal_reply(call);
-                result->message = Message::from_raw_message(msg);
+
+                if (msg)
+                {
+                    result->message = Message::from_raw_message(msg);
+                    // We decrease the reference count as Message::from_raw_message
+                    // always refs the object that it is passed.
+                    dbus_message_unref(msg);
+                }
+            }
+            else
+            {
+                // We need the wrapper to be alive, so we disarm the scope.
+                // Please note that this is the only "good" path through this
+                // mess of setup and notification functions.
+                scope.disarm_for_wrapper();
             }
         }
 
-        return std::dynamic_pointer_cast<core::dbus::PendingCall>(result);
+        return result;
     }
 
     void cancel()
@@ -122,7 +216,7 @@
     PendingCall(DBusPendingCall* call)
         : pending_call(call)
     {
-    }
+    }    
 };
 }
 }

=== modified file 'tests/CMakeLists.txt'
--- tests/CMakeLists.txt	2014-01-27 12:22:43 +0000
+++ tests/CMakeLists.txt	2014-10-02 20:47:37 +0000
@@ -14,13 +14,14 @@
 #
 # Authored by: Thomas Voss <[e-mail]>
 
-set (OLD_CMAKE_CXX_FLAGS ${CMAKE_CXX_FLAGS})
-# Don't treat warnings as errors in 3rd_party/{gmock,cucumber-cpp}
-string (REPLACE " -Werror " " " CMAKE_CXX_FLAGS ${CMAKE_CXX_FLAGS})
-find_package(Gtest REQUIRED)
-include_directories(${GMOCK_INCLUDE_DIR} ${GTEST_INCLUDE_DIR})
-set (CMAKE_CXX_FLAGS ${OLD_CMAKE_CXX_FLAGS})
-
+# Build with system gmock and embedded gtest
+set (GMOCK_INCLUDE_DIR "/usr/include/gmock/include" CACHE PATH "gmock source include directory")
+set (GMOCK_SOURCE_DIR "/usr/src/gmock" CACHE PATH "gmock source directory")
+set (GTEST_INCLUDE_DIR "${GMOCK_SOURCE_DIR}/gtest/include" CACHE PATH "gtest source include directory")
+
+add_subdirectory(${GMOCK_SOURCE_DIR} "${CMAKE_CURRENT_BINARY_DIR}/gmock")
+
+set(GTEST_BOTH_LIBRARIES gmock gtest gtest_main)
 find_package(Threads)
 
 add_definitions(-DCORE_DBUS_ENABLE_GOOGLE_TEST_FIXTURE)
@@ -41,6 +42,11 @@
 )
 
 add_executable(
+  async_execution_load_test
+  async_execution_load_test.cpp
+  )
+
+add_executable(
   bus_test
   bus_test.cpp
   )
@@ -111,6 +117,19 @@
   )
 
 target_link_libraries(
+  async_execution_load_test
+
+  dbus-cpp
+  dbus-cppc-helper
+
+  ${CMAKE_THREAD_LIBS_INIT}
+  ${Boost_LIBRARIES}
+  ${DBUS_LIBRARIES}
+  ${GTEST_BOTH_LIBRARIES}
+  ${PROCESS_CPP_LIBRARIES}
+)
+
+target_link_libraries(
   bus_test
 
   dbus-cpp
@@ -279,6 +298,7 @@
   ${GTEST_BOTH_LIBRARIES}
   )
 
+add_test(async_execution_load_test ${CMAKE_CURRENT_BINARY_DIR}/async_execution_load_test)
 add_test(bus_test ${CMAKE_CURRENT_BINARY_DIR}/bus_test)
 add_test(cache_test ${CMAKE_CURRENT_BINARY_DIR}/cache_test)
 add_test(dbus_test ${CMAKE_CURRENT_BINARY_DIR}/dbus_test)

=== added file 'tests/async_execution_load_test.cpp'
--- tests/async_execution_load_test.cpp	1970-01-01 00:00:00 +0000
+++ tests/async_execution_load_test.cpp	2014-10-02 20:47:37 +0000
@@ -0,0 +1,146 @@
+/*
+ * Copyright © 2014 Canonical Ltd.
+ *
+ * This program is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License version 3,
+ * as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * Authored by: Thomas Voß <[e-mail]>
+ */
+
+#include <core/dbus/fixture.h>
+#include <core/dbus/macros.h>
+#include <core/dbus/object.h>
+#include <core/dbus/service.h>
+
+#include <core/dbus/asio/executor.h>
+
+#include <core/dbus/types/stl/string.h>
+#include <core/dbus/types/stl/vector.h>
+
+#include <core/testing/fork_and_run.h>
+
+#include <gmock/gmock.h>
+
+#include "test_data.h"
+
+namespace
+{
+// Our fixture for starting up private system- and session-bus instances.
+struct AsyncExecutionLoadTest : public core::dbus::testing::Fixture
+{
+};
+
+auto session_bus_config_file =
+        core::dbus::testing::Fixture::default_session_bus_config_file() =
+        core::testing::session_bus_configuration_file();
+
+auto system_bus_config_file =
+        core::dbus::testing::Fixture::default_system_bus_config_file() =
+        core::testing::system_bus_configuration_file();
+
+struct DBus
+{
+    static const std::string& name()
+    {
+        static const std::string s{DBUS_SERVICE_DBUS};
+        return s;
+    }
+
+    DBUS_CPP_METHOD_WITH_TIMEOUT_DEF(ListNames, DBus, 25000)
+};
+
+struct CountingEventCollector
+{
+    CountingEventCollector(int expected)
+        : expected{expected},
+          counter{0}
+    {
+    }
+
+    void update()
+    {
+        if (++counter == expected)
+            wait_condition.notify_all();
+    }
+
+    ::testing::AssertionResult wait_for(const std::chrono::milliseconds& ms)
+    {
+        std::unique_lock<std::mutex> ul(guard);
+
+        auto result = wait_condition.wait_for(ul, ms, [this]() { return counter == expected; });
+
+        if (result)
+            return ::testing::AssertionSuccess();
+
+        return ::testing::AssertionFailure() << "Current count of "
+                                             << counter << " does not match " << expected;
+    }
+
+    int expected;
+    std::atomic<int> counter;
+
+    std::mutex guard;
+    std::condition_variable wait_condition;
+};
+
+void invoke_list_names_n_times_and_update_event_collector(
+        // The object referring to the bus daemon
+        const core::dbus::Object::Ptr& dbus,
+        // Number of iterations
+        std::size_t n,
+        // The event collector instance that should be updated
+        const std::shared_ptr<CountingEventCollector>& ec)
+{
+    for (unsigned int i = 0; i < n; i++)
+    {
+        dbus->invoke_method_asynchronously_with_callback<DBus::ListNames, std::vector<std::string>>([ec](const core::dbus::Result<std::vector<std::string>>& vs)
+        {
+            if (not vs.is_error())
+                ec->update();
+        });
+    }
+}
+}
+
+TEST_F(AsyncExecutionLoadTest, RepeatedlyInvokingAnAsyncFunctionWorks)
+{
+    using namespace ::testing;
+
+    auto bus = session_bus();
+    bus->install_executor(core::dbus::asio::make_executor(bus));
+
+    std::thread worker{[bus]() { bus->run(); }};
+
+    auto service = core::dbus::Service::use_service(bus, DBus::name());
+    auto dbus = service->object_for_path(core::dbus::types::ObjectPath{DBUS_PATH_DBUS});
+
+    auto ec = std::make_shared<CountingEventCollector>(500);
+
+    std::thread t1{[dbus, ec]() {invoke_list_names_n_times_and_update_event_collector(dbus, 100, ec);}};
+    std::thread t2{[dbus, ec]() {invoke_list_names_n_times_and_update_event_collector(dbus, 100, ec);}};
+    std::thread t3{[dbus, ec]() {invoke_list_names_n_times_and_update_event_collector(dbus, 100, ec);}};
+    std::thread t4{[dbus, ec]() {invoke_list_names_n_times_and_update_event_collector(dbus, 100, ec);}};
+    std::thread t5{[dbus, ec]() {invoke_list_names_n_times_and_update_event_collector(dbus, 100, ec);}};
+
+    EXPECT_TRUE(ec->wait_for(std::chrono::seconds{60}));
+
+    bus->stop();
+
+    if (t1.joinable()) t1.join();
+    if (t2.joinable()) t2.join();
+    if (t3.joinable()) t3.join();
+    if (t4.joinable()) t4.join();
+    if (t5.joinable()) t5.join();
+
+    if (worker.joinable())
+        worker.join();
+}

=== modified file 'tests/executor_test.cpp'
--- tests/executor_test.cpp	2014-06-23 10:19:48 +0000
+++ tests/executor_test.cpp	2014-10-02 20:47:37 +0000
@@ -263,8 +263,6 @@
     EXPECT_EQ(core::testing::ForkAndRunResult::empty, core::testing::fork_and_run(service, client));
 }
 
-
-
 /*TEST(Bus, TimeoutThrowsForNullDBusWatch)
 {
     boost::asio::io_service io_service;