Merge lp:~thomas-voss/location-service/factor-out-runtime into lp:location-service/15.04

[Thomas Voß]

- Factor out service::Runtime from daemon.cpp into its own .h/.cpp pair of files.
- Add test cases around correct operation of service::Runtime.

- Factor out service::Runtime from daemon.cpp into its own .h/.cpp pair of files.
- Add test cases around correct operation of service::Runtime.

[lp:~thomas-voss/location-service/factor-out-runtime]

134. By Thomas Voß on 2015-11-25

Reuse service::Runtime in RemoteProviderDaemon.

135. By Thomas Voß on 2015-11-25

Support explicit starting of service::Runtime.

136. By Thomas Voß on 2015-11-25

Fix service::Runtime test cases to explicitly start the runtime.

137. By Thomas Voß on 2015-11-25

Ensure two distinct bus instances.

138. By Thomas Voß on 2015-11-26

Remove dependency on robustify-...

Unmerged revisions

=== modified file 'src/location_service/com/ubuntu/location/CMakeLists.txt'
--- src/location_service/com/ubuntu/location/CMakeLists.txt	2015-11-25 16:02:34 +0000
+++ src/location_service/com/ubuntu/location/CMakeLists.txt	2015-11-25 16:02:35 +0000
@@ -39,6 +39,7 @@
   service/harvester.cpp
   service/demultiplexing_reporter.h
   service/demultiplexing_reporter.cpp
+  service/runtime.cpp
   service/runtime_tests.h
   service/runtime_tests.cpp
   service/trust_store_permission_manager.cpp

=== modified file 'src/location_service/com/ubuntu/location/service/daemon.cpp'
--- src/location_service/com/ubuntu/location/service/daemon.cpp	2015-11-25 16:02:34 +0000
+++ src/location_service/com/ubuntu/location/service/daemon.cpp	2015-11-25 16:02:35 +0000
@@ -34,6 +34,7 @@
 
 #include "program_options.h"
 #include "daemon.h"
+#include "runtime.h"
 
 #include <core/dbus/announcer.h>
 #include <core/dbus/resolver.h>
@@ -78,88 +79,6 @@
     }
 };
 
-// We bundle our "global" runtime dependencies here, specifically
-// a dispatcher to decouple multiple in-process providers from one
-// another , forcing execution to a well known set of threads.
-struct Runtime
-{
-    // Our default concurrency setup.
-    static constexpr const std::uint32_t worker_threads = 2;
-
-    // Our global singleton instance.
-    static Runtime& instance()
-    {
-        static Runtime runtime;
-        return runtime;
-    }
-
-    Runtime()
-        : running{true},
-          service{worker_threads},
-          strand{service},
-          keep_alive{service}
-    {
-        for (unsigned int i = 0; i < worker_threads; i++)
-            workers.push_back(std::thread
-            {
-                [this]()
-                {
-                    while(running)
-                    {
-                        try
-                        {
-                            service.run();
-                            break;
-                        }
-                        catch (const std::exception& e)
-                        {
-                            LOG(WARNING) << e.what();
-                        }
-                        catch (...)
-                        {
-                            LOG(WARNING) << "Unknown exception caught while executing boost::asio::io_service";
-                        }
-                    }
-                }
-            });
-    }
-
-    ~Runtime()
-    {
-        stop();
-    }
-
-    void stop()
-    {
-        VLOG(1) << __PRETTY_FUNCTION__;
-        running = false;
-        service.stop();
-        VLOG(1) << "\t Service stopped.";
-
-        for (auto& worker : workers)
-            if (worker.joinable())
-                worker.join();
-
-        VLOG(1) << "\t Worker threads joined.";
-    }
-
-    // Allows for reusing the runtime in components that require a dispatcher
-    // to control execution of tasks.
-    std::function<void(std::function<void()>)> to_dispatcher_functional()
-    {
-        return [this](std::function<void()> task)
-        {
-            strand.post(task);
-        };
-    }
-
-    bool running;
-    boost::asio::io_service service;
-    boost::asio::io_service::strand strand;
-    boost::asio::io_service::work keep_alive;
-    std::vector<std::thread> workers;
-};
-
 location::ProgramOptions init_daemon_options()
 {
     location::ProgramOptions options;
@@ -263,6 +182,8 @@
         trap->stop();
     });
 
