Mir

Merge lp:~raof/mir/observer-multiplexer into lp:mir

observer-multiplexer
Merge into development-branch

Proposed by Chris Halse Rogers on 2016-10-04

Status:	Merged
Approved by:	Daniel van Vugt on 2016-10-31
Approved revision:	3771
Merged at revision:	3795
Proposed branch:	lp:~raof/mir/observer-multiplexer
Merge into:	lp:mir
Prerequisite:	lp:~raof/mir/shared-mutex
Diff against target:	1064 lines (+940/-2) 12 files modified include/server/mir/executor.h (+57/-0) include/server/mir/main_loop.h (+6/-2) include/server/mir/observer_registrar.h (+92/-0) src/include/server/mir/glib_main_loop.h (+2/-0) src/include/server/mir/observer_multiplexer.h (+126/-0) src/server/CMakeLists.txt (+5/-0) src/server/glib_main_loop.cpp (+16/-0) tests/include/mir/test/doubles/mock_main_loop.h (+8/-0) tests/include/mir/test/doubles/triggered_main_loop.h (+4/-0) tests/mir_test_doubles/triggered_main_loop.cpp (+15/-0) tests/unit-tests/CMakeLists.txt (+1/-0) tests/unit-tests/test_observer_multiplexer.cpp (+608/-0)
To merge this branch:	bzr merge lp:~raof/mir/observer-multiplexer
Related bugs:	Link a bug report

Reviewer	Review Type	Status
Mir CI Bot	continuous-integration	Approve on 2016-10-31
Andreas Pokorny (community)		Approve on 2016-10-26
Alan Griffiths		Abstain on 2016-10-24
Cemil Azizoglu (community)		Approve on 2016-10-13
Chris Halse Rogers		Abstain on 2016-10-13
Review via email: mp+307501@code.launchpad.net

Commit Message

Add ObserverRegistrar interface and an ObserverMultiplexer helper implementation.

An ObserverRegistrar manages registering and unregistering Observers of some Mir state.

The ObserverMultiplexer is an implementation of ObserverRegistrar which itself is an Observer, so can be handed out to internal Mir components as a singular Observer and to shells using Mir as an ObserverRegistrar.

At the moment ObserverMultiplexer requires code to implement Observer be provided manually. In The Glorious Future, C++ will provide compile-time reflection which will make it possible to automatically implement.

Mir CI Bot (mir-ci-bot) wrote on 2016-10-04:

Click here to trigger a rebuild:
https://mir-jenkins.ubuntu.com/job/mir-ci/1876/rebuild

review: Approve (continuous-integration)

Chris Halse Rogers (raof) wrote on 2016-10-05:

I add a needs-information here for a design discussion:

I have been tempted to make the ObserverMultiplexer offload notification to the server event loop. This means we never have to care about the context in which the observer is called.

Observers loose the ability to deny an operation by throwing an exception, and get notified slightly later.

review: Needs Information

Andreas Pokorny (andreas-pokorny) wrote on 2016-10-05:

> I add a needs-information here for a design discussion:
>
> I have been tempted to make the ObserverMultiplexer offload notification to
> the server event loop. This means we never have to care about the context in
> which the observer is called.

If you could unify the scheduling part of MultiplexingDispatchable and the server event loop - that would be great. Then again I think for users of libmirserver the server event loop might be sufficient, since they dont have access to other mir threads.

Have you considered letting the owner of the observer specify the thread(pool) of execution on registry?

> Observers loose the ability to deny an operation by throwing an exception, and
> get notified slightly later.

I believe the first sounds more like a policy interface. The delayed / asynchronous notification of observers is necessary for the use case of this utility. We cannot make a reasonable local decision that it is acceptable to synchronously execute unknown code added by 3rd parties. The case would be different for previously mentioned policy interfaces that help with a specific decision...

The problem of delayed execution and needing to know "are we done yet?" is something that we might have to address on a higher level?

Mir CI Bot (mir-ci-bot) wrote on 2016-10-12:

Click here to trigger a rebuild:
https://mir-jenkins.ubuntu.com/job/mir-ci/1949/rebuild

review: Needs Fixing (continuous-integration)

Chris Halse Rogers (raof) wrote on 2016-10-12:

There you go, now with a C++17-ish Executor abstraction.

Mir CI Bot (mir-ci-bot) wrote on 2016-10-12:

Click here to trigger a rebuild:
https://mir-jenkins.ubuntu.com/job/mir-ci/1950/rebuild

review: Needs Fixing (continuous-integration)

Chris Halse Rogers (raof) wrote on 2016-10-13:

