Merge lp:~mir-team/mir/add-dispatchable-interface into lp:mir

[Chris Halse Rogers]

Add a Dispatchable interface, and transition StreamSocketTransport.
(LP: #1397375)

The Dispatchable interface is useful for anything that can provide a monitorable file descriptor.

A rough guide for the reviewer:

This whole enchilada is necessary because we want to expose a fd based wait/dispatch event model to clients: https://bugs.launchpad.net/bugs/1397375 (This is good for GTK, XMir, and makes a plymouth backend much more feasible).

In turn, this means that all event generators in the client library need to provide some mechanism we can aggregate into the client dispatch fd; epoll is a very nice interface for this, so in the end the event generators will provide an fd that the client library can add to an epoll fd and pass that epoll fd out to the client to wait on.

The meat of this MP is the mir::dispatch::Dispatchable interface, and the implementation of this in mir::client::rpc::StreamSocketTransport.

SimpleDispatchThread is split out of StreamSocketTransport to do the newly-necessary event dispatch for MirConnection.

The next step in this series is to turn the AndroidInputDispatcher into a Dispatchable.

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:2232
No commit message was specified in the merge proposal. Click on the following link and set the commit message (if you want a jenkins rebuild you need to trigger it yourself):
https://code.launchpad.net/~mir-team/mir/add-dispatchable-interface/+merge/246372/+edit-commit-message

http://jenkins.qa.ubuntu.com/job/mir-ci/2636/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-android-vivid-i386-build/837/console
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-vivid-amd64-build/837
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-vivid-touch/799/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/633/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/799/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-ci/2636/rebuild

review: Needs Fixing (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:2241
http://jenkins.qa.ubuntu.com/job/mir-ci/2638/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-android-vivid-i386-build/840
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-vivid-amd64-build/840
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-vivid-touch/802/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/635/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/802/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-ci/2638/rebuild

review: Needs Fixing (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:2242
http://jenkins.qa.ubuntu.com/job/mir-ci/2639/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-android-vivid-i386-build/841
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-vivid-amd64-build/841
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-vivid-touch/803/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/636/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/803/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-ci/2639/rebuild

review: Needs Fixing (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:2243
http://jenkins.qa.ubuntu.com/job/mir-ci/2640/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-android-vivid-i386-build/843
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-vivid-amd64-build/843
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-vivid-touch/805
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/637
        deb: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/637/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/805
        deb: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/805/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-runner-mako/3925
    SUCCESS: http://s-jenkins.ubuntu-ci:8080/job/touch-flash-device/17061

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-ci/2640/rebuild

review: Approve (continuous-integration)

[Alexandros Frantzis (afrantzis)]

A first pass...

This MP introduces another event loop implementation. We are already using the GLib event loop API in the server, perhaps we should also do so in the client. I guess the counter-argument is that we don't want to impose extra dependencies on the client. I am OK with having this custom implementation as long as it doesn't get too complex (in which case we can switch to GLib or other library).

34 +enum fd_event : uint32_t {
41 +using fd_events = uint32_t;

Types should use PascalCase.

76 + virtual fd_events relevant_events() const = 0;

Just noting that this is not used at the moment. Not really an issue since its utility is evident, but still it would be good to use it in wait_for_events_forever() since it's there.

review: Needs Fixing

[Chris Halse Rogers (raof)]

Yeah, it does introduce another event loop. I think that it's sufficiently simple now that it's not really worth replacing with g_main_loop, and although subsequent MPs will extend it, I think they're extending it in ways not supported by g_main_loop anyway.

[Chris Halse Rogers (raof)]

(In particular, in later MPs SimpleDispatchThread becomes ThreadedDispatcer which dispatches from multiple threads, which is not supported for GMainContext)

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:2246
http://jenkins.qa.ubuntu.com/job/mir-ci/2655/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-android-vivid-i386-build/862
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-vivid-amd64-build/862
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-vivid-touch/824
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/652
        deb: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/652/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/824
        deb: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/824/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-runner-mako/3941
    SUCCESS: http://s-jenkins.ubuntu-ci:8080/job/touch-flash-device/17096

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-ci/2655/rebuild

review: Approve (continuous-integration)

[Robert Carr (robertcarr)]

I think some of the tests could be made easier to read without much effort.

test_simple_dispatch_thread:

Since there are no expectations you can use a StubDispatchable and clean up the test body a little. Likewise you can move at least the dispatcher and the signal and uint64_t dummy to the fixture.

test_stream_transport:

+ // A valid fd is >= 0, and we know that stdin, stdout, and stderr aren't correct.

fnctl(fd, O_GETFD) is used elsewhere.

+ uint64_t dummy{0xdeadbeef};

Can be moved to fixture or

1084:
+ EXPECT_EQ(sizeof(dummy), write(this->test_fd, &dummy, sizeof(dummy)));

Perhaps move to fixture...write_to_fd()

1138: + std::this_thread::sleep_for(std::chrono::seconds{1});

:(

1095: + auto observer = std::make_shared<NiceMock<MockObserver>>();

Stub

review: Needs Fixing

[Robert Carr (robertcarr)]

I guess the needs fixing are please fix the stub v. mocks and fixtures in order to clear up test bodies. Will finish review tomorrow :)

[Cemil Azizoglu (cemil-azizoglu)]

Thanks for the guide, it was useful.

Some nits, otherwise all good.

63 + * Dispatch should no longer be called.
Indentation

65 + * specify in \ref event (eg: readable | remote_closed) then dispatch
you mean "specified", I think.

534 +# Copyright © 2014 Canonical Ltd.
s/2014/2015

548 +# Authored by: Robert Carr <email address hidden>
Correct attribution?

562 + * Copyright © 2014 Canonical Ltd.
s/2014/2015

review: Approve

[Alexandros Frantzis (afrantzis)]

693 + shutdown_fd = mir::Fd{pipefds[1]};
<snip>
713 + ::close(shutdown_fd);

This can lead to nasty double close bug: after the call to explicitly close the shutdown_fd, some other thread in the process may open a file and get the fd number that was just closed. When the shutdown_fd destructor is finally called the newly created file will be closed.

1709 + EXPECT_FALSE(dispatched_more_than_enough->wait_for(std::chrono::seconds{1}));
1741 + EXPECT_FALSE(dispatched_closed->wait_for(std::chrono::seconds{1}));

It's difficult (and often not possible) to test that an event will not occur, without waiting for some amount of time. I am a bit torn about this. On the one hand, unit tests should be super-fast. On the other hand one could argue that with the ability to run tests in parallel, small delays don't matter.

> 1095: + auto observer = std::make_shared<NiceMock<MockObserver>>();
>
> Stub

+1

review: Needs Fixing

[Chris Halse Rogers (raof)]

> 1709 +
> EXPECT_FALSE(dispatched_more_than_enough->wait_for(std::chrono::seconds{1}));
> 1741 + EXPECT_FALSE(dispatched_closed->wait_for(std::chrono::seconds{1}));
>
> It's difficult (and often not possible) to test that an event will not occur,
> without waiting for some amount of time. I am a bit torn about this. On the
> one hand, unit tests should be super-fast. On the other hand one could argue
> that with the ability to run tests in parallel, small delays don't matter.

Right. We do ideally want to avoid long tests, but the testsuite still takes ~5 seconds for a full run, which is close enough to instant for me.

[Chris Halse Rogers (raof)]

In case you're wondering about the long stream of commits Launchpad sees, that's an unfortunate artefact of git-remote-bzr; as far as I can tell I can't merge from trunk, I need to rebase on it. It shouldn't be visible in the final bzr log, and I won't use git-remote-bzr for any further branches.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:2275
http://jenkins.qa.ubuntu.com/job/mir-ci/2731/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-android-vivid-i386-build/952
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-vivid-amd64-build/952
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-vivid-touch/914
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/728
        deb: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/728/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/914
        deb: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/914/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-runner-mako/4014
    SUCCESS: http://s-jenkins.ubuntu-ci:8080/job/touch-flash-device/17261

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-ci/2731/rebuild

review: Approve (continuous-integration)

[Alexandros Frantzis (afrantzis)]

Looks good.

review: Approve

[Andreas Pokorny (andreas-pokorny)]

side note:
+ 41 FdEvents could be made type safer with:

http://bazaar.launchpad.net/~andreas-pokorny/mir/evdev-input-platform/view/head:/src/include/common/mir/flags.h

hmm it only lacks toggling and disabling bits..

Why is that in include/, instead of src/include?

review: Needs Information

[Robert Carr (robertcarr)]

I still think it maybe worth trying to clean up the tests a little anyway don't want to block this on that though....:)

review: Approve

[Daniel van Vugt (vanvugt)]

Adding to the public API:
1 === added directory 'include/common/mir/dispatch'
2 === added file 'include/common/mir/dispatch/dispatchable.h'
3 --- include/common/mir/dispatch/dispatchable.h 1970-01-01 00:00:00 +0000
4 +++ include/common/mir/dispatch/dispatchable.h 2015-01-21 01:20:26 +0000
also means we need to update the sha1sums.

review: Needs Fixing

[Chris Halse Rogers (raof)]

@Andreas: I've looked at using mir::Flags<FdEvent>, but it becomes a bit ugly - particularly, you can't easily check against (md::FdEvent::readable | md::FdEvent::writable). Sadly, I think that a better thing would end up using the C pre-processor.

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:2276
http://jenkins.qa.ubuntu.com/job/mir-ci/2768/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-android-vivid-i386-build/1000/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-clang-vivid-amd64-build/999/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-vivid-touch/961/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/765/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/961/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-ci/2768/rebuild

review: Needs Fixing (continuous-integration)

[Daniel van Vugt (vanvugt)]

review: Abstain

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:2277
http://jenkins.qa.ubuntu.com/job/mir-ci/2820/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-android-vivid-i386-build/1076/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-clang-vivid-amd64-build/1075/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-vivid-touch/1037/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/817/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/1037/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-ci/2820/rebuild

review: Needs Fixing (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:2278
http://jenkins.qa.ubuntu.com/job/mir-ci/2833/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-android-vivid-i386-build/1096
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-vivid-amd64-build/1095
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-vivid-touch/1057/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/830/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/1057/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-ci/2833/rebuild

review: Needs Fixing (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:2279
http://jenkins.qa.ubuntu.com/job/mir-ci/2835/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-android-vivid-i386-build/1098
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-vivid-amd64-build/1097
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-vivid-touch/1059
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/832
        deb: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/832/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/1059
        deb: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/1059/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-runner-mako/4114
    SUCCESS: http://s-jenkins.ubuntu-ci:8080/job/touch-flash-device/17533

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/mir-ci/2835/rebuild

review: Approve (continuous-integration)

=== modified file 'common-ABI-sha1sums'
--- common-ABI-sha1sums	2015-01-27 19:46:30 +0000
+++ common-ABI-sha1sums	2015-01-27 23:19:58 +0000
@@ -1,4 +1,6 @@
 3329ada91412ded2f127aee9a92f065e57b81cb2  include/common/mir/cached_ptr.h
+d20a846dfa0e46fb276b4c181a6a5a56e7a884da  include/common/mir/dispatch/dispatchable.h
+7bdc0e58dc228ac655a618208bf133c6a499ca13  include/common/mir/dispatch/simple_dispatch_thread.h
 6473b6b17e72053a4ca3f7dcf8e2d81156fbfb85  include/common/mir/events/event_builders.h
 82ff9499ef62739379616e02164dc98f9914c329  include/common/mir/fd.h
 fe0275d9c64e7a2d99990382d04b1fe956d7b7e4  include/common/mir/frontend/surface_id.h

=== added directory 'include/common/mir/dispatch'
=== added file 'include/common/mir/dispatch/dispatchable.h'
--- include/common/mir/dispatch/dispatchable.h	1970-01-01 00:00:00 +0000
+++ include/common/mir/dispatch/dispatchable.h	2015-01-27 23:19:58 +0000
@@ -0,0 +1,78 @@
+/*
+ * 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 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: Christopher James Halse Rogers <[e-mail]>
+ */
+
+#ifndef MIR_DISPATCH_DISPATCHABLE_H_
+#define MIR_DISPATCH_DISPATCHABLE_H_
+
+#include "mir/fd.h"
+
+namespace mir
+{
+namespace dispatch
+{
+
+enum FdEvent : uint32_t {
+    readable =      1<<0,
+    writable =      1<<1,
+    remote_closed = 1<<2,
+    error =         1<<3
+};
+
+using FdEvents = uint32_t;
+
+class Dispatchable
+{
+public:
+    Dispatchable() = default;
+    virtual ~Dispatchable() = default;
+
+    Dispatchable& operator=(Dispatchable const&) = delete;
+    Dispatchable(Dispatchable const&) = delete;
+
+    /**
+     * \brief Get a poll()able file descriptor
+     * \return A file descriptor usable with poll() or equivalent function calls.
+     *         relevant_events() contains the set of event types to watch for.
+     */
+    virtual Fd watch_fd() const = 0;
+
+    /**
+     * \brief Dispatch one pending event
+     * \param [in] event    The set of events current on the file-descriptor
+     * \returns False iff no more events will be produced by this Dispatchable.
+     *          Dispatch should no longer be called.
+     * \note This will dispatch at most one event. If there are multiple events
+     *       specified in \ref event (eg: readable | remote_closed) then dispatch
+     *       will process only one.
+     * \note It is harmless to call dispatch() with an event that does not contain
+     *       any of the events from relevant_events(). The function will do
+     *       nothing in such a case.
+     * \note An implementation of dispatch() MUST handle FdEvent::error,
+     *       if only to return false and terminate further event dispatch.
+     */
+    virtual bool dispatch(FdEvents events) = 0;
+
+    /**
+     * \brief The set of file-descriptor events this Dispatchable handles.
+     */
+    virtual FdEvents relevant_events() const = 0;
+};
+}
+}
+
+#endif // MIR_DISPATCH_DISPATCHABLE_H_

=== added file 'include/common/mir/dispatch/simple_dispatch_thread.h'
--- include/common/mir/dispatch/simple_dispatch_thread.h	1970-01-01 00:00:00 +0000
+++ include/common/mir/dispatch/simple_dispatch_thread.h	2015-01-27 23:19:58 +0000
@@ -0,0 +1,48 @@
+/*
+ * 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 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: Christopher James Halse Rogers <[e-mail]>
+ */
+
+#ifndef MIR_DISPATCH_SIMPLE_DISPATCH_THREAD_H_
+#define MIR_DISPATCH_SIMPLE_DISPATCH_THREAD_H_
+
+#include <memory>
+#include <thread>
+#include "mir/fd.h"
+
+namespace mir
+{
+namespace dispatch
+{
+class Dispatchable;
+
+
+class SimpleDispatchThread
+{
+public:
+    SimpleDispatchThread(std::shared_ptr<Dispatchable> const& dispatchee);
+    ~SimpleDispatchThread() noexcept;
+
+private:
+    Fd shutdown_fd;
+    std::thread eventloop;
+};
+
+}
+}
+
+
+#endif // MIR_DISPATCH_SIMPLE_DISPATCH_THREAD_H_

=== modified file 'platform-ABI-sha1sums'
--- platform-ABI-sha1sums	2015-01-27 19:46:30 +0000
+++ platform-ABI-sha1sums	2015-01-27 23:19:58 +0000
@@ -1,4 +1,6 @@
 3329ada91412ded2f127aee9a92f065e57b81cb2  include/common/mir/cached_ptr.h
+d20a846dfa0e46fb276b4c181a6a5a56e7a884da  include/common/mir/dispatch/dispatchable.h
+7bdc0e58dc228ac655a618208bf133c6a499ca13  include/common/mir/dispatch/simple_dispatch_thread.h
 6473b6b17e72053a4ca3f7dcf8e2d81156fbfb85  include/common/mir/events/event_builders.h
 82ff9499ef62739379616e02164dc98f9914c329  include/common/mir/fd.h
 fe0275d9c64e7a2d99990382d04b1fe956d7b7e4  include/common/mir/frontend/surface_id.h

=== modified file 'server-ABI-sha1sums'
--- server-ABI-sha1sums	2015-01-27 19:46:30 +0000
+++ server-ABI-sha1sums	2015-01-27 23:19:58 +0000
@@ -1,4 +1,6 @@
 3329ada91412ded2f127aee9a92f065e57b81cb2  include/common/mir/cached_ptr.h
+d20a846dfa0e46fb276b4c181a6a5a56e7a884da  include/common/mir/dispatch/dispatchable.h
+7bdc0e58dc228ac655a618208bf133c6a499ca13  include/common/mir/dispatch/simple_dispatch_thread.h
 6473b6b17e72053a4ca3f7dcf8e2d81156fbfb85  include/common/mir/events/event_builders.h
 82ff9499ef62739379616e02164dc98f9914c329  include/common/mir/fd.h
 fe0275d9c64e7a2d99990382d04b1fe956d7b7e4  include/common/mir/frontend/surface_id.h

=== modified file 'src/client/mir_connection.cpp'
--- src/client/mir_connection.cpp	2015-01-26 23:09:56 +0000
+++ src/client/mir_connection.cpp	2015-01-27 23:19:58 +0000
@@ -26,6 +26,8 @@
 #include "mir/client_platform.h"
 #include "mir/client_platform_factory.h"
 #include "rpc/mir_basic_rpc_channel.h"
+#include "mir/dispatch/dispatchable.h"
+#include "mir/dispatch/simple_dispatch_thread.h"
 #include "connection_configuration.h"
 #include "display_configuration.h"
 #include "connection_surface_map.h"
@@ -41,6 +43,7 @@
 #include <boost/exception/diagnostic_information.hpp>
 
 namespace mcl = mir::client;
+namespace md = mir::dispatch;
 namespace mircv = mir::input::receiver;
 namespace gp = google::protobuf;
 
@@ -109,7 +112,8 @@
         display_configuration(conf.the_display_configuration()),
         lifecycle_control(conf.the_lifecycle_control()),
         surface_map(conf.the_surface_map()),
-        event_handler_register(conf.the_event_handler_register())
+        event_handler_register(conf.the_event_handler_register()),
+        eventloop{new md::SimpleDispatchThread{std::dynamic_pointer_cast<md::Dispatchable>(channel)}}
 {
     connect_result.set_error("connect not called");
     {

=== modified file 'src/client/mir_connection.h'
--- src/client/mir_connection.h	2015-01-20 22:23:20 +0000
+++ src/client/mir_connection.h	2015-01-27 23:19:58 +0000
@@ -69,6 +69,11 @@
 {
 class Logger;
 }
+
+namespace dispatch
+{
+class SimpleDispatchThread;
+}
 }
 
 struct MirConnection : mir::client::ClientContext
@@ -192,6 +197,8 @@
 
     std::shared_ptr<mir::client::EventHandlerRegister> const event_handler_register;
 
+    std::unique_ptr<mir::dispatch::SimpleDispatchThread> const eventloop;
+
     std::vector<int> extra_platform_data;
     
     std::shared_ptr<mir::client::ClientBufferStreamFactory> buffer_stream_factory;

=== modified file 'src/client/rpc/mir_protobuf_rpc_channel.cpp'
--- src/client/rpc/mir_protobuf_rpc_channel.cpp	2015-01-26 23:09:56 +0000
+++ src/client/rpc/mir_protobuf_rpc_channel.cpp	2015-01-27 23:19:58 +0000
@@ -40,6 +40,7 @@
 
 namespace mcl = mir::client;
 namespace mclr = mir::client::rpc;
+namespace md = mir::dispatch;
 
 namespace
 {
@@ -354,3 +355,18 @@
 {
     notify_disconnected();
 }
+
+mir::Fd mir::client::rpc::MirProtobufRpcChannel::watch_fd() const
+{
+    return transport->watch_fd();
+}
+
+bool mir::client::rpc::MirProtobufRpcChannel::dispatch(md::FdEvents events)
+{
+    return transport->dispatch(events);
+}
+
+md::FdEvents mclr::MirProtobufRpcChannel::relevant_events() const
+{
+    return transport->relevant_events();
+}

=== modified file 'src/client/rpc/mir_protobuf_rpc_channel.h'
--- src/client/rpc/mir_protobuf_rpc_channel.h	2015-01-14 06:39:13 +0000
+++ src/client/rpc/mir_protobuf_rpc_channel.h	2015-01-27 23:19:58 +0000
@@ -21,6 +21,7 @@
 
 #include "mir_basic_rpc_channel.h"
 #include "stream_transport.h"
+#include "mir/dispatch/dispatchable.h"
 
 #include <google/protobuf/service.h>
 #include <google/protobuf/descriptor.h>
@@ -52,7 +53,8 @@
 
 class MirProtobufRpcChannel :
         public MirBasicRpcChannel,
-        public StreamTransport::Observer
+        public StreamTransport::Observer,
+        public dispatch::Dispatchable
 {
 public:
     MirProtobufRpcChannel(std::unique_ptr<StreamTransport> transport,
@@ -64,8 +66,14 @@
 
     ~MirProtobufRpcChannel() = default;
 
+    // StreamTransport::Observer
     void on_data_available() override;
     void on_disconnected() override;
+
+    // Dispatchable
+    Fd watch_fd() const override;
+    bool dispatch(mir::dispatch::FdEvents events) override;
+    mir::dispatch::FdEvents relevant_events() const override;
 private:
     virtual void CallMethod(const google::protobuf::MethodDescriptor* method, google::protobuf::RpcController*,
         const google::protobuf::Message* parameters, google::protobuf::Message* response,

=== modified file 'src/client/rpc/stream_socket_transport.cpp'
--- src/client/rpc/stream_socket_transport.cpp	2015-01-14 06:39:13 +0000
+++ src/client/rpc/stream_socket_transport.cpp	2015-01-27 23:19:58 +0000
@@ -23,7 +23,6 @@
 
 #include <system_error>
 
-#include <signal.h>
 #include <errno.h>
 #include <sys/epoll.h>
 #include <sys/types.h>
@@ -34,11 +33,11 @@
 #include <boost/throw_exception.hpp>
 
 namespace mclr = mir::client::rpc;
+namespace md = mir::dispatch;
 
 mclr::StreamSocketTransport::StreamSocketTransport(mir::Fd const& fd)
     : socket_fd{fd}
 {
-    init();
 }
 
 mclr::StreamSocketTransport::StreamSocketTransport(std::string const& socket_path)
@@ -46,17 +45,6 @@
 {
 }
 
-mclr::StreamSocketTransport::~StreamSocketTransport()
-{
-    int dummy{0};
-    send(shutdown_fd, &dummy, sizeof(dummy), MSG_NOSIGNAL);
-    if (io_service_thread.joinable())
-    {
-        io_service_thread.join();
-    }
-    close(shutdown_fd);
-}
-
 void mclr::StreamSocketTransport::register_observer(std::shared_ptr<Observer> const& observer)
 {
     std::lock_guard<decltype(observer_mutex)> lock(observer_mutex);
@@ -176,110 +164,41 @@
         mir::send_fds(socket_fd, fds);
 }
 
-void mclr::StreamSocketTransport::init()
+mir::Fd mclr::StreamSocketTransport::watch_fd() const
 {
-    // We use sockets rather than a pipe so that we can control
-    // EPIPE behaviour; we don't want SIGPIPE when the IO loop terminates.
-    int socket_fds[2];
-    socketpair(AF_UNIX, SOCK_STREAM, 0, socket_fds);
-    this->shutdown_fd = mir::Fd{socket_fds[1]};
+    return socket_fd;
+}
 
-    auto shutdown_fd = mir::Fd{socket_fds[0]};
-    io_service_thread = std::thread([this, shutdown_fd]
+bool mclr::StreamSocketTransport::dispatch(md::FdEvents events)
+{
+    if (events & (md::FdEvent::remote_closed | md::FdEvent::error))
     {
-        // Our IO threads must not receive any signals
-        sigset_t all_signals;
-        sigfillset(&all_signals);
-
-        if (auto error = pthread_sigmask(SIG_BLOCK, &all_signals, NULL))
-            BOOST_THROW_EXCEPTION(
-                boost::enable_error_info(
-                    std::runtime_error("Failed to block signals on IO thread")) << boost::errinfo_errno(error));
-
-        mir::set_thread_name("Client IO loop");
-
-        int epoll_fd = epoll_create1(0);
-
-        epoll_event event;
-        // Make valgrind happy, harder
-        memset(&event, 0, sizeof(event));
-
-        event.events = EPOLLIN | EPOLLRDHUP;
-        event.data.fd = socket_fd;
-        epoll_ctl(epoll_fd, EPOLL_CTL_ADD, socket_fd, &event);
-
-        event.events = EPOLLIN | EPOLLRDHUP;
-        event.data.fd = shutdown_fd;
-        epoll_ctl(epoll_fd, EPOLL_CTL_ADD, shutdown_fd, &event);
-
-        bool shutdown_requested{false};
-        while (!shutdown_requested)
+        if (events & md::FdEvent::readable)
         {
-            epoll_event event;
-            epoll_wait(epoll_fd, &event, 1, -1);
-            if (event.data.fd == socket_fd)
-            {
-                if (event.events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR))
-                {
-                    if (event.events & EPOLLIN)
-                    {
-                        // If the remote end shut down cleanly it's possible there's some more
-                        // data left to read, or that reads will now return 0 (EOF)
-                        //
-                        // If there's more data left to read, notify of this before disconnect.
-                        int dummy;
-                        if (recv(socket_fd, &dummy, sizeof(dummy), MSG_PEEK | MSG_NOSIGNAL) > 0)
-                        {
-                            try
-                            {
-                                notify_data_available();
-                            }
-                            catch(...)
-                            {
-                                //It's quite likely that notify_data_available() will lead to
-                                //an exception being thrown; after all, the remote has closed
-                                //the connection.
-                                //
-                                //This doesn't matter; we're already shutting down.
-                            }
-                        }
-                    }
-                    notify_disconnected();
-                    shutdown_requested = true;
-                }
-                else if (event.events & EPOLLIN)
-                {
-                    try
-                    {
-                        notify_data_available();
-                    }
-                    catch (socket_disconnected_error &err)
-                    {
-                        // We've already notified of disconnection.
-                        shutdown_requested = true;
-                    }
-                    // These need not be fatal.
-                    catch (fd_reception_error &err)
-                    {
-                    }
-                    catch (socket_error &err)
-                    {
-                    }
-                    catch (...)
-                    {
-                        // We've no idea what the problem is, so clean up as best we can.
-                        notify_disconnected();
-                        shutdown_requested = true;
-                    }
-                }
-            }
-            if (event.data.fd == shutdown_fd)
-            {
-                shutdown_requested = true;
+            // If the remote end shut down cleanly it's possible there's some more
+            // data left to read, or that reads will now return 0 (EOF)
+            //
+            // If there's more data left to read, notify of this before disconnect.
+            int dummy;
+            if (recv(socket_fd, &dummy, sizeof(dummy), MSG_PEEK | MSG_NOSIGNAL) > 0)
+            {
+                notify_data_available();
+                return true;
             }
         }
-        ::close(epoll_fd);
-    });
+        notify_disconnected();
+        return false;
+    }
+    else if (events & md::FdEvent::readable)
+    {
+        notify_data_available();
+    }
+    return true;
+}
+
+md::FdEvents mclr::StreamSocketTransport::relevant_events() const
+{
+    return md::FdEvent::readable | md::FdEvent::remote_closed;
 }
 
 mir::Fd mclr::StreamSocketTransport::open_socket(std::string const& path)

=== modified file 'src/client/rpc/stream_socket_transport.h'
--- src/client/rpc/stream_socket_transport.h	2015-01-14 06:39:13 +0000
+++ src/client/rpc/stream_socket_transport.h	2015-01-27 23:19:58 +0000
@@ -38,22 +38,21 @@
 public:
     StreamSocketTransport(Fd const& fd);
     StreamSocketTransport(std::string const& socket_path);
-    ~StreamSocketTransport() override;
 
     void register_observer(std::shared_ptr<Observer> const& observer) override;
     void receive_data(void* buffer, size_t bytes_requested) override;
     void receive_data(void* buffer, size_t bytes_requested, std::vector<Fd>& fds) override;
     void send_message(std::vector<uint8_t> const& buffer, std::vector<mir::Fd> const& fds) override;
 
+    Fd watch_fd() const override;
+    bool dispatch(mir::dispatch::FdEvents event) override;
+    mir::dispatch::FdEvents relevant_events() const override;
 private:
-    void init();
     Fd open_socket(std::string const& path);
     void notify_data_available();
     void notify_disconnected();
 
-    std::thread io_service_thread;
     Fd const socket_fd;
-    Fd shutdown_fd;
 
     std::mutex observer_mutex;
     std::vector<std::shared_ptr<Observer>> observers;

=== modified file 'src/client/rpc/stream_transport.h'
--- src/client/rpc/stream_transport.h	2015-01-14 06:39:13 +0000
+++ src/client/rpc/stream_transport.h	2015-01-27 23:19:58 +0000
@@ -25,6 +25,7 @@
 #include <stdint.h>
 
 #include "mir/fd.h"
+#include "mir/dispatch/dispatchable.h"
 
 namespace mir
 {
@@ -64,7 +65,7 @@
  *       from different threads. Multiple threads calling the same
  *       function need synchronisation.
  */
-class StreamTransport
+class StreamTransport : public dispatch::Dispatchable
 {
 public:
     /**
@@ -80,8 +81,8 @@
 
     /**
      * \brief Observer of IO status
-     * \note The Transport may call Observer members from arbitrary threads.
-     *       The Observer implementation is responsible for any synchronisation.
+     * \note The Transport will only call Observers in response to dispatch(),
+     *       and on the thread calling dispatch().
      */
     class Observer
     {
@@ -106,8 +107,6 @@
     /**
      * \brief Register an IO observer
      * \param [in] observer
-     * \note There is no guarantee which thread will call into the observer.
-     *       Synchronisation is the responsibility of the caller.
      */
     virtual void register_observer(std::shared_ptr<Observer> const& observer) = 0;
 

=== modified file 'src/common/CMakeLists.txt'
--- src/common/CMakeLists.txt	2015-01-22 23:12:46 +0000
+++ src/common/CMakeLists.txt	2015-01-27 23:19:58 +0000
@@ -17,6 +17,8 @@
 add_subdirectory(sharedlibrary)
 add_subdirectory(thread)
 add_subdirectory(time)
+add_subdirectory(dispatch)
+
 list(APPEND MIR_COMMON_SOURCES
   $<TARGET_OBJECTS:mirtime>
   ${CMAKE_CURRENT_SOURCE_DIR}/log.cpp

=== added directory 'src/common/dispatch'
=== added file 'src/common/dispatch/CMakeLists.txt'
--- src/common/dispatch/CMakeLists.txt	1970-01-01 00:00:00 +0000
+++ src/common/dispatch/CMakeLists.txt	2015-01-27 23:19:58 +0000
@@ -0,0 +1,22 @@
+# 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 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: Christopher James Halse Rogers <[e-mail]>
+
+list(
+  APPEND MIR_COMMON_SOURCES
+  ${CMAKE_CURRENT_SOURCE_DIR}/simple_dispatch_thread.cpp
+)
+
+set(MIR_COMMON_SOURCES ${MIR_COMMON_SOURCES} PARENT_SCOPE)

=== added file 'src/common/dispatch/simple_dispatch_thread.cpp'
--- src/common/dispatch/simple_dispatch_thread.cpp	1970-01-01 00:00:00 +0000
+++ src/common/dispatch/simple_dispatch_thread.cpp	2015-01-27 23:19:58 +0000
@@ -0,0 +1,158 @@
+/*
+ * 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 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: Christopher James Halse Rogers <[e-mail]>
+ */
+
+#include "mir/dispatch/simple_dispatch_thread.h"
+#include "mir/dispatch/dispatchable.h"
+
+#include <sys/epoll.h>
+#include <unistd.h>
+#include <system_error>
+#include <signal.h>
+#include <boost/exception/all.hpp>
+
+namespace md = mir::dispatch;
+
+namespace
+{
+md::FdEvents epoll_to_fd_event(epoll_event const& event)
+{
+    md::FdEvents val{0};
+    if (event.events & EPOLLIN)
+    {
+        val |= md::FdEvent::readable;
+    }
+    if (event.events & EPOLLOUT)
+    {
+        val = md::FdEvent::writable;
+    }
+    if (event.events & (EPOLLHUP | EPOLLRDHUP))
+    {
+        val |= md::FdEvent::remote_closed;
+    }
+    if (event.events & EPOLLERR)
+    {
+        val = md::FdEvent::error;
+    }
+    return val;
+}
+
+int fd_event_to_epoll(md::FdEvents const& event)
+{
+    int epoll_value{0};
+    if (event & md::FdEvent::readable)
+    {
+        epoll_value |= EPOLLIN;
+    }
+    if (event & md::FdEvent::writable)
+    {
+        epoll_value |= EPOLLOUT;
+    }
+    if (event & md::FdEvent::remote_closed)
+    {
+        epoll_value |= EPOLLRDHUP | EPOLLHUP;
+    }
+    if (event & md::FdEvent::error)
+    {
+        epoll_value |= EPOLLERR;
+    }
+    return epoll_value;
+}
+
+void wait_for_events_forever(std::shared_ptr<md::Dispatchable> const& dispatchee, mir::Fd shutdown_fd)
+{
+    auto epoll_fd = mir::Fd{epoll_create1(0)};
+    if (epoll_fd == mir::Fd::invalid)
+    {
+        BOOST_THROW_EXCEPTION((std::system_error{errno,
+                                                 std::system_category(),
+                                                 "Failed to create epoll IO monitor"}));
+    }
+    epoll_event event;
+    memset(&event, 0, sizeof(event));
+
+    enum fd_names : uint32_t {
+        shutdown,
+        dispatchee_fd
+    };
+
+    // We only care when the shutdown pipe has been closed
+    event.data.u32 = fd_names::shutdown;
+    event.events = EPOLLRDHUP;
+    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, shutdown_fd, &event);
+
+    // Ask the dispatchee what it events it's interested in...
+    event.data.u32 = fd_names::dispatchee_fd;
+    event.events = fd_event_to_epoll(dispatchee->relevant_events());
+    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, dispatchee->watch_fd(), &event);
+
+    for (;;)
+    {
+        epoll_wait(epoll_fd, &event, 1, -1);
+        if (event.data.u32 == fd_names::dispatchee_fd)
+        {
+            if (!dispatchee->dispatch(epoll_to_fd_event(event)))
+            {
+                // No need to keep looping, the Dispatchable's not going to produce any more events.
+                return;
+            }
+        }
+        else if (event.data.u32 == fd_names::shutdown)
+        {
+            // The only thing we do with the shutdown fd is to close it.
+            return;
+        }
+    }
+}
+
+}
+
+md::SimpleDispatchThread::SimpleDispatchThread(std::shared_ptr<md::Dispatchable> const& dispatchee)
+{
+    int pipefds[2];
+    if (pipe(pipefds) < 0)
+    {
+        BOOST_THROW_EXCEPTION((std::system_error{errno,
+                                                 std::system_category(),
+                                                 "Failed to create shutdown pipe for IO thread"}));
+    }
+    shutdown_fd = mir::Fd{pipefds[1]};
+    mir::Fd const terminate_fd = mir::Fd{pipefds[0]};
+    eventloop = std::thread{[dispatchee, terminate_fd]()
+                            {
+                                // Our IO threads must not receive any signals
+                                sigset_t all_signals;
+                                sigfillset(&all_signals);
+
+                                if (auto error = pthread_sigmask(SIG_BLOCK, &all_signals, NULL))
+                                    BOOST_THROW_EXCEPTION((
+                                                std::system_error{error,
+                                                                  std::system_category(),
+                                                                  "Failed to block signals on IO thread"}));
+
+                                wait_for_events_forever(dispatchee, terminate_fd);
+                            }};
+}
+
+md::SimpleDispatchThread::~SimpleDispatchThread() noexcept
+{
+    shutdown_fd = mir::Fd{};
+    if (eventloop.joinable())
+    {
+        eventloop.join();
+    }
+}

=== modified file 'src/common/symbols.map'
--- src/common/symbols.map	2015-01-26 19:12:43 +0000
+++ src/common/symbols.map	2015-01-27 23:19:58 +0000
@@ -216,6 +216,10 @@
 MIR_COMMON_3.2 {
   global:
     extern "C++" {
-        mir::events::*
+      mir::dispatch::SimpleDispatchThread::SimpleDispatchThread*;
+      mir::dispatch::SimpleDispatchThread::?SimpleDispatchThread*;
+      typeinfo?for?mir::dispatch::SimpleDispatchThread;
+      vtable?for?mir::dispatch::SimpleDispatchThread;
+      mir::events::*
     };
-} MIR_COMMON_3;
+} MIR_COMMON_3.1;

=== modified file 'tests/CMakeLists.txt'
--- tests/CMakeLists.txt	2015-01-22 13:01:46 +0000
+++ tests/CMakeLists.txt	2015-01-27 23:19:58 +0000
@@ -44,6 +44,7 @@
   ${PROJECT_SOURCE_DIR}/include/platform
   ${PROJECT_SOURCE_DIR}/include/server
   ${PROJECT_SOURCE_DIR}/include/client
+  ${PROJECT_SOURCE_DIR}/include/common
   ${PROJECT_SOURCE_DIR}/src/platforms/mesa/include
 )
 

=== added file 'tests/include/mir_test/fd_utils.h'
--- tests/include/mir_test/fd_utils.h	1970-01-01 00:00:00 +0000
+++ tests/include/mir_test/fd_utils.h	2015-01-27 23:19:58 +0000
@@ -0,0 +1,69 @@
+/*
+ * Copyright © 2015 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 <[e-mail]>
+ */
+
+#ifndef MIR_TEST_FD_UTILS_H_
+#define MIR_TEST_FD_UTILS_H_
+
+#include "mir/fd.h"
+#include <chrono>
+
+#include <poll.h>
+
+#include <gtest/gtest.h>
+
+namespace mir
+{
+namespace test
+{
+::testing::AssertionResult std_call_succeeded(int retval);
+
+::testing::AssertionResult fd_is_readable(mir::Fd const& fd);
+
+template<typename Period, typename Rep>
+::testing::AssertionResult fd_becomes_readable(mir::Fd const& fd,
+                                               std::chrono::duration<Period, Rep> timeout)
+{
+    int timeout_ms = std::chrono::duration_cast<std::chrono::milliseconds>(timeout).count();
+
+    pollfd readable;
+    readable.events = POLLIN;
+    readable.fd = fd;
+
+    auto result = std_call_succeeded(poll(&readable, 1, timeout_ms));
+    if (result)
+    {
+        if (readable.revents & POLLERR)
+        {
+            return ::testing::AssertionFailure() << "error condition on fd";
+        }
+        if (readable.revents & POLLNVAL)
+        {
+            return ::testing::AssertionFailure() << "fd is invalid";
+        }
+        if (!(readable.revents & POLLIN))
+        {
+            return ::testing::AssertionFailure() << "fd is not readable";
+        }
+        return ::testing::AssertionSuccess();
+    }
+    return result;
+}
+}
+}
+
+#endif // MIR_TEST_FD_UTILS_H_

=== modified file 'tests/include/mir_test/pipe.h'
--- tests/include/mir_test/pipe.h	2013-07-03 09:54:10 +0000
+++ tests/include/mir_test/pipe.h	2015-01-27 23:19:58 +0000
@@ -19,6 +19,8 @@
 #ifndef MIR_TEST_PIPE_H_
 #define MIR_TEST_PIPE_H_
 
+#include "mir/fd.h"
+
 namespace mir
 {
 namespace test
@@ -28,16 +30,18 @@
 {
 public:
     Pipe();
-    ~Pipe();
+    Pipe(int flags);
+    ~Pipe() = default;
 
-    int read_fd() const;
-    int write_fd() const;
+    Fd read_fd() const;
+    Fd write_fd() const;
 
 private:
     Pipe(Pipe const&) = delete;
     Pipe& operator=(Pipe const&) = delete;
 
-    int pipefd[2];
+    Fd reader;
+    Fd writer;
 };
 
 }

=== added file 'tests/include/mir_test/test_dispatchable.h'
--- tests/include/mir_test/test_dispatchable.h	1970-01-01 00:00:00 +0000
+++ tests/include/mir_test/test_dispatchable.h	2015-01-27 23:19:58 +0000
@@ -0,0 +1,60 @@
+/*
+ * Copyright © 2015 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 <[e-mail]>
+ */
+
+
+#ifndef MIR_TEST_TEST_DISPATCHABLE_H_
+#define MIR_TEST_TEST_DISPATCHABLE_H_
+
+#include "mir_test/pipe.h"
+#include "mir/dispatch/dispatchable.h"
+
+#include <functional>
+
+namespace mir
+{
+namespace test
+{
+
+class TestDispatchable : public dispatch::Dispatchable
+{
+public:
+    TestDispatchable(std::function<bool(dispatch::FdEvents)> const& target,
+                     dispatch::FdEvents relevant_events);
+    TestDispatchable(std::function<bool(dispatch::FdEvents)> const& target);
+    TestDispatchable(std::function<void()> const& target);
+
+    mir::Fd watch_fd() const override;
+    bool dispatch(dispatch::FdEvents events) override;
+    dispatch::FdEvents relevant_events() const override;
+
+    void trigger();
+    void untrigger();
+    void hangup();
+
+private:
+    mir::Fd read_fd;
+    mir::Fd write_fd;
+    std::function<bool(dispatch::FdEvents)> const target;
+    dispatch::FdEvents const eventmask;
+};
+
+}
+}
+
+
+#endif // MIR_TEST_TEST_DISPATCHABLE_H_

=== modified file 'tests/include/mir_test/test_protobuf_client.h'
--- tests/include/mir_test/test_protobuf_client.h	2015-01-14 06:39:13 +0000
+++ tests/include/mir_test/test_protobuf_client.h	2015-01-27 23:19:58 +0000
@@ -30,6 +30,10 @@
 
 namespace mir
 {
+namespace dispatch
+{
+class SimpleDispatchThread;
+}
 namespace test
 {
 namespace doubles
@@ -42,6 +46,7 @@
 
     std::shared_ptr<doubles::MockRpcReport> rpc_report;
     std::shared_ptr<google::protobuf::RpcChannel> channel;
+    std::shared_ptr<dispatch::SimpleDispatchThread> eventloop;
     mir::protobuf::DisplayServer::Stub display_server;
     mir::protobuf::ConnectParameters connect_parameters;
     mir::protobuf::SurfaceParameters surface_parameters;

=== modified file 'tests/integration-tests/client/test_screencast.cpp'
--- tests/integration-tests/client/test_screencast.cpp	2015-01-20 22:23:20 +0000
+++ tests/integration-tests/client/test_screencast.cpp	2015-01-27 23:19:58 +0000
@@ -18,6 +18,8 @@
 
 #include "mir_protobuf.pb.h"
 #include "src/client/default_connection_configuration.h"
+#include "mir/dispatch/simple_dispatch_thread.h"
+#include "mir/dispatch/dispatchable.h"
 
 #include "mir/frontend/connector.h"
 #include "mir_test/test_protobuf_server.h"
@@ -31,6 +33,7 @@
 
 namespace mcl = mir::client;
 namespace mt = mir::test;
+namespace md = mir::dispatch;
 
 namespace
 {
@@ -74,6 +77,9 @@
             mcl::DefaultConnectionConfiguration{test_socket}.the_rpc_channel();
         protobuf_server =
             std::make_shared<mir::protobuf::DisplayServer::Stub>(rpc_channel.get());
+        eventloop =
+            std::make_shared<md::SimpleDispatchThread>(
+                std::dynamic_pointer_cast<md::Dispatchable>(rpc_channel));
     }
 
     char const* const test_socket = "./test_socket_screencast";
@@ -81,6 +87,7 @@
     std::shared_ptr<mt::TestProtobufServer> test_server;
     std::shared_ptr<google::protobuf::RpcChannel> rpc_channel;
     std::shared_ptr<mir::protobuf::DisplayServer> protobuf_server;
+    std::shared_ptr<mir::dispatch::SimpleDispatchThread> eventloop;
 };
 
 }

=== modified file 'tests/mir_test/CMakeLists.txt'
--- tests/mir_test/CMakeLists.txt	2015-01-14 06:39:13 +0000
+++ tests/mir_test/CMakeLists.txt	2015-01-27 23:19:58 +0000
@@ -11,6 +11,8 @@
   wait_object.cpp
   current_thread_name.cpp
   validity_matchers.cpp
+  fd_utils.cpp
+  test_dispatchable.cpp
 )
 
 target_link_libraries(mir-test mirprotobuf)

=== added file 'tests/mir_test/fd_utils.cpp'
--- tests/mir_test/fd_utils.cpp	1970-01-01 00:00:00 +0000
+++ tests/mir_test/fd_utils.cpp	2015-01-27 23:19:58 +0000
@@ -0,0 +1,40 @@
+/*
+ * Copyright © 2015 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 <[e-mail]>
+ */
+
+#include "mir_test/fd_utils.h"
+
+::testing::AssertionResult mir::test::std_call_succeeded(int retval)
+{
+    if (retval >= 0)
+    {
+        return ::testing::AssertionSuccess();
+    }
+    else
+    {
+        return ::testing::AssertionFailure() << "errno: "
+                                             << errno
+                                             << " ["
+                                             << strerror(errno)
+                                             << "]";
+    }
+}
+
+::testing::AssertionResult mir::test::fd_is_readable(mir::Fd const& fd)
+{
+    return fd_becomes_readable(fd, std::chrono::seconds{0});
+}

=== modified file 'tests/mir_test/pipe.cpp'
--- tests/mir_test/pipe.cpp	2015-01-14 06:39:13 +0000
+++ tests/mir_test/pipe.cpp	2015-01-27 23:19:58 +0000
@@ -24,32 +24,35 @@
 #include <system_error>
 
 #include <unistd.h>
+#include <fcntl.h>
 
 namespace mt = mir::test;
 
 mt::Pipe::Pipe()
-{
-    if (pipe(pipefd))
+    : Pipe(0)
+{
+}
+
+mt::Pipe::Pipe(int flags)
+{
+    int pipefd[2];
+    if (pipe2(pipefd, flags))
     {
         BOOST_THROW_EXCEPTION(
             boost::enable_error_info(std::system_error(errno,
                                                        std::system_category(),
                                                        "Failed to create pipe")));
     }
-}
-
-mt::Pipe::~Pipe()
-{
-    close(pipefd[0]);
-    close(pipefd[1]);
-}
-
-int mt::Pipe::read_fd() const
-{
-    return pipefd[0];
-}
-
-int mt::Pipe::write_fd() const
-{
-    return pipefd[1];
+    reader = mir::Fd{pipefd[0]};
+    writer = mir::Fd{pipefd[1]};
+}
+
+mir::Fd mt::Pipe::read_fd() const
+{
+    return reader;
+}
+
+mir::Fd mt::Pipe::write_fd() const
+{
+    return writer;
 }

=== added file 'tests/mir_test/test_dispatchable.cpp'
--- tests/mir_test/test_dispatchable.cpp	1970-01-01 00:00:00 +0000
+++ tests/mir_test/test_dispatchable.cpp	2015-01-27 23:19:58 +0000
@@ -0,0 +1,98 @@
+/*
+ * Copyright © 2015 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 <[e-mail]>
+ */
+
+#include <fcntl.h>
+
+#include "mir_test/pipe.h"
+#include "mir_test/test_dispatchable.h"
+
+#include <gtest/gtest.h>
+#include <gmock/gmock.h>
+
+namespace mt = mir::test;
+namespace md = mir::dispatch;
+
+mt::TestDispatchable::TestDispatchable(std::function<bool(md::FdEvents)> const& target,
+                                       md::FdEvents relevant_events)
+    : target{target},
+      eventmask{relevant_events}
+{
+    // Need to use O_NONBLOCK here to ensure concurrent dispatch doesn't block indefinitely
+    // in read()
+    mt::Pipe pipe{O_NONBLOCK};
+    read_fd = pipe.read_fd();
+    write_fd = pipe.write_fd();
+}
+
+mt::TestDispatchable::TestDispatchable(std::function<bool(md::FdEvents)> const& target)
+    : TestDispatchable(target, md::FdEvent::readable)
+{
+}
+
+mt::TestDispatchable::TestDispatchable(std::function<void()> const& target)
+    : TestDispatchable([target](md::FdEvents) { target(); return true; })
+{
+}
+
+mir::Fd mt::TestDispatchable::watch_fd() const
+{
+    return read_fd;
+}
+
+bool mt::TestDispatchable::dispatch(md::FdEvents events)
+{
+    auto continue_dispatch = target(events);
+    if (!(events & md::FdEvent::remote_closed))
+    {
+        // There's no way to untrigger remote hangup :)
+        untrigger();
+    }
+    return continue_dispatch;
+}
+
+md::FdEvents mt::TestDispatchable::relevant_events() const
+{
+    return eventmask;
+}
+
+void mt::TestDispatchable::trigger()
+{
+    using namespace testing;
+    char dummy{0};
+    EXPECT_THAT(::write(write_fd, &dummy, sizeof(dummy)), Eq(sizeof(dummy)));
+}
+
+void mt::TestDispatchable::untrigger()
+{
+    using namespace testing;
+    char dummy{0};
+    auto val = ::read(read_fd, &dummy, sizeof(dummy));
+    if (val < 0)
+    {
+        EXPECT_THAT(errno, Eq(EAGAIN));
+    }
+    else
+    {
+        EXPECT_THAT(val, Eq(sizeof(dummy)));
+    }
+}
+
+void mt::TestDispatchable::hangup()
+{
+    write_fd = mir::Fd{};
+}

=== modified file 'tests/mir_test_doubles/test_protobuf_client.cpp'
--- tests/mir_test_doubles/test_protobuf_client.cpp	2015-01-21 00:49:28 +0000
+++ tests/mir_test_doubles/test_protobuf_client.cpp	2015-01-27 23:19:58 +0000
@@ -27,22 +27,27 @@
 #include "src/client/lifecycle_control.h"
 #include "src/client/rpc/make_rpc_channel.h"
 #include "src/client/rpc/mir_basic_rpc_channel.h"
+#include "mir/dispatch/dispatchable.h"
+#include "mir/dispatch/simple_dispatch_thread.h"
 
 #include <thread>
 
 namespace mtd = mir::test::doubles;
+namespace mclr = mir::client::rpc;
+namespace md = mir::dispatch;
 
 mir::test::TestProtobufClient::TestProtobufClient(
     std::string socket_file,
     int timeout_ms) :
     rpc_report(std::make_shared<testing::NiceMock<doubles::MockRpcReport>>()),
-    channel(mir::client::rpc::make_rpc_channel(
+    channel(mclr::make_rpc_channel(
         socket_file,
         std::make_shared<mir::client::ConnectionSurfaceMap>(),
         std::make_shared<mir::client::DisplayConfiguration>(),
         rpc_report,
         std::make_shared<mir::client::LifecycleControl>(),
         std::make_shared<mtd::NullClientEventSink>())),
+    eventloop{std::make_shared<md::SimpleDispatchThread>(std::dynamic_pointer_cast<md::Dispatchable>(channel))},
     display_server(channel.get(), ::google::protobuf::Service::STUB_DOESNT_OWN_CHANNEL),
     maxwait(timeout_ms),
     connect_done_called(false),

=== modified file 'tests/unit-tests/CMakeLists.txt'
--- tests/unit-tests/CMakeLists.txt	2015-01-27 04:48:26 +0000
+++ tests/unit-tests/CMakeLists.txt	2015-01-27 23:19:58 +0000
@@ -63,6 +63,7 @@
 add_subdirectory(scene/)
 add_subdirectory(examples/)
 add_subdirectory(thread/)
+add_subdirectory(dispatch/)
 
 link_directories(${LIBRARY_OUTPUT_PATH})
 

=== modified file 'tests/unit-tests/client/test_mir_connection.cpp'
--- tests/unit-tests/client/test_mir_connection.cpp	2015-01-26 23:09:56 +0000
+++ tests/unit-tests/client/test_mir_connection.cpp	2015-01-27 23:19:58 +0000
@@ -27,6 +27,7 @@
 #include "src/client/display_configuration.h"
 #include "src/client/mir_surface.h"
 #include "mir/client_buffer_factory.h"
+#include "mir/dispatch/dispatchable.h"
 
 #include "src/server/frontend/resource_cache.h" /* needed by test_server.h */
 #include "mir_test/test_protobuf_server.h"
@@ -44,12 +45,19 @@
 namespace mp = mir::protobuf;
 namespace geom = mir::geometry;
 namespace mtd = mir::test::doubles;
+namespace md = mir::dispatch;
 
 namespace
 {
 
-struct MockRpcChannel : public mir::client::rpc::MirBasicRpcChannel
+struct MockRpcChannel : public mir::client::rpc::MirBasicRpcChannel,
+                        public mir::dispatch::Dispatchable
 {
+    MockRpcChannel()
+    {
+        ON_CALL(*this, watch_fd()).WillByDefault(testing::Return(mir::Fd{}));
+    }
+
     void CallMethod(const google::protobuf::MethodDescriptor* method,
                     google::protobuf::RpcController*,
                     const google::protobuf::Message* parameters,
@@ -77,6 +85,10 @@
     MOCK_METHOD1(drm_auth_magic, void(const mp::DRMMagic*));
     MOCK_METHOD2(connect, void(mp::ConnectParameters const*,mp::Connection*));
     MOCK_METHOD1(configure_display_sent, void(mp::DisplayConfiguration const*));
+
+    MOCK_CONST_METHOD0(watch_fd, mir::Fd());
+    MOCK_METHOD1(dispatch, bool(md::FdEvents));
+    MOCK_CONST_METHOD0(relevant_events, md::FdEvents());
 };
 
 struct MockClientPlatform : public mcl::ClientPlatform

=== modified file 'tests/unit-tests/client/test_protobuf_rpc_channel.cpp'
--- tests/unit-tests/client/test_protobuf_rpc_channel.cpp	2015-01-14 06:39:13 +0000
+++ tests/unit-tests/client/test_protobuf_rpc_channel.cpp	2015-01-27 23:19:58 +0000
@@ -41,6 +41,7 @@
 namespace mcl = mir::client;
 namespace mclr = mir::client::rpc;
 namespace mtd = mir::test::doubles;
+namespace md = mir::dispatch;
 
 namespace
 {
@@ -108,6 +109,9 @@
     MOCK_METHOD2(receive_data, void(void*, size_t));
     MOCK_METHOD3(receive_data, void(void*, size_t, std::vector<mir::Fd>&));
     MOCK_METHOD2(send_message, void(std::vector<uint8_t> const&, std::vector<mir::Fd> const&));
+    MOCK_CONST_METHOD0(watch_fd, mir::Fd());
+    MOCK_METHOD1(dispatch, bool(md::FdEvents));
+    MOCK_CONST_METHOD0(relevant_events, md::FdEvents());
 
     // Transport interface
     void register_observer_default(std::shared_ptr<Observer> const& observer)

=== modified file 'tests/unit-tests/client/test_stream_transport.cpp'
--- tests/unit-tests/client/test_stream_transport.cpp	2015-01-14 06:39:13 +0000
+++ tests/unit-tests/client/test_stream_transport.cpp	2015-01-27 23:19:58 +0000
@@ -22,6 +22,7 @@
 
 #include "mir_test/auto_unblock_thread.h"
 #include "mir_test/signal.h"
+#include "mir_test/fd_utils.h"
 #include "mir/raii.h"
 
 #include <sys/socket.h>
@@ -43,6 +44,8 @@
 #include <gmock/gmock.h>
 
 namespace mclr = mir::client::rpc;
+namespace mt = mir::test;
+namespace md = mir::dispatch;
 
 namespace
 {
@@ -71,12 +74,6 @@
         transport = std::make_shared<TransportMechanism>(transport_fd);
     }
 
-    virtual ~StreamTransportTest()
-    {
-        // We don't care about errors, so unconditionally close the test fd.
-        close(test_fd);
-    }
-
     mir::Fd transport_fd;
     mir::Fd test_fd;
     std::shared_ptr<TransportMechanism> transport;
@@ -85,51 +82,199 @@
 typedef ::testing::Types<mclr::StreamSocketTransport> Transports;
 TYPED_TEST_CASE(StreamTransportTest, Transports);
 
+TYPED_TEST(StreamTransportTest, ReturnsValidWatchFd)
+{
+    // A valid fd is >= 0, and we know that stdin, stdout, and stderr aren't correct.
+    EXPECT_GE(this->transport->watch_fd(), 3);
+}
+
+TYPED_TEST(StreamTransportTest, WatchFdIsPollable)
+{
+    pollfd socket_readable;
+    socket_readable.events = POLLIN;
+    socket_readable.fd = this->transport->watch_fd();
+
+    ASSERT_TRUE(mt::std_call_succeeded(poll(&socket_readable, 1, 0)));
+
+    EXPECT_FALSE(socket_readable.revents & POLLERR);
+    EXPECT_FALSE(socket_readable.revents & POLLNVAL);
+}
+
+TYPED_TEST(StreamTransportTest, WatchFdNotifiesReadableWhenDataPending)
+{
+    uint64_t dummy{0xdeadbeef};
+    EXPECT_EQ(sizeof(dummy), write(this->test_fd, &dummy, sizeof(dummy)));
+
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+}
+
+TYPED_TEST(StreamTransportTest, WatchFdRemainsUnreadableUntilEventPending)
+{
+    EXPECT_FALSE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+
+    uint64_t dummy{0xdeadbeef};
+    EXPECT_EQ(sizeof(dummy), write(this->test_fd, &dummy, sizeof(dummy)));
+
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+}
+
+TYPED_TEST(StreamTransportTest, WatchFdIsNoLongerReadableAfterEventProcessing)
+{
+    using namespace testing;
+
+    uint64_t dummy{0xdeadbeef};
+
+    auto observer = std::make_shared<NiceMock<MockObserver>>();
+
+    ON_CALL(*observer, on_data_available())
+        .WillByDefault(Invoke([dummy, this]()
+                              {
+                                  decltype(dummy) buffer;
+                                  this->transport->receive_data(&buffer, sizeof(dummy));
+                              }));
+
+    this->transport->register_observer(observer);
+
+    EXPECT_EQ(sizeof(dummy), write(this->test_fd, &dummy, sizeof(dummy)));
+
+    ASSERT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+
+    this->transport->dispatch(md::FdEvent::readable);
+
+    EXPECT_FALSE(mt::fd_is_readable(this->test_fd));
+}
+
+TYPED_TEST(StreamTransportTest, NoEventsDispatchedUntilDispatchCalled)
+{
+    using namespace testing;
+
+    auto observer = std::make_shared<NiceMock<MockObserver>>();
+    bool data_available{false};
+    bool disconnected{false};
+
+    uint64_t dummy{0xdeadbeef};
+    EXPECT_EQ(sizeof(dummy), write(this->test_fd, &dummy, sizeof(dummy)));
+    ::close(this->test_fd);
+
+    ON_CALL(*observer, on_data_available()).WillByDefault(Invoke([this, dummy, &data_available]()
+                                                                 {
+                                                                     decltype(dummy) buffer;
+                                                                     this->transport->receive_data(&buffer, sizeof(buffer));
+                                                                     data_available = true;
+                                                                 }));
+    ON_CALL(*observer, on_disconnected()).WillByDefault(Invoke([&disconnected]()
+                                                               { disconnected = true; }));
+
+    this->transport->register_observer(observer);
+
+    std::this_thread::sleep_for(std::chrono::seconds{1});
+    EXPECT_FALSE(data_available);
+    EXPECT_FALSE(disconnected);
+
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+    while (mt::fd_is_readable(this->transport->watch_fd()) &&
+           this->transport->dispatch(md::FdEvent::readable | md::FdEvent::remote_closed))
+        ;
+
+    EXPECT_TRUE(data_available);
+    EXPECT_TRUE(disconnected);
+}
+
+TYPED_TEST(StreamTransportTest, DispatchesSingleEventAtATime)
+{
+    using namespace testing;
+
+    auto observer = std::make_shared<NiceMock<MockObserver>>();
+    bool data_available{false};
+    bool disconnected{false};
+
+    uint64_t dummy{0xdeadbeef};
+    EXPECT_EQ(sizeof(dummy), write(this->test_fd, &dummy, sizeof(dummy)));
+    ::close(this->test_fd);
+
+    ON_CALL(*observer, on_data_available()).WillByDefault(Invoke([this, dummy, &data_available]()
+                                                                 {
+                                                                     decltype(dummy) buffer;
+                                                                     this->transport->receive_data(&buffer, sizeof(buffer));
+                                                                     data_available = true;
+                                                                 }));
+    ON_CALL(*observer, on_disconnected()).WillByDefault(Invoke([&disconnected]()
+                                                               { disconnected = true; }));
+
+    this->transport->register_observer(observer);
+
+    EXPECT_FALSE(data_available);
+    EXPECT_FALSE(disconnected);
+
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+
+    this->transport->dispatch(md::FdEvent::readable | md::FdEvent::remote_closed);
+
+    EXPECT_TRUE(data_available xor disconnected);
+
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+
+    this->transport->dispatch(md::FdEvent::readable | md::FdEvent::remote_closed);
+
+    EXPECT_TRUE(data_available);
+    EXPECT_TRUE(disconnected);
+}
+
 TYPED_TEST(StreamTransportTest, NoticesRemoteDisconnect)
 {
     using namespace testing;
     auto observer = std::make_shared<NiceMock<MockObserver>>();
-    auto done = std::make_shared<mir::test::Signal>();
+    bool disconnected{false};
 
-    ON_CALL(*observer, on_disconnected()).WillByDefault(Invoke([done]()
-                                                               { done->raise(); }));
+    ON_CALL(*observer, on_disconnected()).WillByDefault(Invoke([&disconnected]()
+                                                               { disconnected = true; }));
 
     this->transport->register_observer(observer);
 
     close(this->test_fd);
 
-    EXPECT_TRUE(done->wait_for(std::chrono::seconds{1}));
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+    while (mt::fd_is_readable(this->transport->watch_fd()) &&
+           this->transport->dispatch(md::FdEvent::remote_closed))
+        ;
+
+    EXPECT_TRUE(disconnected);
 }
 
-TYPED_TEST(StreamTransportTest, NoticesRemoteDisconnectWhileReadingInIOLoop)
+TYPED_TEST(StreamTransportTest, NoticesRemoteDisconnectWhileReading)
 {
     using namespace testing;
     auto observer = std::make_shared<NiceMock<MockObserver>>();
-    auto done = std::make_shared<mir::test::Signal>();
-    bool data_notified{false};
-    bool finished_read{false};
-
-    ON_CALL(*observer, on_disconnected()).WillByDefault(Invoke([done]()
-                                                               { done->raise(); }));
-    ON_CALL(*observer, on_data_available())
-        .WillByDefault(Invoke([this, &data_notified, &finished_read]()
-                              {
-                                  data_notified = true;
-                                  char buffer[8];
-                                  this->transport->receive_data(buffer, sizeof(buffer));
-                                  finished_read = true;
-                              }));
-
+    bool disconnected{false};
+    bool receive_error_detected{false};
+
+    ON_CALL(*observer, on_disconnected()).WillByDefault(Invoke([&disconnected]()
+                                                               { disconnected = true; }));
     this->transport->register_observer(observer);
 
-    uint32_t dummy{0xdeadbeef};
-    EXPECT_EQ(sizeof(dummy), write(this->test_fd, &dummy, sizeof(dummy)));
-
-    close(this->test_fd);
-
-    EXPECT_TRUE(done->wait_for(std::chrono::seconds{1}));
-    EXPECT_TRUE(data_notified);
-    EXPECT_FALSE(finished_read);
+    mir::test::AutoJoinThread closer{[this]()
+                                     {
+                                         std::this_thread::sleep_for(std::chrono::seconds{1});
+                                         ::close(this->test_fd);
+                                     }};
+
+    try
+    {
+        char buffer[8];
+        this->transport->receive_data(buffer, sizeof(buffer));
+    }
+    catch (std::runtime_error)
+    {
+        receive_error_detected = true;
+    }
+
+    // There should now be a disconnect event pending...
+    EXPECT_TRUE(mt::fd_is_readable(this->transport->watch_fd()));
+
+    this->transport->dispatch(md::FdEvent::remote_closed);
+
+    EXPECT_TRUE(disconnected);
+    EXPECT_TRUE(receive_error_detected);
 }
 
 TYPED_TEST(StreamTransportTest, NotifiesOnDataAvailable)
@@ -137,39 +282,21 @@
     using namespace testing;
 
     auto observer = std::make_shared<NiceMock<MockObserver>>();
-    auto done = std::make_shared<mir::test::Signal>();
-
-    ON_CALL(*observer, on_data_available()).WillByDefault(Invoke([done]()
-                                                                 { done->raise(); }));
-
-    this->transport->register_observer(observer);
-
-    uint64_t dummy{0xdeadbeef};
-    EXPECT_EQ(sizeof(dummy), write(this->test_fd, &dummy, sizeof(dummy)));
-
-    EXPECT_TRUE(done->wait_for(std::chrono::seconds{1}));
-}
-
-TYPED_TEST(StreamTransportTest, DoesntNotifyUntilDataAvailable)
-{
-    using namespace testing;
-
-    auto observer = std::make_shared<NiceMock<MockObserver>>();
-    auto done = std::make_shared<mir::test::Signal>();
-
-    ON_CALL(*observer, on_data_available()).WillByDefault(Invoke([done]()
-                                                                 { done->raise(); }));
-
-    this->transport->register_observer(observer);
-
-    std::this_thread::sleep_for(std::chrono::seconds{1});
-
-    EXPECT_FALSE(done->raised());
-
-    uint64_t dummy{0xdeadbeef};
-    EXPECT_EQ(sizeof(dummy), write(this->test_fd, &dummy, sizeof(dummy)));
-
-    EXPECT_TRUE(done->wait_for(std::chrono::seconds{1}));
+    bool notified_data_available{false};
+
+    ON_CALL(*observer, on_data_available()).WillByDefault(Invoke([&notified_data_available]()
+                                                                 { notified_data_available = true; }));
+
+    this->transport->register_observer(observer);
+
+    uint64_t dummy{0xdeadbeef};
+    EXPECT_EQ(sizeof(dummy), write(this->test_fd, &dummy, sizeof(dummy)));
+
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+
+    this->transport->dispatch(md::FdEvent::readable);
+
+    EXPECT_TRUE(notified_data_available);
 }
 
 TYPED_TEST(StreamTransportTest, KeepsNotifyingOfAvailableDataUntilAllIsRead)
@@ -177,98 +304,101 @@
     using namespace testing;
 
     auto observer = std::make_shared<NiceMock<MockObserver>>();
-    auto done = std::make_shared<mir::test::Signal>();
 
     std::array<uint8_t, sizeof(int) * 256> data;
     data.fill(0);
-    std::atomic<size_t> bytes_left{data.size()};
+    size_t bytes_left{data.size()};
 
     ON_CALL(*observer, on_data_available())
-        .WillByDefault(Invoke([done, &bytes_left, this]()
+        .WillByDefault(Invoke([&bytes_left, this]()
                               {
                                   int dummy;
                                   this->transport->receive_data(&dummy, sizeof(dummy));
-                                  bytes_left.fetch_sub(sizeof(dummy));
-                                  if (bytes_left.load() == 0)
-                                  {
-                                      done->raise();
-                                  }
+                                  bytes_left -= sizeof(dummy);
                               }));
 
     this->transport->register_observer(observer);
 
     EXPECT_EQ(data.size(), write(this->test_fd, data.data(), data.size()));
 
-    EXPECT_TRUE(done->wait_for(std::chrono::seconds{5}));
-    EXPECT_EQ(0, bytes_left.load());
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+    while (mt::fd_is_readable(this->transport->watch_fd()) &&
+           this->transport->dispatch(md::FdEvent::readable))
+        ;
+
+    EXPECT_EQ(0, bytes_left);
 }
 
 TYPED_TEST(StreamTransportTest, StopsNotifyingOnceAllDataIsRead)
 {
     using namespace testing;
-    int const buffer_size{256};
 
     auto observer = std::make_shared<NiceMock<MockObserver>>();
-    auto done = std::make_shared<mir::test::Signal>();
+
+    std::array<uint8_t, sizeof(int) * 256> data;
+    data.fill(0);
+    size_t bytes_left{data.size()};
 
     ON_CALL(*observer, on_data_available())
-        .WillByDefault(Invoke([this, done]()
+        .WillByDefault(Invoke([&bytes_left, this]()
                               {
-                                  if (done->raised())
-                                  {
-                                      FAIL() << "on_data_available called without new data available";
-                                  }
-                                  uint8_t dummy_buffer[buffer_size];
-                                  this->transport->receive_data(dummy_buffer, sizeof(dummy_buffer));
-                                  done->raise();
+                                  int dummy;
+                                  this->transport->receive_data(&dummy, sizeof(dummy));
+                                  bytes_left -= sizeof(dummy);
                               }));
+
     this->transport->register_observer(observer);
 
-    EXPECT_FALSE(done->raised());
-    uint8_t dummy_buffer[buffer_size];
-    memset(dummy_buffer, 0xab, sizeof(dummy_buffer));
-    EXPECT_EQ(sizeof(dummy_buffer), write(this->test_fd, dummy_buffer, sizeof(dummy_buffer)));
-
-    EXPECT_TRUE(done->wait_for(std::chrono::seconds{1}));
-
-    std::this_thread::sleep_for(std::chrono::seconds{1});
+    EXPECT_EQ(data.size(), write(this->test_fd, data.data(), data.size()));
+
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+    while (bytes_left > 0)
+    {
+        this->transport->dispatch(md::FdEvent::readable);
+    }
+
+    EXPECT_FALSE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
 }
 
 TYPED_TEST(StreamTransportTest, DoesntSendDataAvailableNotificationOnDisconnect)
 {
     using namespace testing;
-    int const buffer_size{256};
 
     auto observer = std::make_shared<NiceMock<MockObserver>>();
-    auto read_done = std::make_shared<mir::test::Signal>();
-    auto disconnect_done = std::make_shared<mir::test::Signal>();
-    std::atomic<int> notify_count{0};
-
+    int notify_count{0};
+    bool disconnected{false};
+
+    uint64_t dummy{0xdeedfaac};
+    EXPECT_EQ(sizeof(dummy), write(this->test_fd, &dummy, sizeof(dummy)));
+
+    ON_CALL(*observer, on_disconnected()).WillByDefault(Invoke([&disconnected]()
+                                                               { disconnected = true; }));
     ON_CALL(*observer, on_data_available())
-        .WillByDefault(Invoke([this, read_done, &notify_count]()
+        .WillByDefault(Invoke([dummy, &notify_count, this]()
                               {
                                   notify_count++;
-                                  uint8_t dummy_buffer[buffer_size];
-                                  this->transport->receive_data(dummy_buffer, sizeof(dummy_buffer));
-                                  read_done->raise();
+
+                                  decltype(dummy) buffer;
+                                  this->transport->receive_data(&buffer, sizeof(buffer));
                               }));
-    ON_CALL(*observer, on_disconnected()).WillByDefault(Invoke([this, disconnect_done]()
-                                                               { disconnect_done->raise(); }));
 
     this->transport->register_observer(observer);
 
-    EXPECT_FALSE(read_done->raised());
-    uint8_t dummy_buffer[buffer_size];
-    memset(dummy_buffer, 0xab, sizeof(dummy_buffer));
-    EXPECT_EQ(sizeof(dummy_buffer), write(this->test_fd, dummy_buffer, sizeof(dummy_buffer)));
-
-    EXPECT_TRUE(read_done->wait_for(std::chrono::seconds{1}));
-    EXPECT_EQ(1, notify_count);
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+    while (mt::fd_is_readable(this->transport->watch_fd()))
+    {
+        this->transport->dispatch(md::FdEvent::readable);
+    }
 
     ::close(this->test_fd);
-    EXPECT_TRUE(disconnect_done->wait_for(std::chrono::seconds{1}));
+
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+    while (mt::fd_is_readable(this->transport->watch_fd()) &&
+           this->transport->dispatch(md::FdEvent::remote_closed | md::FdEvent::readable))
+        ;
 
     EXPECT_EQ(1, notify_count);
+    EXPECT_TRUE(disconnected);
 }
 
 TYPED_TEST(StreamTransportTest, ReadsCorrectData)
@@ -276,23 +406,25 @@
     using namespace testing;
 
     auto observer = std::make_shared<NiceMock<MockObserver>>();
-    auto done = std::make_shared<mir::test::Signal>();
 
     std::string expected{"I am the very model of a modern major general"};
     std::vector<char> received(expected.size());
 
     ON_CALL(*observer, on_data_available())
-        .WillByDefault(Invoke([done, &received, this]()
+        .WillByDefault(Invoke([&received, this]()
                               {
                                   this->transport->receive_data(received.data(), received.size());
-                                  done->raise();
                               }));
 
     this->transport->register_observer(observer);
 
     EXPECT_EQ(expected.size(), write(this->test_fd, expected.data(), expected.size()));
 
-    ASSERT_TRUE(done->wait_for(std::chrono::seconds{1}));
+    EXPECT_TRUE(mt::fd_becomes_readable(this->transport->watch_fd(), std::chrono::seconds{1}));
+    while (mt::fd_is_readable(this->transport->watch_fd()) &&
+           this->transport->dispatch(md::FdEvent::readable))
+           ;
+
     EXPECT_EQ(0, memcmp(expected.data(), received.data(), expected.size()));
 }
 

=== added directory 'tests/unit-tests/dispatch'
=== added file 'tests/unit-tests/dispatch/CMakeLists.txt'
--- tests/unit-tests/dispatch/CMakeLists.txt	1970-01-01 00:00:00 +0000
+++ tests/unit-tests/dispatch/CMakeLists.txt	2015-01-27 23:19:58 +0000
@@ -0,0 +1,5 @@
+list(APPEND UNIT_TEST_SOURCES
+  ${CMAKE_CURRENT_SOURCE_DIR}/test_simple_dispatch_thread.cpp
+)
+
+set(UNIT_TEST_SOURCES ${UNIT_TEST_SOURCES} PARENT_SCOPE)

=== added file 'tests/unit-tests/dispatch/test_simple_dispatch_thread.cpp'
--- tests/unit-tests/dispatch/test_simple_dispatch_thread.cpp	1970-01-01 00:00:00 +0000
+++ tests/unit-tests/dispatch/test_simple_dispatch_thread.cpp	2015-01-27 23:19:58 +0000
@@ -0,0 +1,185 @@
+/*
+ * Copyright © 2015 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 <[e-mail]>
+ */
+
+#include "mir/dispatch/simple_dispatch_thread.h"
+#include "mir/dispatch/dispatchable.h"
+#include "mir/fd.h"
+#include "mir_test/pipe.h"
+#include "mir_test/signal.h"
+#include "mir_test/test_dispatchable.h"
+
+#include <fcntl.h>
+
+#include <atomic>
+
+#include <gtest/gtest.h>
+#include <gmock/gmock.h>
+
+namespace md = mir::dispatch;
+namespace mt = mir::test;
+
+namespace
+{
+class SimpleDispatchThreadTest : public ::testing::Test
+{
+public:
+    SimpleDispatchThreadTest()
+    {
+        mt::Pipe pipe{O_NONBLOCK};
+        watch_fd = pipe.read_fd();
+        test_fd = pipe.write_fd();
+    }
+
+    mir::Fd watch_fd;
+    mir::Fd test_fd;
+};
+
+class MockDispatchable : public md::Dispatchable
+{
+public:
+    MOCK_CONST_METHOD0(watch_fd, mir::Fd());
+    MOCK_METHOD1(dispatch, bool(md::FdEvents));
+    MOCK_CONST_METHOD0(relevant_events, md::FdEvents());
+};
+
+}
+
+TEST_F(SimpleDispatchThreadTest, calls_dispatch_when_fd_is_readable)
+{
+    using namespace testing;
+
+    auto dispatched = std::make_shared<mt::Signal>();
+    auto dispatchable = std::make_shared<mt::TestDispatchable>([dispatched]() { dispatched->raise(); });
+
+    md::SimpleDispatchThread dispatcher{dispatchable};
+
+    dispatchable->trigger();
+
+    EXPECT_TRUE(dispatched->wait_for(std::chrono::seconds{1}));
+}
+
+TEST_F(SimpleDispatchThreadTest, stops_calling_dispatch_once_fd_is_not_readable)
+{
+    using namespace testing;
+
+    std::atomic<int> dispatch_count{0};
+    auto dispatchable = std::make_shared<mt::TestDispatchable>([&dispatch_count]() { ++dispatch_count; });
+
+    md::SimpleDispatchThread dispatcher{dispatchable};
+
+    dispatchable->trigger();
+
+    std::this_thread::sleep_for(std::chrono::seconds{1});
+
+    EXPECT_THAT(dispatch_count, Eq(1));
+}
+
+TEST_F(SimpleDispatchThreadTest, passes_dispatch_events_through)
+{
+    using namespace testing;
+
+    auto dispatched_with_only_readable = std::make_shared<mt::Signal>();
+    auto dispatched_with_hangup = std::make_shared<mt::Signal>();
+    auto delegate = [dispatched_with_only_readable, dispatched_with_hangup](md::FdEvents events)
+    {
+        if (events == md::FdEvent::readable)
+        {
+            dispatched_with_only_readable->raise();
+        }
+        if (events & md::FdEvent::remote_closed)
+        {
+            dispatched_with_hangup->raise();
+            return false;
+        }
+        return true;
+    };
+    auto dispatchable = std::make_shared<mt::TestDispatchable>(delegate, md::FdEvent::readable | md::FdEvent::remote_closed);
+
+    md::SimpleDispatchThread dispatcher{dispatchable};
+
+    dispatchable->trigger();
+    EXPECT_TRUE(dispatched_with_only_readable->wait_for(std::chrono::seconds{1}));
+
+    dispatchable->hangup();
+    EXPECT_TRUE(dispatched_with_hangup->wait_for(std::chrono::seconds{1}));
+}
+
+TEST_F(SimpleDispatchThreadTest, doesnt_call_dispatch_after_first_false_return)
+{
+    using namespace testing;
+
+    int constexpr expected_count{10};
+    auto dispatched_more_than_enough = std::make_shared<mt::Signal>();
+
+    auto delegate = [dispatched_more_than_enough](md::FdEvents)
+    {
+        static std::atomic<int> dispatch_count{0};
+
+        if (++dispatch_count == expected_count)
+        {
+            return false;
+        }
+        if (dispatch_count > expected_count)
+        {
+            dispatched_more_than_enough->raise();
+        }
+        return true;
+    };
+    auto dispatchable = std::make_shared<mt::TestDispatchable>(delegate);
+
+    md::SimpleDispatchThread dispatcher{dispatchable};
+
+    for (int i = 0; i < expected_count + 1; ++i)
+    {
+        dispatchable->trigger();
+    }
+
+    EXPECT_FALSE(dispatched_more_than_enough->wait_for(std::chrono::seconds{1}));
+}
+
+TEST_F(SimpleDispatchThreadTest, only_calls_dispatch_with_remote_closed_when_relevant)
+{
+    using namespace testing;
+
+    auto dispatchable = std::make_shared<NiceMock<MockDispatchable>>();
+    ON_CALL(*dispatchable, watch_fd()).WillByDefault(Return(test_fd));
+    ON_CALL(*dispatchable, relevant_events()).WillByDefault(Return(md::FdEvent::writable));
+    auto dispatched_writable = std::make_shared<mt::Signal>();
+    auto dispatched_closed = std::make_shared<mt::Signal>();
+
+    ON_CALL(*dispatchable, dispatch(_)).WillByDefault(Invoke([=](md::FdEvents events)
+    {
+        if (events & md::FdEvent::writable)
+        {
+            dispatched_writable->raise();
+        }
+        if (events & md::FdEvent::remote_closed)
+        {
+            dispatched_closed->raise();
+        }
+        return true;
+    }));
+
+    md::SimpleDispatchThread dispatcher{dispatchable};
+
+    EXPECT_TRUE(dispatched_writable->wait_for(std::chrono::seconds{1}));
+
+    // Make the fd remote-closed...
+    watch_fd = mir::Fd{};
+    EXPECT_FALSE(dispatched_closed->wait_for(std::chrono::seconds{1}));
+}