+    auto runtime = location::service::Runtime::create();
+
     const location::Configuration empty_provider_configuration;
 
     std::set<location::Provider::Ptr> instantiated_providers;
@@ -281,7 +202,7 @@
             if (p)
                 instantiated_providers.insert(
                             location::DispatchingProvider::create(
-                                Runtime::instance().to_dispatcher_functional(), p));
+                                runtime->to_dispatcher_functional(), p));
             else
                 throw std::runtime_error("Problem instantiating provider");
 
@@ -291,8 +212,10 @@
         }
     }
 
-    config.incoming->install_executor(dbus::asio::make_executor(config.incoming, Runtime::instance().service));
-    config.outgoing->install_executor(dbus::asio::make_executor(config.outgoing, Runtime::instance().service));
+    config.incoming->install_executor(dbus::asio::make_executor(config.incoming, runtime->service()));
+    config.outgoing->install_executor(dbus::asio::make_executor(config.outgoing, runtime->service()));
+
+    runtime->start();
 
     location::service::DefaultConfiguration dc;
 

=== modified file 'src/location_service/com/ubuntu/location/service/provider_daemon.cpp'
--- src/location_service/com/ubuntu/location/service/provider_daemon.cpp	2014-10-27 21:58:16 +0000
+++ src/location_service/com/ubuntu/location/service/provider_daemon.cpp	2015-11-25 16:02:35 +0000
@@ -25,6 +25,7 @@
 
 #include <com/ubuntu/location/service/configuration.h>
 #include <com/ubuntu/location/service/program_options.h>
+#include <com/ubuntu/location/service/runtime.h>
 
 #include <core/dbus/asio/executor.h>
 
@@ -65,7 +66,6 @@
     location::service::ProviderDaemon::Configuration result;
 
     result.connection = factory(mutable_daemon_options().bus());
-    result.connection->install_executor(core::dbus::asio::make_executor(result.connection));
 
     auto service = core::dbus::Service::add_service(
                 result.connection,
@@ -108,7 +108,7 @@
     return result;
 }
 
-int location::service::ProviderDaemon::main(const location::service::ProviderDaemon::Configuration& configuration)
+int location::service::ProviderDaemon::main(const location::service::ProviderDaemon::Configuration& config)
 {
     auto trap = core::posix::trap_signals_for_all_subsequent_threads(
     {
@@ -121,27 +121,22 @@
         trap->stop();
     });
 
-    std::thread worker
-    {
-        [configuration]()
-        {
-            configuration.connection->run();
-        }
-    };
+    auto runtime = location::service::Runtime::create(1);
+
+    config.connection->install_executor(core::dbus::asio::make_executor(config.connection, runtime->service()));
 
     auto skeleton = location::providers::remote::skeleton::create_with_configuration(location::providers::remote::skeleton::Configuration
     {
-        configuration.object,
-        configuration.connection,
-        configuration.provider
+        config.object,
+        config.connection,
+        config.provider
     });
 
+    runtime->start();
+
     trap->run();
 
-    configuration.connection->stop();
-
-    if (worker.joinable())
-        worker.join();
+    config.connection->stop();
 
     return EXIT_SUCCESS;
 }

