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]

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

Remove dependency on robustify-...

[Alberto Mardegan (mardy)]

LGTM!

review: Approve

=== modified file 'src/location_service/com/ubuntu/location/CMakeLists.txt'
--- src/location_service/com/ubuntu/location/CMakeLists.txt	2015-04-23 14:48:44 +0000
+++ src/location_service/com/ubuntu/location/CMakeLists.txt	2015-11-26 08:30:38 +0000
@@ -38,6 +38,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-10-23 13:43:32 +0000
+++ src/location_service/com/ubuntu/location/service/daemon.cpp	2015-11-26 08:30:38 +0000
@@ -20,6 +20,7 @@
 #include <com/ubuntu/location/boost_ptree_settings.h>
 #include <com/ubuntu/location/provider_factory.h>
 
+#include <com/ubuntu/location/logging.h>
 #include <com/ubuntu/location/connectivity/dummy_connectivity_manager.h>
 
 #include <com/ubuntu/location/service/default_configuration.h>
@@ -32,6 +33,7 @@
 
 #include "program_options.h"
 #include "daemon.h"
+#include "runtime.h"
 
 #include <core/dbus/announcer.h>
 #include <core/dbus/resolver.h>
@@ -39,6 +41,8 @@
 
 #include <core/posix/signal.h>
 
+#include <boost/asio.hpp>
+
 #include <system_error>
 #include <thread>
 
@@ -177,6 +181,8 @@
         trap->stop();
     });
 
+    auto runtime = location::service::Runtime::create();
+
     const location::Configuration empty_provider_configuration;
 
     std::set<location::Provider::Ptr> instantiated_providers;
@@ -203,8 +209,10 @@
         }
     }
 
-    config.incoming->install_executor(dbus::asio::make_executor(config.incoming));
-    config.outgoing->install_executor(dbus::asio::make_executor(config.outgoing));
+    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;
 
@@ -222,50 +230,9 @@
     };
 
     auto location_service = std::make_shared<location::service::Implementation>(configuration);
-    // We need to ensure that any exception raised by the executor does not crash the app
-    // and also gets logged.
-    auto execute = [] (std::shared_ptr<core::dbus::Bus> bus) {
-        while(true)
-        {
-            try
-            {
-                VLOG(10) << "Starting a bus executor";
-                bus->run();
-                break; // run() exited normally
-            }
-            catch (const std::exception& e)
-            {
-                LOG(WARNING) << e.what();
-            }
-            catch (...)
-            {
-                LOG(WARNING) << "Unexpected exception was raised by the bus executor";
-            }
-        }
-    };
-
-    std::thread t1{execute, config.incoming};
-    std::thread t2{execute, config.incoming};
-    std::thread t3{execute, config.incoming};
-    std::thread t4{execute, config.outgoing};
 
     trap->run();
 
-    config.incoming->stop();
-    config.outgoing->stop();
-
-    if (t1.joinable())
-        t1.join();
-
-    if (t2.joinable())
-        t2.join();
-
-    if (t3.joinable())
-        t3.join();
-
-    if (t4.joinable())
-        t4.join();
-
     return EXIT_SUCCESS;
 }
 

=== 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-26 08:30:38 +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-26 08:30:38 +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-26 08:30:38 +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-26 08:30:38 +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-02-13 13:19:18 +0000
+++ tests/acceptance_tests.cpp	2015-11-26 08:30:38 +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>
@@ -708,7 +710,7 @@
 
         options.add(Keys::update_period,
                     "Update period length for dummy::Provider setup.",
-                    std::uint32_t{100});
+                    std::uint32_t{10});
 
         options.add(Keys::client_count,
                     "Number of clients that should be fired up.",
@@ -761,6 +763,8 @@
 };
 }
 
+#include "did_finish_successfully.h"
+
 TEST_F(LocationServiceStandaloneLoad, MultipleClientsConnectingAndDisconnectingWorks)
 {
     EXPECT_TRUE(trust_store_is_set_up_for_testing);
@@ -801,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 =
         {
@@ -829,7 +833,7 @@
                     status;
     }, core::posix::StandardStream::empty);
 
-    std::this_thread::sleep_for(std::chrono::seconds{2});
+    std::this_thread::sleep_for(std::chrono::seconds{15});
 
     auto client = [this]()
     {
@@ -957,17 +961,11 @@
     {
         VLOG(1) << "Stopping client...: " << client.pid();
         client.send_signal_or_throw(core::posix::Signal::sig_term);
-        auto result = client.wait_for(core::posix::wait::Flags::untraced);
-
-        EXPECT_EQ(core::posix::wait::Result::Status::exited, result.status);
-        EXPECT_EQ(core::posix::exit::Status::success, result.detail.if_exited.status);
+        EXPECT_TRUE(did_finish_successfully(client.wait_for(core::posix::wait::Flags::untraced)));
     }
 
     VLOG(1) << "Cleaned up clients, shutting down the service...";
 
     server.send_signal_or_throw(core::posix::Signal::sig_term);
-    auto result = server.wait_for(core::posix::wait::Flags::untraced);
-
-    EXPECT_EQ(core::posix::wait::Result::Status::exited, result.status);
-    EXPECT_EQ(core::posix::exit::Status::success, result.detail.if_exited.status);
+    EXPECT_TRUE(did_finish_successfully(server.wait_for(core::posix::wait::Flags::untraced)));
 }

=== modified file 'tests/position_test.cpp'
--- tests/position_test.cpp	2014-06-20 07:40:34 +0000
+++ tests/position_test.cpp	2015-11-26 08:30:38 +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-26 08:30:38 +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);
+}