07:32:29 [0;32m[ RUN ] [mAndroidInputReceiverSetup.slow_raw_input_doesnt_cause_frameskipping
07:32:29 /build/mir-xOiOne/mir-0.25.0+vivid2543bzr3769/tests/unit-tests/client/input/test_android_input_receiver.cpp:272: Failure
07:32:29 Expected: (duration) > (1ms), actual: 8-byte object <55-FC 0E-00 00-00 00-00> vs 8-byte object <01-00 00-00 00-00 00-00>
07:32:29 [0;31m[ FAILED ] [mAndroidInputReceiverSetup.slow_raw_input_doesnt_cause_frameskipping (4 ms)

That's not related to this branch.

Mir CI Bot (mir-ci-bot) wrote on 2016-10-13:

Click here to trigger a rebuild:
https://mir-jenkins.ubuntu.com/job/mir-ci/1966/rebuild

review: Needs Fixing (continuous-integration)

Mir CI Bot (mir-ci-bot) wrote on 2016-10-13:

Click here to trigger a rebuild:
https://mir-jenkins.ubuntu.com/job/mir-ci/1969/rebuild

review: Needs Fixing (continuous-integration)

Chris Halse Rogers (raof) wrote on 2016-10-13:

05:00:09 [0;32m[ PASSED ] [m1543 tests.
05:00:09 [0;31m[ FAILED ] [m1 test, listed below:
05:00:09 [0;31m[ FAILED ] [mAndroidInputReceiverSetup.slow_raw_input_doesnt_cause_frameskipping

AAAAand again.

Mir CI Bot (mir-ci-bot) wrote on 2016-10-13:

Click here to trigger a rebuild:
https://mir-jenkins.ubuntu.com/job/mir-ci/1970/rebuild

review: Approve (continuous-integration)

Chris Halse Rogers (raof) wrote on 2016-10-13:

Oh. I no longer need information!

review: Abstain

Cemil Azizoglu (cemil-azizoglu) wrote on 2016-10-13:

Some plumbing for the dependent branch. Got it.

review: Approve

Andreas Pokorny (andreas-pokorny) wrote on 2016-10-18:

What happened to the mircommon symbols.map?

Otherwise lgtm

review: Needs Information

Mir CI Bot (mir-ci-bot) wrote on 2016-10-19:

Click here to trigger a rebuild:
https://mir-jenkins.ubuntu.com/job/mir-ci/2001/rebuild

review: Needs Fixing (continuous-integration)

Mir CI Bot (mir-ci-bot) wrote on 2016-10-19:

Click here to trigger a rebuild:
https://mir-jenkins.ubuntu.com/job/mir-ci/2003/rebuild

review: Approve (continuous-integration)

Alan Griffiths (alan-griffiths) wrote on 2016-10-24:

I'm not yet quite sure what problem this solves. (No followup branch demonstrating use?)

But, if the functionality is useful, then the code looks reasonable.

review: Abstain

Mir CI Bot (mir-ci-bot) wrote on 2016-10-24:

review: Needs Fixing (continuous-integration)

Alan Griffiths (alan-griffiths) wrote on 2016-10-25:

lp:1539321

Mir CI Bot (mir-ci-bot) wrote on 2016-10-25:

review: Needs Fixing (continuous-integration)

Andreas Pokorny (andreas-pokorny) wrote on 2016-10-26:

review: Approve

Daniel van Vugt (vanvugt) wrote on 2016-10-31:

^^^
CI failure is bug 1523621. Try again.

Mir CI Bot (mir-ci-bot) on 2016-10-31:

review: Approve (continuous-integration)

Preview Diff

[H/L] Next/Prev Comment, [J/K] Next/Prev File, [N/P] Next/Prev Hunk

Subscribers

People subscribed via source and target branches

to all changes:

Chris Halse Rogers

Emanuele Antonio Faraone

Gerry Boland

Mir development team

 === added file 'include/server/mir/executor.h'
 --- include/server/mir/executor.h	1970-01-01 00:00:00 +0000
 +++ include/server/mir/executor.h	2016-10-19 06:34:58 +0000
@@ -0,0 +1,57 @@
++/*
++ * Copyright © 2016 Canonical Ltd.
++ *
++ * This program is free software: you can redistribute it and/or modify it
++ * under the terms of the GNU 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 General Public License for more details.
++ *
++ * You should have received a copy of the GNU General Public License
++ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
++ *
++ * Authored by: Christopher James Halse Rogers <christopher.halse.rogers@canonical.com>
++ */
++
++#ifndef MIR_EXECUTOR_H_
++#define MIR_EXECUTOR_H_
++
++#include <functional>
++
++namespace mir
++{
++
++/**
++ * An executor abstraction mostly compatible with C++ proposal N4414
++ *
++ * As specified in http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2015/p0008r0.pdf
++ *
++ * This should hopefully be compatible with whatever the C++ Executors and Schedulers group
++ * come up with as a final design, and will eventually be replaced by the standard version.
++ *
++ * If not, this minimal interface should be easy to implement on top of whatever emerges
++ * from the standards body.
++ */
++class Executor
++{
++public:
++    /**
++     * Schedule some function to be called sometime in the future.
++     *
++     * It is expected that this \b not eagerly execute, and will return
++     * \b before work() is executed.
++     *
++     * \param [in] work     Function to execute in Executor specified environment.
++     */
++    virtual void spawn(std::function<void()>&& work) = 0;
++
++protected:
++    virtual ~Executor() = default;
++};
++
++}
++
++#endif // MIR_EXECUTOR_H_
 === modified file 'include/server/mir/main_loop.h'
 --- include/server/mir/main_loop.h	2015-04-28 07:54:10 +0000
 +++ include/server/mir/main_loop.h	2016-10-19 06:34:58 +0000
@@ -22,12 +22,16 @@
  #include "mir/graphics/event_handler_register.h"
  #include "mir/time/alarm_factory.h"
  #include "mir/server_action_queue.h"
++#include "mir/executor.h"
  namespace mir
+ {
--class MainLoop : public graphics::EventHandlerRegister, public time::AlarmFactory,
--                 public ServerActionQueue
++class MainLoop :
++    public graphics::EventHandlerRegister,
++    public time::AlarmFactory,
++    public ServerActionQueue,
++    public Executor
+ {
  public:
      virtual void run() = 0;
 === added file 'include/server/mir/observer_registrar.h'
 --- include/server/mir/observer_registrar.h	1970-01-01 00:00:00 +0000
 +++ include/server/mir/observer_registrar.h	2016-10-19 06:34:58 +0000
@@ -0,0 +1,92 @@
++/*
++ * Copyright © 2016 Canonical Ltd.
++ *
++ * This program is free software: you can redistribute it and/or modify it
++ * under the terms of the GNU 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 General Public License for more details.
++ *
++ * You should have received a copy of the GNU General Public License
++ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
++ *
++ * Authored by: Christopher James Halse Rogers <christopher.halse.rogers@canonical.com>
++ */
++
++#ifndef MIR_OBSERVER_REGISTRAR_H_
++#define MIR_OBSERVER_REGISTRAR_H_
++
++#include <memory>
++
++namespace mir
++{
++class Executor;
++
++/**
++ * Register observers for a subsystem.
++ *
++ * \tparam Observer The Observer type to register
++ */
++template<class Observer>
++class ObserverRegistrar
++{
++public:
++    /**
++     * Add an observer to the set notified of all observations
++     *
++     * The ObserverRegistrar does not take any ownership of \p observer, and will
++     * automatically remove it when \p observer expires.
++     *
++     * \param [in] observer The observer to register
++     */
++    virtual void register_interest(std::weak_ptr<Observer> const& observer) = 0;
++
++    /**
++     * Add an observer with specified execution environment
++     *
++     * This is threadsafe and can be called in any context.
++     *
++     * The ObserverRegistrar does not take any ownership of \p observer, and will
++     * automatically remove it when \p observer expires.
++     *
++     * All calls to \p observer methods are performed in the context of
++     * \p executor.
++     *
++     * The \p executor should process work in a delayed fashion. Particularly,
++     * executor::spawn(work) is expected to \b not run \p work in the current
++     * stack. Eager execution of work may result in deadlocks if calls to the
++     * observer result in calls into the ObserverRegistrar.
++     *
++     * \param [in] observer The observer to register
++     * \param [in] executor Execution environment for calls to \p observer methods.
++     *                          The caller is responsible for ensuring \p executor outlives
++     *                          \p observer.
++     */
++    virtual void register_interest(
++        std::weak_ptr<Observer> const& observer,
++        Executor& executor) = 0;
++
++    /**
++     * Remove an observer from the set notified of all observations.
++     *
++     * This is threadsafe and can be called in any context.
++     * It is \b not guaranteed that methods of \p observer will not be called after
++     * this returns.
++     *
++     * \param observer [in] The observer to unregister
++     */
++    virtual void unregister_interest(Observer const& observer) = 0;
++
++protected:
++    ObserverRegistrar() = default;
++    virtual ~ObserverRegistrar() = default;
++    ObserverRegistrar(ObserverRegistrar const&) = delete;
++    ObserverRegistrar& operator=(ObserverRegistrar const&) = delete;
++};
++
++}
++
++#endif //MIR_OBSERVER_REGISTRAR_H_
 === modified file 'src/include/server/mir/glib_main_loop.h'
 --- src/include/server/mir/glib_main_loop.h	2016-07-18 07:38:38 +0000
 +++ src/include/server/mir/glib_main_loop.h	2016-10-19 06:34:58 +0000
@@ -85,6 +85,8 @@
      std::unique_ptr<mir::time::Alarm> create_alarm(
          std::unique_ptr<LockableCallback> callback) override;
++    void spawn(std::function<void()>&& work) override;
++
      void reprocess_all_sources();
  private:
 === added file 'src/include/server/mir/observer_multiplexer.h'
 --- src/include/server/mir/observer_multiplexer.h	1970-01-01 00:00:00 +0000
 +++ src/include/server/mir/observer_multiplexer.h	2016-10-19 06:34:58 +0000
@@ -0,0 +1,126 @@
++/*
++ * Copyright © 2016 Canonical Ltd.
++ *
++ * This program is free software: you can redistribute it and/or modify it
++ * under the terms of the GNU 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 General Public License for more details.
++ *
++ * You should have received a copy of the GNU General Public License
++ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
++ *
++ * Authored by: Christopher James Halse Rogers <christopher.halse.rogers@canonical.com>
++ */
++
++#ifndef MIR_OBSERVER_MULTIPLEXER_H_
++#define MIR_OBSERVER_MULTIPLEXER_H_
++
++#include "mir/observer_registrar.h"
++#include "mir/raii.h"
++#include "mir/posix_rw_mutex.h"
++#include "mir/executor.h"
++#include "mir/main_loop.h"
++
++#include <vector>
++#include <algorithm>
++#include <mutex>
++#include <thread>
++#include <shared_mutex>
++
++namespace mir
++{
++template<class Observer>
++class ObserverMultiplexer : public ObserverRegistrar<Observer>, public Observer
++{
++public:
++    void register_interest(std::weak_ptr<Observer> const& observer) override;
++    void register_interest(
++        std::weak_ptr<Observer> const& observer,
++        Executor& executor) override;
++    void unregister_interest(Observer const& observer) override;
++
++protected:
++    /**
++     * \param [in] default_executor Executor that will be used as the execution environment
++     *                                  for any observer that does not specify its own.
++     * \note \p default_executor must outlive any observer.
++     */
++    explicit ObserverMultiplexer(Executor& default_executor)
++        : default_executor{default_executor}
++    {
++    }
++
++    /**
++     *  Invoke a member function of Observer on each registered observer.
++     *
++     * \tparam MemberFn Must be (Observer::*)(Args...)
++     * \tparam Args     Parameter pack of arguments of Observer member function.
++     * \param f         Pointer to Observer member function to invoke.
++     * \param args  Arguments for member function invocation.
++     */
++    template<typename MemberFn, typename... Args>
++    void for_each_observer(MemberFn f, Args&&... args);
++private:
++    Executor& default_executor;
++
++    PosixRWMutex observer_mutex;
++    std::vector<std::pair<Executor&, std::weak_ptr<Observer>>> observers;
++};
++
++template<class Observer>
++void ObserverMultiplexer<Observer>::register_interest(std::weak_ptr<Observer> const& observer)
++{
++    register_interest(observer, default_executor);
++}
++
++template<class Observer>
++void ObserverMultiplexer<Observer>::register_interest(
++    std::weak_ptr<Observer> const& observer,
++    Executor& executor)
++{
++    std::lock_guard<decltype(observer_mutex)> lock{observer_mutex};
++
++    observers.emplace_back(std::make_pair(std::ref(executor), observer));
++}
++
++template<class Observer>
++void ObserverMultiplexer<Observer>::unregister_interest(Observer const& observer)
++{
++    std::lock_guard<decltype(observer_mutex)> lock{observer_mutex};
++    observers.erase(
++        std::remove_if(
++            observers.begin(),
++            observers.end(),
++            [&observer](auto const& candidate)
++            {
++                auto const resolved_candidate = candidate.second.lock().get();
++                return (resolved_candidate == nullptr) || (resolved_candidate == &observer);
++            }),
++        observers.end());
++}
++
++template<class Observer>
++template<typename MemberFn, typename... Args>
++void ObserverMultiplexer<Observer>::for_each_observer(MemberFn f, Args&&... args)
++{
++    static_assert(
++        std::is_member_function_pointer<MemberFn>::value,
++        "f must be of type (Observer::*)(Args...), a pointer to an Observer member function.");
++    auto const invokable_mem_fn = std::mem_fn(f);
++    std::shared_lock<decltype(observer_mutex)> lock{observer_mutex};
++    for (auto& observer_pair: observers)
++    {
++        if (auto observer = observer_pair.second.lock())
++        {
++            observer_pair.first.spawn(std::bind(invokable_mem_fn, observer.get(), std::forward<Args>(args)...));
++        }
++    }
++}
++}
++
++
++#endif //MIR_OBSERVER_MULTIPLEXER_H_
 === modified file 'src/server/CMakeLists.txt'
 --- src/server/CMakeLists.txt	2016-10-12 06:03:15 +0000
 +++ src/server/CMakeLists.txt	2016-10-19 06:34:58 +0000
@@ -61,6 +61,11 @@
    basic_callback.cpp
    ${PROJECT_SOURCE_DIR}/include/server/mir/time/alarm_factory.h
    ${PROJECT_SOURCE_DIR}/include/server/mir/time/alarm.h
++  ${PROJECT_SOURCE_DIR}/include/server/mir/observer_registrar.h
++  ${PROJECT_SOURCE_DIR}/include/server/mir/executor.h
++  ${PROJECT_SOURCE_DIR}/src/include/server/mir/observer_multiplexer.h
++  ${PROJECT_SOURCE_DIR}/src/include/server/mir/glib_main_loop.h
++  ${PROJECT_SOURCE_DIR}/src/include/server/mir/glib_main_loop_sources.h
+ )
  set(MIR_SERVER_OBJECTS
 === modified file 'src/server/glib_main_loop.cpp'
 --- src/server/glib_main_loop.cpp	2016-07-18 07:38:38 +0000
 +++ src/server/glib_main_loop.cpp	2016-10-19 06:34:58 +0000
@@ -340,3 +340,19 @@
      main_loop_exception = e;
      stop();
+ }
++
++void mir::GLibMainLoop::spawn(std::function<void()>&& work)
++{
++    auto const action_with_exception_handling =
++        [this, action = std::move(work)]
++        {
++            try { action(); }
++            catch (...) { handle_exception(std::current_exception()); }
++        };
++
++    detail::add_server_action_gsource(
++        main_context,
++        nullptr,
++        action_with_exception_handling,
++        [](auto) { return true; });
++}
 === modified file 'tests/include/mir/test/doubles/mock_main_loop.h'
 --- tests/include/mir/test/doubles/mock_main_loop.h	2016-08-11 07:24:14 +0000
 +++ tests/include/mir/test/doubles/mock_main_loop.h	2016-10-19 06:34:58 +0000
@@ -60,6 +60,8 @@
      MOCK_METHOD1(pause_processing_for,void (void const*));
      MOCK_METHOD1(resume_processing_for,void (void const*));
++    MOCK_METHOD1(spawn, void(std::function<void()>&));
++
      MOCK_METHOD1(create_alarm, std::unique_ptr<time::Alarm>(std::function<void()> const& callback));
      MOCK_METHOD1(create_alarm, std::unique_ptr<time::Alarm>(LockableCallback* callback));
@@ -82,6 +84,12 @@
+     {
          register_fd_handler_module_ptr(fds, owner, *handler);
+     }
++
++    void spawn(std::function<void()>&& work) override
++    {
++        std::function<void()> work_copy{std::move(work)};
++        spawn(work_copy);
++    }
  };
+ }
 === modified file 'tests/include/mir/test/doubles/triggered_main_loop.h'
 --- tests/include/mir/test/doubles/triggered_main_loop.h	2015-06-25 03:00:08 +0000
 +++ tests/include/mir/test/doubles/triggered_main_loop.h	2016-10-19 06:34:58 +0000
@@ -42,9 +42,12 @@
      std::unique_ptr<mir::time::Alarm> create_alarm(callback const& call) override;
      void enqueue(void const* owner, ServerAction const& action) override;
++    void spawn(std::function<void()>&& work) override;
++
      void trigger_pending_fds();
      void fire_all_alarms();
      void trigger_server_actions();
++    void trigger_spawned_work();
  private:
      std::vector<callback> timeout_callbacks;
@@ -57,6 +60,7 @@
      };
      std::vector<Item> fd_callbacks;
      std::vector<ServerAction> actions;
++    std::vector<std::function<void()>> work;
  };
+ }
 === modified file 'tests/mir_test_doubles/triggered_main_loop.cpp'
 --- tests/mir_test_doubles/triggered_main_loop.cpp	2016-08-25 08:16:10 +0000
 +++ tests/mir_test_doubles/triggered_main_loop.cpp	2016-10-19 06:34:58 +0000
@@ -101,3 +101,18 @@
      for(auto const& callback : timeout_callbacks)
          callback();
+ }
++
++void mtd::TriggeredMainLoop::spawn(std::function<void()>&& work)
++{
++    base::spawn(std::function<void()>{work});
++    this->work.emplace_back(std::move(work));
++}
++
++void mtd::TriggeredMainLoop::trigger_spawned_work()
++{
++    for (auto const& action : work)
++    {
++        action();
++    }
++    work.clear();
++}
 === modified file 'tests/unit-tests/CMakeLists.txt'
 --- tests/unit-tests/CMakeLists.txt	2016-10-13 00:44:47 +0000
 +++ tests/unit-tests/CMakeLists.txt	2016-10-19 06:34:58 +0000
@@ -80,6 +80,7 @@
    test_module_deleter.cpp
    test_mir_cookie.cpp
    test_posix_rw_mutex.cpp
++  test_observer_multiplexer.cpp
+ )
  CMAKE_DEPENDENT_OPTION(
 === added file 'tests/unit-tests/test_observer_multiplexer.cpp'
 --- tests/unit-tests/test_observer_multiplexer.cpp	1970-01-01 00:00:00 +0000
 +++ tests/unit-tests/test_observer_multiplexer.cpp	2016-10-19 06:34:58 +0000
@@ -0,0 +1,608 @@
++/*
++ * Copyright © 2016 Canonical Ltd.
++ *
++ * This program is free software: you can redistribute it and/or modify it
++ * under the terms of the GNU 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 General Public License for more details.
++ *
++ * You should have received a copy of the GNU General Public License
++ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
++ *
++ * Authored by: Christopher James Halse Rogers <christopher.halse.rogers@canonical.com>
++ */
++
++#include "mir/observer_multiplexer.h"
++
++#include "mir/test/barrier.h"
++#include "mir/test/auto_unblock_thread.h"
++
++#include <memory>
++#include <array>
++#include <thread>
++#include <atomic>
++#include <queue>
++
++#include <gtest/gtest.h>
++#include <gmock/gmock.h>
++
++namespace mt = mir::test;
++
++namespace
++{
++class TestObserver
++{
++public:
++    virtual ~TestObserver() = default;
++
++    virtual void observation_made(std::string const& arg) = 0;
++    virtual void multi_argument_observation(std::string const& arg, int another_one, float third) = 0;
++};
++
++class MockObserver : public TestObserver
++{
++public:
++    MOCK_METHOD1(observation_made, void(std::string const&));
++    MOCK_METHOD3(multi_argument_observation, void(std::string const&, int, float));
++};
++
++class ThreadedExecutor : public mir::Executor
++{
++public:
++    ThreadedExecutor()
++    {
++        worker = std::thread{
++            [this]()
++            {
++                while (running)
++                {
++                    std::function<void()> work{nullptr};
++                    {
++                        std::unique_lock<decltype(work_mutex)> lock{work_mutex};
++                        work_changed.wait(lock, [this]() { return (work_pending.size() > 0) || !running; });
++
++                        if (work_pending.size() > 0)
++                        {
++                            work = work_pending.front();
++                        }
++                    }
++
++                    if (work)
++                    {
++                        work();
++
++                        {
++                            std::lock_guard<decltype(work_mutex)> lock{work_mutex};
++                            work_pending.pop();
++                            work_changed.notify_all();
++                        }
++                        work_changed.notify_all();
++                    }
++                }
++            }
++        };
++    }
++
++    void drain_work()
++    {
++        std::unique_lock<decltype(work_mutex)> lock{work_mutex};
++        work_changed.wait(lock, [this]() { return work_pending.size() == 0; });
++    }
++
++    ~ThreadedExecutor()
++    {
++        drain_work();
++
++        running = false;
++        work_changed.notify_all();
++
++        worker.join();
++    }
++
++    void spawn(std::function<void()>&& work) override
++    {
++        std::lock_guard<decltype(work_mutex)> lock{work_mutex};
++        work_pending.emplace(std::move(work));
++        work_changed.notify_all();
++    }
++
++private:
++    std::thread worker;
++    std::atomic<bool> running{true};
++
++    std::mutex work_mutex;
++    std::condition_variable work_changed;
++    std::queue<std::function<void()>> work_pending;
++};
++
++class TestObserverMultiplexer : public mir::ObserverMultiplexer<TestObserver>
++{
++public:
++    TestObserverMultiplexer(mir::Executor& executor)
++        : ObserverMultiplexer(executor)
++    {
++    }
++
++    void observation_made(std::string const& arg) override
++    {
++        for_each_observer(&TestObserver::observation_made, arg);
++    }
++
++    void multi_argument_observation(std::string const& arg, int another_one, float third) override
++    {
++        for_each_observer(&TestObserver::multi_argument_observation, arg, another_one, third);
++    }
++};
++}
++
++TEST(ObserverMultiplexer, each_added_observer_recieves_observations)
++{
++    using namespace testing;
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++    std::string const value = "Hello, my name is Inigo Montoya.";
++
++
++    auto observer_one = std::make_shared<NiceMock<MockObserver>>();
++    auto observer_two = std::make_shared<NiceMock<MockObserver>>();
++
++    EXPECT_CALL(*observer_one, observation_made(StrEq(value))).Times(2);
++    EXPECT_CALL(*observer_two, observation_made(StrEq(value))).Times(1);
++
++
++    multiplexer.register_interest(observer_one);
++    multiplexer.observation_made(value);
++
++    multiplexer.register_interest(observer_two);
++    multiplexer.observation_made(value);
++
++    executor.drain_work();
++}
++
++TEST(ObserverMultiplexer, removed_observers_do_not_recieve_observations)
++{
++    using namespace testing;
++    std::string const value = "The girl of my dreams is giving me nightmares";
++
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++
++    auto observer_one = std::make_shared<NiceMock<MockObserver>>();
++    auto observer_two = std::make_shared<NiceMock<MockObserver>>();
++
++    EXPECT_CALL(*observer_one, observation_made(StrEq(value))).Times(2);
++    EXPECT_CALL(*observer_two, observation_made(StrEq(value))).Times(1);
++
++    multiplexer.register_interest(observer_one);
++    multiplexer.register_interest(observer_two);
++    multiplexer.observation_made(value);
++
++    multiplexer.unregister_interest(*observer_two);
++    multiplexer.observation_made(value);
++
++    executor.drain_work();
++}
++
++TEST(ObserverMultiplexer, can_remove_observer_from_callback)
++{
++    using namespace testing;
++    std::string const value = "Goldfinger";
++
++    auto observer_one = std::make_shared<NiceMock<MockObserver>>();
++    auto observer_two = std::make_shared<NiceMock<MockObserver>>();
++
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++
++    EXPECT_CALL(*observer_one, observation_made(StrEq(value)))
++        .WillOnce(Invoke(
++            [observer_one = observer_one.get(), &multiplexer, &value](auto)
++            {
++                multiplexer.unregister_interest(*observer_one);
++                multiplexer.observation_made(value);
++            }));
++    EXPECT_CALL(*observer_two, observation_made(StrEq(value)))
++        .Times(2);
++
++    multiplexer.register_interest(observer_one);
++    multiplexer.register_interest(observer_two);
++
++    multiplexer.observation_made(value);
++
++    executor.drain_work();
++}
++
++TEST(ObserverMultiplexer, multiple_threads_can_simultaneously_make_observations)
++{
++    using namespace testing;
++
++    std::array<bool, 10> values_seen;
++    std::array<std::string, values_seen.size()> values;
++    for (auto i = 0u; i < values.size(); ++i)
++    {
++        values[i] = std::to_string(i);
++    }
++
++    auto observer = std::make_shared<NiceMock<MockObserver>>();
++    ON_CALL(*observer, observation_made(_))
++        .WillByDefault(
++            Invoke(
++                [&values, &values_seen](auto const& value)
++                {
++                    for (auto i = 0u; i < values.size(); ++i)
++                    {
++                        if (value == values[i])
++                        {
++                            values_seen[i] = true;
++                        }
++                    }
++                }));
++
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++    multiplexer.register_interest(observer);
++
++    mt::Barrier threads_ready(values.size());
++    mt::Barrier observations_made(values.size() + 1);
++    std::array<mt::AutoJoinThread, values_seen.size()> threads;
++
++    for (auto i = 0u; i < values.size(); ++i)
++    {
++        threads[i] = mt::AutoJoinThread(
++            [&threads_ready, &observations_made, &multiplexer, value = values[i]]()
++            {
++                threads_ready.ready();
++                multiplexer.observation_made(value);
++                observations_made.ready();
++            });
++    }
++    observations_made.ready();
++
++    executor.drain_work();
++
++    EXPECT_THAT(
++        std::vector<bool>(values_seen.begin(), values_seen.end()),
++        ContainerEq(std::vector<bool>(values_seen.size(), true)));
++}
++
++TEST(ObserverMultiplexer, multiple_threads_registering_unregistering_and_observing)
++{
++    using namespace testing;
++
++    auto observer_one = std::make_shared<NiceMock<MockObserver>>();
++    auto observer_two = std::make_shared<NiceMock<MockObserver>>();
++
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++
++    ON_CALL(*observer_one, observation_made(_))
++        .WillByDefault(
++            Invoke(
++                [&multiplexer, observer_two](std::string const& value)
++                {
++                    if (!value.compare("3"))
++                    {
++                        multiplexer.unregister_interest(*observer_two);
++                    }
++                    if (!value.compare("5"))
++                    {
++                        multiplexer.register_interest(observer_two);
++                    }
++                }));
++    ON_CALL(*observer_one, observation_made(_))
++        .WillByDefault(
++            Invoke(
++                [&multiplexer, observer_two](std::string const& value)
++                {
++                    if (!value.compare("8"))
++                    {
++                        multiplexer.unregister_interest(*observer_two);
++                    }
++                }));
++
++    multiplexer.register_interest(observer_two);
++    multiplexer.register_interest(observer_one);
++
++    std::array<mt::AutoJoinThread, 10> threads;
++    mt::Barrier threads_done(threads.size() + 1);
++
++    for (auto i = 0u; i < threads.size(); ++i)
++    {
++        threads[i] = mt::AutoJoinThread(
++            [&threads_done, &multiplexer]()
++            {
++                for (auto i = 0; i < 50; ++i)
++                {
++                    for (auto j = 0; j < 10; ++j)
++                    {
++                        multiplexer.observation_made(std::to_string(j));
++                    }
++                }
++                threads_done.ready();
++            });
++    }
++    threads_done.ready();
++
++    executor.drain_work();
++}
++
++TEST(ObserverMultiplexer, multiple_threads_unregistering_same_observer_is_safe)
++{
++    using namespace testing;
++
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++
++    auto observer_one = std::make_shared<NiceMock<MockObserver>>();
++    auto observer_two = std::make_shared<NiceMock<MockObserver>>();
++
++    std::atomic<int> call_count{0};
++    ON_CALL(*observer_one, observation_made(_))
++        .WillByDefault(
++            Invoke(
++                [&multiplexer, observer_two, &call_count](auto)
++                {
++                    ++call_count;
++                    multiplexer.unregister_interest(*observer_two);
++                }));
++    std::atomic<bool> victim_called{false};
++    ON_CALL(*observer_two, observation_made(_))
++        .WillByDefault(Invoke([&victim_called](auto) { victim_called = true; }));
++
++    multiplexer.register_interest(observer_one);
++    multiplexer.register_interest(observer_two);
++
++    std::array<mt::AutoJoinThread, 10> threads;
++    mt::Barrier threads_done(threads.size() + 1);
++
++    for (auto i = 0u; i < threads.size(); ++i)
++    {
++        threads[i] = mt::AutoJoinThread(
++            [&threads_done, &multiplexer]()
++            {
++                multiplexer.observation_made("Hello");
++                threads_done.ready();
++            });
++    }
++    threads_done.ready();
++
++    executor.drain_work();
++
++    auto precount = call_count.load();
++
++    multiplexer.observation_made("Banana");
++
++    executor.drain_work();
++
++    EXPECT_THAT(call_count, Eq(precount + 1));
++}
++
++TEST(ObserverMultiplexer, registering_is_threadsafe)
++{
++    using namespace testing;
++
++    std::array<bool, 100> observer_notified;
++    std::array<std::shared_ptr<NiceMock<MockObserver>>, observer_notified.size()> observers;
++
++    for (auto i = 0u; i < observers.size(); ++i)
++    {
++        observers[i] = std::make_shared<NiceMock<MockObserver>>();
++        ON_CALL(*observers[i], observation_made(_))
++            .WillByDefault(Invoke([notified = &observer_notified[i]](auto) { *notified = true; }));
++    }
++    std::array<mt::AutoJoinThread, observer_notified.size()> threads;
++    mt::Barrier threads_done(threads.size() + 1);
++
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++
++    for (auto i = 0u; i < threads.size(); ++i)
++    {
++        threads[i] = mt::AutoJoinThread(
++            [&threads_done, &multiplexer, observer = observers[i]]()
++            {
++                multiplexer.register_interest(observer);
++                threads_done.ready();
++            });
++    }
++    threads_done.ready();
++
++    multiplexer.observation_made("Foo");
++
++    executor.drain_work();
++
++    EXPECT_THAT(
++        std::vector<bool>(observer_notified.begin(), observer_notified.end()),
++        ContainerEq(std::vector<bool>(observer_notified.size(), true)));
++}
++
++TEST(ObserverMultiplexer, unregistering_is_threadsafe)
++{
++    using namespace testing;
++
++    std::array<bool, 10> observer_notified;
++    for (auto& notified : observer_notified)
++    {
++        notified = false;
++    }
++
++    std::array<std::shared_ptr<NiceMock<MockObserver>>, observer_notified.size()> observers;
++
++    for (auto i = 0u; i < observers.size(); ++i)
++    {
++        observers[i] = std::make_shared<NiceMock<MockObserver>>();
++        ON_CALL(*observers[i], observation_made(_))
++            .WillByDefault(Invoke([notified = &observer_notified[i]](auto) { *notified = true; }));
++    }
++    std::array<mt::AutoJoinThread, observer_notified.size()> threads;
++    mt::Barrier threads_done(threads.size() + 1);
++
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++
++    for (auto observer : observers)
++    {
++        multiplexer.register_interest(observer);
++    }
++
++    for (auto i = 0u; i < threads.size(); ++i)
++    {
++        threads[i] = mt::AutoJoinThread(
++            [&threads_done, &multiplexer, observer = observers[i]]()
++        {
++            multiplexer.unregister_interest(*observer);
++            threads_done.ready();
++        });
++    }
++    threads_done.ready();
++
++    multiplexer.observation_made("Foo");
++
++    executor.drain_work();
++
++    EXPECT_THAT(
++        std::vector<bool>(observer_notified.begin(), observer_notified.end()),
++        ContainerEq(std::vector<bool>(observer_notified.size(), false)));
++}
++
++TEST(ObserverMultiplexer, can_trigger_observers_from_observers)
++{
++    using namespace testing;
++    constexpr char const* first_observation = "Elementary, my dear Watson";
++    constexpr char const* second_observation = "You're one microscopic cog in his catastrophic plan";
++
++    auto observer = std::make_shared<NiceMock<MockObserver>>();
++
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++
++    EXPECT_CALL(*observer, observation_made(StrEq(first_observation)))
++        .WillOnce(InvokeWithoutArgs([&multiplexer]() { multiplexer.observation_made(second_observation); }));
++    EXPECT_CALL(*observer, observation_made(StrEq(second_observation)));
++
++    multiplexer.register_interest(observer);
++
++    multiplexer.observation_made(first_observation);
++
++    executor.drain_work();
++}
++
++TEST(ObserverMultiplexer, addition_takes_effect_immediately_even_in_callback)
++{
++    using namespace testing;
++    constexpr char const* first_observation = "Rhythm & Blues Alibi";
++    constexpr char const* second_observation = "Blue Moon Rising";
++
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++
++    auto observer_one = std::make_shared<NiceMock<MockObserver>>();
++    auto observer_two = std::make_shared<NiceMock<MockObserver>>();
++
++    EXPECT_CALL(*observer_one, observation_made(StrEq(first_observation)))
++        .WillOnce(
++            InvokeWithoutArgs(
++                [&multiplexer, observer_two]()
++                {
++                    multiplexer.register_interest(observer_two);
++                    multiplexer.observation_made(second_observation);
++                }));
++    EXPECT_CALL(*observer_one, observation_made(Not(StrEq(first_observation))))
++        .Times(AnyNumber());
++    EXPECT_CALL(*observer_two, observation_made(StrEq(second_observation)));
++
++    multiplexer.register_interest(observer_one);
++
++    multiplexer.observation_made(first_observation);
++
++    executor.drain_work();
++}
++
++TEST(ObserverMultiplexer, observations_can_be_delegated_to_specified_executor)
++{
++    using namespace testing;
++
++    class CountingExecutor : public mir::Executor
++    {
++    public:
++        void spawn(std::function<void()>&& work) override
++        {
++            std::lock_guard<decltype(work_mutex)> lock{work_mutex};
++            ++spawn_count;
++            work_queue.emplace(std::move(work));
++        }
++
++        void do_work()
++        {
++            std::unique_lock<decltype(work_mutex)> lock{work_mutex};
++            while (!work_queue.empty())
++            {
++                auto work = work_queue.front();
++                work_queue.pop();
++                lock.unlock();
++
++                work();
++
++                lock.lock();
++            }
++        }
++
++        int work_spawned()
++        {
++            std::lock_guard<decltype(work_mutex)> lock{work_mutex};
++            return spawn_count;
++        }
++    private:
++        int spawn_count{0};
++
++        std::mutex work_mutex;
++        std::queue<std::function<void()>> work_queue;
++    } counting_executor;
++
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++
++    auto observer_one = std::make_shared<NiceMock<MockObserver>>();
++    auto observer_two = std::make_shared<NiceMock<MockObserver>>();
++
++    EXPECT_CALL(*observer_one, observation_made(_));
++    EXPECT_CALL(*observer_two, observation_made(_));
++
++    multiplexer.register_interest(observer_one);
++    multiplexer.register_interest(observer_two, counting_executor);
++
++    EXPECT_THAT(counting_executor.work_spawned(), Eq(0));
++
++    multiplexer.observation_made("Hello!");
++
++    EXPECT_THAT(counting_executor.work_spawned(), Eq(1));
++
++    counting_executor.do_work();
++    executor.drain_work();
++}
++
++TEST(ObserverMultiplexer, multi_argument_observations_work)
++{
++    using namespace testing;
++
++    std::string const a{"The Owls Go"};
++    constexpr int b{55};
++    constexpr float c{3.1415f};
++
++    ThreadedExecutor executor;
++    TestObserverMultiplexer multiplexer{executor};
++
++    auto observer = std::make_shared<NiceMock<MockObserver>>();
++
++    multiplexer.register_interest(observer);
++
++    EXPECT_CALL(*observer, multi_argument_observation(a, b, c));
++
++    multiplexer.multi_argument_observation(a, b, c);
++
++    executor.drain_work();
++}

Mir

Merge lp:~raof/mir/observer-multiplexer into lp:mir

Commit Message

Description of the Change

Preview Diff

Subscribers