=== added file 'src/location_service/com/ubuntu/location/service/runtime.cpp'
--- src/location_service/com/ubuntu/location/service/runtime.cpp	1970-01-01 00:00:00 +0000
+++ src/location_service/com/ubuntu/location/service/runtime.cpp	2015-11-25 16:02:35 +0000
@@ -0,0 +1,109 @@
+/*
+ * Copyright © 2015 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 PARTIlocationAR 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 <com/ubuntu/location/service/runtime.h>
+
+#include <com/ubuntu/location/logging.h>
+
+namespace culs = com::ubuntu::location::service;
+
+namespace
+{
+// exception_safe_run runs service, catching all exceptions and
+// restarting operation until an explicit shutdown has been requested.
+//
+// TODO(tvoss): Catching all exceptions is risky as they might signal unrecoverable
+// errors. We should enable calling code to decide whether an exception should be considered
+// fatal or not.
+void exception_safe_run(boost::asio::io_service& service)
+{
+    while (true)
+    {
+        try
+        {
+            service.run();
+            // a clean return from run only happens in case of
+            // stop() being called (we are keeping the service alive with
+            // a service::work instance).
+            break;
+        }
+        catch (const std::exception& e)
+        {
+            LOG(WARNING) << e.what();
+        }
+        catch (...)
+        {
+            LOG(WARNING) << "Unknown exception caught while executing boost::asio::io_service";
+        }
+    }
+}
+}
+
+std::shared_ptr<culs::Runtime> culs::Runtime::create(std::uint32_t pool_size)
+{
+    return std::shared_ptr<culs::Runtime>(new culs::Runtime(pool_size));
+}
+
+culs::Runtime::Runtime(std::uint32_t pool_size)
+    : pool_size_{pool_size},
+      service_{pool_size_},
+      strand_{service_},
+      keep_alive_{service_}
+{
+}
+
+culs::Runtime::~Runtime()
+{
+    try
+    {
+        stop();
+    } catch(...)
+    {
+        // Dropping all exceptions to satisfy the nothrow guarantee.
+    }
+}
+
+void culs::Runtime::start()
+{
+    for (unsigned int i = 0; i < pool_size_; i++)
+        workers_.push_back(std::thread{exception_safe_run, std::ref(service_)});
+}
+
+void culs::Runtime::stop()
+{
+    service_.stop();
+
+    for (auto& worker : workers_)
+        if (worker.joinable())
+            worker.join();
+}
+
+std::function<void(std::function<void()>)> culs::Runtime::to_dispatcher_functional()
+{
+    // We have to make sure that we stay alive for as long as
+    // calling code requires the dispatcher to work.
+    auto sp = shared_from_this();
+    return [sp](std::function<void()> task)
+    {
+        sp->strand_.post(task);
+    };
+}
+
+boost::asio::io_service& culs::Runtime::service()
+{
+    return service_;
+}

=== added file 'src/location_service/com/ubuntu/location/service/runtime.h'
--- src/location_service/com/ubuntu/location/service/runtime.h	1970-01-01 00:00:00 +0000
+++ src/location_service/com/ubuntu/location/service/runtime.h	2015-11-25 16:02:35 +0000
@@ -0,0 +1,90 @@
+/*
+ * Copyright © 2015 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 PARTIlocationAR 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]>
+ */
+#ifndef COM_UBUNTU_LOCATION_SERVICE_RUNTIME_H_
+#define COM_UBUNTU_LOCATION_SERVICE_RUNTIME_H_
+
+#include <boost/asio.hpp>
+
+#include <functional>
+#include <memory>
+#include <thread>
+#include <vector>
+
+#include <cstdint>
+
+namespace com
+{
+namespace ubuntu
+{
+namespace location
+{
+namespace service
+{
+// We bundle our "global" runtime dependencies here, specifically
+// a dispatcher to decouple multiple in-process providers from one
+// another , forcing execution to a well known set of threads.
+class Runtime : public std::enable_shared_from_this<Runtime>
+{
+public:
+    // Our default concurrency setup.
+    static constexpr const std::uint32_t worker_threads = 2;
+
+    // create returns a Runtime instance with pool_size worker threads
+    // executing the underlying service.
+    static std::shared_ptr<Runtime> create(std::uint32_t pool_size = worker_threads);
+
+    Runtime(const Runtime&) = delete;
+    Runtime(Runtime&&) = delete;
+    // Tears down the runtime, stopping all worker threads.
+    ~Runtime() noexcept(true);
+    Runtime& operator=(const Runtime&) = delete;
+    Runtime& operator=(Runtime&&) = delete;
+
+    // start executes the underlying io_service on a thread pool with
+    // the size configured at creation time.
+    void start();
+
+    // stop cleanly shuts down a Runtime instance,
+    // joining all worker threads.
+    void stop();
+
+    // to_dispatcher_functional returns a function for integration 
+    // with components that expect a dispatcher for operation. 
+    std::function<void(std::function<void()>)> to_dispatcher_functional();
+
+    // service returns the underlying boost::asio::io_service that is executed
+    // by the Runtime.
+    boost::asio::io_service& service();
+
+private:
+    // Runtime constructs a new instance, firing up pool_size 
+    // worker threads.
+    Runtime(std::uint32_t pool_size);
+
+    std::uint32_t pool_size_;
+    boost::asio::io_service service_;
+    boost::asio::io_service::strand strand_;
+    boost::asio::io_service::work keep_alive_;
+    std::vector<std::thread> workers_;
+};
+}
+}
+}
+}
+
+#endif // COM_UBUNTU_LOCATION_SERVICE_RUNTIME_H_

=== modified file 'tests/CMakeLists.txt'
--- tests/CMakeLists.txt	2015-04-23 17:04:09 +0000
+++ tests/CMakeLists.txt	2015-11-25 16:02:35 +0000
@@ -92,6 +92,7 @@
 LOCATION_SERVICE_ADD_TEST(provider_test provider_test.cpp)
 LOCATION_SERVICE_ADD_TEST(wgs84_test wgs84_test.cpp)
 LOCATION_SERVICE_ADD_TEST(trust_store_permission_manager_test trust_store_permission_manager_test.cpp)
+LOCATION_SERVICE_ADD_TEST(runtime_test runtime_test.cpp)
 
 # Provider-specific test-cases go here.
 if (LOCATION_SERVICE_ENABLE_GPS_PROVIDER)

=== modified file 'tests/acceptance_tests.cpp'
--- tests/acceptance_tests.cpp	2015-11-25 16:02:34 +0000
+++ tests/acceptance_tests.cpp	2015-11-25 16:02:35 +0000
@@ -38,12 +38,14 @@
 #include <com/ubuntu/location/service/stub.h>
 
 #include <core/dbus/announcer.h>
+#include <core/dbus/bus.h>
 #include <core/dbus/fixture.h>
 #include <core/dbus/resolver.h>
 
 #include <core/dbus/asio/executor.h>
 
 #include <core/posix/signal.h>
+#include <core/posix/this_process.h>
 
 #include <core/testing/cross_process_sync.h>
 #include <core/testing/fork_and_run.h>
@@ -803,8 +805,8 @@
         };
 
         cul::service::Daemon::Configuration config;
-        config.incoming = session_bus();
-        config.outgoing = session_bus();
+        config.incoming = std::make_shared<core::dbus::Bus>(core::posix::this_process::env::get_or_throw("DBUS_SESSION_BUS_ADDRESS"));
+        config.outgoing = std::make_shared<core::dbus::Bus>(core::posix::this_process::env::get_or_throw("DBUS_SESSION_BUS_ADDRESS"));
         config.is_testing_enabled = false;
         config.providers =
         {

=== modified file 'tests/position_test.cpp'
--- tests/position_test.cpp	2014-06-20 07:40:34 +0000
+++ tests/position_test.cpp	2015-11-25 16:02:35 +0000
@@ -40,7 +40,7 @@
         cul::wgs84::Latitude{},
         cul::wgs84::Longitude{},
         cul::wgs84::Altitude{}};
-    EXPECT_TRUE(p.altitude);
+    EXPECT_TRUE(p.altitude ? true : false);
 }
 
 #include <com/ubuntu/location/codec.h>

=== added file 'tests/runtime_test.cpp'
--- tests/runtime_test.cpp	1970-01-01 00:00:00 +0000
+++ tests/runtime_test.cpp	2015-11-25 16:02:35 +0000
@@ -0,0 +1,118 @@
+/*
+ * Copyright © 2015Canonical 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 <com/ubuntu/location/service/runtime.h>
+
+#include <gtest/gtest.h>
+
+#include <condition_variable>
+#include <thread>
+
+namespace culs = com::ubuntu::location::service;
+
+TEST(Runtime, cleanly_shuts_down_threads)
+{
+    culs::Runtime::create();
+}
+
+TEST(Runtime, executes_service)
+{
+    std::mutex m;
+    std::unique_lock<std::mutex> ul{m};
+    std::condition_variable wc;
+
+    bool signaled = false;
+
+    auto rt = culs::Runtime::create();
+    rt->start();
+    boost::asio::deadline_timer timer{rt->service(), boost::posix_time::milliseconds(500)};
+    timer.async_wait([&wc, &signaled](const boost::system::error_code&)
+    {
+        signaled = true;
+        wc.notify_all();
+    });
+
+    auto result = wc.wait_for(ul, std::chrono::seconds{1}, [&signaled]() { return signaled; });
+    EXPECT_TRUE(result);
+}
+
+TEST(Runtime, catches_exceptions_thrown_from_handlers)
+{
+    std::mutex m;
+    std::unique_lock<std::mutex> ul{m};
+    std::condition_variable wc;
+
+    bool signaled = false;
+
+    auto rt = culs::Runtime::create();
+    rt->start();
+    boost::asio::deadline_timer fast{rt->service(), boost::posix_time::milliseconds(100)};
+    fast.async_wait([](const boost::system::error_code&)
+    {
+        throw std::runtime_error{"Should not propagate"};
+    });
+
+    boost::asio::deadline_timer slow{rt->service(), boost::posix_time::milliseconds(500)};
+    slow.async_wait([&wc, &signaled](const boost::system::error_code&)
+    {
+        signaled = true;
+        wc.notify_all();
+    });
+
+    auto result = wc.wait_for(ul, std::chrono::seconds{1}, [&signaled]() { return signaled; });
+    EXPECT_TRUE(result);
+}
+
+// sets_up_pool_of_threads ensures that the pool size 
+// passed to the Runtime on construction is honored. The idea is simple:
+// We set up two deadline timers, fast and slow. fast fires before slow,
+// with fast blocking in a wait on a common condition_variable. When slow 
+// fires, it notifies the condition variable, thereby unblocking the handler of fast,
+// enabling clean shutdown without errors and timeouts. This only works if the 
+// pool contains at least 2 threads. Otherwise, the handler of slow would not be executed
+// until the handler of fast times out in the wait, marking the test as failed.
+TEST(Runtime, sets_up_pool_of_threads)
+{
+    struct State
+    {
+        bool signaled{false};
+        std::mutex m;
+        std::condition_variable wc;
+    };
+
+    auto state = std::make_shared<State>();
+
+    auto rt = culs::Runtime::create(2);
+    rt->start();
+    boost::asio::deadline_timer fast{rt->service(), boost::posix_time::milliseconds(100)};
+    fast.async_wait([state](const boost::system::error_code&)
+    {
+        std::unique_lock<std::mutex> ul{state->m};
+        EXPECT_TRUE(state->wc.wait_for(ul, std::chrono::seconds{1}, [state]() { return state->signaled; }));
+    });
+
+    boost::asio::deadline_timer slow{rt->service(), boost::posix_time::milliseconds(500)};
+    slow.async_wait([state](const boost::system::error_code&)
+    {
+        state->signaled = true;
+        state->wc.notify_all();
+    });
+
+    std::unique_lock<std::mutex> ul{state->m};
+    auto result = state->wc.wait_for(ul, std::chrono::seconds{1}, [state]() { return state->signaled; });
+    EXPECT_TRUE(result);
+}