Merge lp:~mir-team/mir/eventloop-integration into lp:mir

[Chris Halse Rogers]

Add a client-driven MirConnection dispatch option.

Lots of clients aren't appreciative of our devil-may-care approach to calling their code on threads of our choosing. For them, we offer mir_connection_dispatch().

This will simplify XMir, make a Mir backend for Plymouth feasible, and greatly please Ryan.

Fixes: https://bugs.launchpad.net/mir/+bug/1397375

Add a client-driven MirConnection dispatch option.

Lots of clients aren't appreciative of our devil-may-care approach to calling their code on threads of our choosing. For them, we offer mir_connection_dispatch().

This will simplify XMir, make a Mir backend for Plymouth feasible, and greatly please Ryan.

Fixes: https://bugs.launchpad.net/mir/+bug/1397375

[Chris Halse Rogers (raof)]

Marked as WIP; I'll break this up to make the review easier.

[Chris Halse Rogers (raof)]

Looks like I'll be breaking this up a bit more, but now actually works!

Bounce off CI to fully test that it fully really fully does fully work.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:2387
http://jenkins.qa.ubuntu.com/job/mir-ci/3834/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-android-vivid-i386-build/2478
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-clang-vivid-amd64-build/2477
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-vivid-touch/2426
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/1831
        deb: http://jenkins.qa.ubuntu.com/job/mir-vivid-amd64-ci/1831/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/2426
        deb: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-builder-vivid-armhf/2426/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/mir-mediumtests-runner-mako/5302
    SUCCESS: http://s-jenkins.ubuntu-ci:8080/job/touch-flash-device/20488

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

review: Approve (continuous-integration)

Unmerged revisions

2387. By Chris Halse Rogers on 2015-05-14

Add valgrind suppressions for GDBus bollocks.

2386. By Chris Halse Rogers on 2015-05-14

Free the surface_spec in ClientLibrary test.

Now doesn't leak!

2385. By Chris Halse Rogers on 2015-05-14

Don't call a synchronous method from a manually-dispatched async callback.

This fails for... some reason. Don't know at the moment.

2384. By Chris Halse Rogers on 2015-05-14

Initialise ProtobufRpcChannel::prioritise_next_request.

Thanks, valgrind.

2383. By Chris Halse Rogers on 2015-05-14

Minor improvements to synchronous helper

2382. By Chris Halse Rogers on 2015-05-14

Use make_synchronous_call helper for all synchronous RPC calls

2381. By Chris Halse Rogers on 2015-05-14

Add helper infrastructure for making synchronous RPC calls

2380. By Chris Halse Rogers on 2015-05-14

Enable getting a MirConnection out of a MirScreencast

2379. By Chris Halse Rogers on 2015-05-14

We treat MirBufferStream as a ClientBufferStream internally - make sure the pointers we return *are* ClientBufferStreams

2378. By Chris Halse Rogers on 2015-05-13

Add a way to get the MirConnection associated with a MirSurface

=== modified file 'benchmarks/benchmark_multiplexing_dispatchable.cpp'
--- benchmarks/benchmark_multiplexing_dispatchable.cpp	2015-04-24 15:36:12 +0000
+++ benchmarks/benchmark_multiplexing_dispatchable.cpp	2015-05-14 07:50:26 +0000
@@ -17,9 +17,9 @@
  */
 
 #include "mir/dispatch/multiplexing_dispatchable.h"
-#include "mir/dispatch/simple_dispatch_thread.h"
 
 #include <iostream>
+#include <thread>
 #include <vector>
 #include <memory>
 #include <chrono>
 
=== modified file 'debian/control'
--- debian/control	2015-05-08 15:40:20 +0000
+++ debian/control	2015-05-14 07:50:26 +0000
@@ -245,6 +245,7 @@
 Pre-Depends: ${misc:Pre-Depends}
 Depends: ${misc:Depends},
          ${shlibs:Depends},
+         umockdev,
 Recommends: mir-demos,
 Description: Display Server for Ubuntu - stress tests and other test tools
  Mir is a display server running on linux systems, with a focus on efficiency,
 
=== modified file 'debian/mir-test-tools.install'
--- debian/mir-test-tools.install	2015-04-09 06:20:31 +0000
+++ debian/mir-test-tools.install	2015-05-14 07:50:26 +0000
@@ -8,3 +8,4 @@
 usr/lib/*/mir/server-platform/graphics-dummy.so
 usr/lib/*/mir/server-platform/input-stub.so
 usr/lib/*/mir/client-platform/dummy.so
+usr/share/mir/udev_recordings
 
=== modified file 'include/client/mir_toolkit/mir_connection.h'
--- include/client/mir_toolkit/mir_connection.h	2015-03-31 02:35:42 +0000
+++ include/client/mir_toolkit/mir_connection.h	2015-05-14 07:50:26 +0000
@@ -63,6 +63,35 @@
 MirConnection *mir_connect_sync(char const *server, char const *app_name);
 
 /**
+ * Request a connection to the Mir server.
+ *
+ * The client is responsible for handling event consumption via
+ * mir_connection_dispatch(). Events will be available when the fd returned
+ * by mir_connection_get_event_fd() becomes readable.
+ *
+ * All callbacks for this MirConnection and any objects created on it will be
+ * called from the thread calling mir_connection_dispatch().
+ *
+ * \param [in] server       A string specifying the server to connect to.
+ *                          Connection strings can either be path to the socket file on
+ *                          the filesystem, or an open file descriptor fd://
+ * \param [in] app_name     A name referring to the application
+ * \param [in] callback     Callback function to be invoked when request
+ *                          completes. mir_connection_is_valid() will return false
+ *                          until this callback has completed.
+ * \param [in,out] context  User data passed to the callback function
+ * \returns                 The resulting MirConnection
+ * \todo     Currently manual dispatch interacts awkwardly with *_sync() calls;
+ *           Synchronous calls do *not* automatically dispatch, so unless client
+ *           code has a separate thread running dispatch, any *_sync() call will
+ *           deadlock. This restriction will be fixed in a later release.
+ */
+MirConnection* mir_connect_with_manual_dispatch(char const* server,
+                                                char const* app_name,
+                                                mir_connected_callback callback,
+                                                void* context);
+
+/**
  * Test for a valid connection
  * \param [in] connection  The connection
  * \return                 True if the supplied connection is valid, or
@@ -87,6 +116,28 @@
 void mir_connection_release(MirConnection *connection);
 
 /**
+ * \brief Get the notification fd for this connection
+ * \param [in] connection   The connection. This connection must have been created by
+ *                          mir_connect_with_manual_dispatch().
+ * \return                  A file descriptor that supports select/poll/epoll and similar
+ *                          APIs that becomes readable when there are events to dispatch,
+ *                          or -1 if \ref connection was not created by
+ *                          mir_connect_with_manual_dispatch();
+ * \note    This fd is owned by the MirConnection. Do not close the fd yourself.
+ *          The fd remains valid until mir_connection_release() returns.
+ */
+int mir_connection_get_event_fd(MirConnection* connection);
+
+/**
+ * \brief Dispatch a single pending event on the connection
+ * \param [in] connection   The connection. This connection must have been created by
+ *                          mir_connect_with_manual_dispatch().
+ * \note    It is an error to call this on a \ref connection not created by
+ *          mir_connect_with_manual_dispatch().
+ */
+void mir_connection_dispatch(MirConnection* connection);
+
+/**
  * Query platform-specific data and/or file descriptors that are required to
  * initialize GL/EGL features.
  *   \param [in]  connection        The connection
 
=== modified file 'include/client/mir_toolkit/mir_wait.h'
--- include/client/mir_toolkit/mir_wait.h	2014-03-31 14:36:08 +0000
+++ include/client/mir_toolkit/mir_wait.h	2015-05-14 07:50:26 +0000
@@ -46,6 +46,7 @@
  */
 void mir_wait_for_one(MirWaitHandle *wait_handle);
 
+bool mir_wait_handle_ready(MirWaitHandle* wait_handle);
 
 #ifdef __cplusplus
 }
 
=== removed file 'include/common/mir/dispatch/simple_dispatch_thread.h'
--- include/common/mir/dispatch/simple_dispatch_thread.h	2015-04-23 18:29:35 +0000
+++ include/common/mir/dispatch/simple_dispatch_thread.h	1970-01-01 00:00:00 +0000
@@ -1,50 +0,0 @@
-/*
- * 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(std::shared_ptr<Dispatchable> const& dispatchee,
-                         std::function<void()> const& exception_handler);
-    ~SimpleDispatchThread() noexcept;
-
-private:
-    Fd shutdown_fd;
-    std::thread eventloop;
-};
-
-}
-}
-
-
-#endif // MIR_DISPATCH_SIMPLE_DISPATCH_THREAD_H_
 
=== added file 'include/common/mir/dispatch/threaded_dispatcher.h'
--- include/common/mir/dispatch/threaded_dispatcher.h	1970-01-01 00:00:00 +0000
+++ include/common/mir/dispatch/threaded_dispatcher.h	2015-05-14 07:50:26 +0000
@@ -0,0 +1,75 @@
+/*
+ * 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 <string>
+#include <memory>
+#include <thread>
+#include <vector>
+#include <mutex>
+#include <condition_variable>
+
+#include "mir/dispatch/multiplexing_dispatchable.h"
+#include "mir/fd.h"
+
+namespace mir
+{
+namespace dispatch
+{
+class Dispatchable;
+
+class ThreadedDispatcher
+{
+public:
+    ThreadedDispatcher(std::string const& name, std::shared_ptr<Dispatchable> const& dispatchee);
+    ThreadedDispatcher(std::string const& name,
+                       std::shared_ptr<Dispatchable> const& dispatchee,
+                       std::function<void()> const& exception_handler);
+    ~ThreadedDispatcher() noexcept;
+
+    void add_thread();
+    void remove_thread();
+
+private:
+    class ThreadShutdownRequestHandler;
+    friend class ThreadShutdownRequestHandler;
+
+    std::string const name_base;
+
+    std::shared_ptr<ThreadShutdownRequestHandler> thread_exiter;
+    std::shared_ptr<MultiplexingDispatchable> dispatcher;
+
+    std::mutex thread_pool_mutex;
+    std::vector<std::thread> threadpool;
+
+    std::function<void()> const exception_handler;
+
+    static void dispatch_loop(std::string const& name,
+                              std::shared_ptr<ThreadShutdownRequestHandler> thread_register,
+                              std::shared_ptr<Dispatchable> dispatcher,
+                              std::function<void()> const& exception_handler);
+
+};
+
+}
+}
+
+
+#endif // MIR_DISPATCH_SIMPLE_DISPATCH_THREAD_H_
 
=== modified file 'src/client/CMakeLists.txt'
--- src/client/CMakeLists.txt	2015-05-08 09:01:55 +0000
+++ src/client/CMakeLists.txt	2015-05-14 07:50:26 +0000
@@ -70,6 +70,7 @@
   buffer_stream.cpp
   mir_buffer_stream_api.cpp
   default_client_buffer_stream_factory.cpp
+  synchronous_helper.cpp
   ${MIR_CLIENT_SOURCES}
 )
 
 
=== modified file 'src/client/buffer_stream.cpp'
--- src/client/buffer_stream.cpp	2015-04-23 14:09:57 +0000
+++ src/client/buffer_stream.cpp	2015-05-14 07:50:26 +0000
@@ -87,13 +87,16 @@
 // TODO: It seems like a bit of a wart that we have to pass the Logger specifically here...perhaps
 // due to the lack of an easily mockable client configuration interface (passing around
 // connection can complicate unit tests ala MirSurface and test_client_mir_surface.cpp)
-mcl::BufferStream::BufferStream(mp::DisplayServer& server,
+mcl::BufferStream::BufferStream(
+    MirConnection* allocating_connection,
+    mp::DisplayServer& server,
     mcl::BufferStreamMode mode,
     std::shared_ptr<mcl::ClientPlatform> const& client_platform,
     mp::BufferStream const& protobuf_bs,
     std::shared_ptr<mcl::PerfReport> const& perf_report,
     std::string const& surface_name)
-    : display_server(server),
+    : connection{allocating_connection},
+      display_server(server),
       mode(mode),
       client_platform(client_platform),
       protobuf_bs(protobuf_bs),
@@ -106,13 +109,16 @@
     perf_report->name_surface(surface_name.c_str());
 }
 
-mcl::BufferStream::BufferStream(mp::DisplayServer& server,
+mcl::BufferStream::BufferStream(
+    MirConnection* allocating_connection,
+    mp::DisplayServer& server,
     std::shared_ptr<mcl::ClientPlatform> const& client_platform,
     mp::BufferStreamParameters const& parameters,
     std::shared_ptr<mcl::PerfReport> const& perf_report,
     mir_buffer_stream_callback callback,
     void *context)
-    : display_server(server),
+    : connection{allocating_connection},
+      display_server(server),
       mode(BufferStreamMode::Producer),
       client_platform(client_platform),
       buffer_depository{client_platform->create_buffer_factory(), mir::frontend::client_buffer_cache_size},
@@ -146,7 +152,11 @@
     egl_native_window_ = client_platform->create_egl_native_window(this);
 
     if (callback)
-        callback(reinterpret_cast<MirBufferStream*>(this), context);
+    {
+        // Fun fact! The offset of mcl::ClientBufferStream's vtable in *this is *not* 0!
+        auto cbs_object = static_cast<mcl::ClientBufferStream*>(this);
+        callback(reinterpret_cast<MirBufferStream*>(cbs_object), context);
+    }
     create_wait_handle.result_received();
 }
 
@@ -382,7 +392,11 @@
     mir_buffer_stream_callback callback, void* context)
 {
     if (callback)
-        callback(reinterpret_cast<MirBufferStream*>(this), context);
+    {
+        // Fun fact! The offset of mcl::ClientBufferStream's vtable in *this is *not* 0!
+        auto cbs_object = static_cast<mcl::ClientBufferStream*>(this);
+        callback(reinterpret_cast<MirBufferStream*>(cbs_object), context);
+    }
     release_wait_handle.result_received();
 }
 
@@ -403,3 +417,8 @@
 {
     buffer_depository.set_max_buffers(cache_size);
 }
+
+MirConnection* mcl::BufferStream::get_connection() const
+{
+    return connection;
+}
 
=== modified file 'src/client/buffer_stream.h'
--- src/client/buffer_stream.h	2015-04-23 14:09:57 +0000
+++ src/client/buffer_stream.h	2015-05-14 07:50:26 +0000
@@ -57,14 +57,18 @@
 class BufferStream : public EGLNativeSurface, public ClientBufferStream
 {
 public:
-    BufferStream(mir::protobuf::DisplayServer& server,
+    BufferStream(
+        MirConnection* allocating_connection,
+        mir::protobuf::DisplayServer& server,
         BufferStreamMode mode,
         std::shared_ptr<ClientPlatform> const& native_window_factory,
         protobuf::BufferStream const& protobuf_bs,
         std::shared_ptr<PerfReport> const& perf_report,
         std::string const& surface_name);
     // For surfaceless buffer streams
-    BufferStream(mir::protobuf::DisplayServer& server,
+    BufferStream(
+        MirConnection* allocating_connection,
+        mir::protobuf::DisplayServer& server,
         std::shared_ptr<ClientPlatform> const& native_window_factory,
         mir::protobuf::BufferStreamParameters const& parameters,
         std::shared_ptr<PerfReport> const& perf_report,
@@ -100,6 +104,8 @@
 
     frontend::BufferStreamId rpc_id() const override;
     bool valid() const override;
+
+    MirConnection* get_connection() const override;
     
 protected:
     BufferStream(BufferStream const&) = delete;
@@ -116,6 +122,8 @@
 
     mutable std::mutex mutex; // Protects all members of *this
 
+    MirConnection* const connection;
+
     mir::protobuf::DisplayServer& display_server;
 
     BufferStreamMode const mode;
 
=== modified file 'src/client/client_buffer_stream.h'
--- src/client/client_buffer_stream.h	2015-04-01 20:00:19 +0000
+++ src/client/client_buffer_stream.h	2015-05-14 07:50:26 +0000
@@ -63,6 +63,8 @@
     virtual MirWaitHandle* release(mir_buffer_stream_callback callback, void* context) = 0;
 
     virtual bool valid() const = 0;
+
+    virtual MirConnection* get_connection() const = 0;
     
 protected:
     ClientBufferStream() = default;
 
=== modified file 'src/client/client_buffer_stream_factory.h'
--- src/client/client_buffer_stream_factory.h	2015-04-01 20:00:19 +0000
+++ src/client/client_buffer_stream_factory.h	2015-05-14 07:50:26 +0000
@@ -25,6 +25,8 @@
 
 #include <memory>
 
+class MirConnection;
+
 namespace mir
 {
 namespace client
@@ -34,13 +36,21 @@
 class ClientBufferStreamFactory
 {
 public:
-    virtual std::shared_ptr<ClientBufferStream> make_consumer_stream(protobuf::DisplayServer& server,
-       protobuf::BufferStream const& protobuf_bs, std::string const& surface_name) = 0;
-    virtual std::shared_ptr<ClientBufferStream> make_producer_stream(protobuf::DisplayServer& server,
-       protobuf::BufferStream const& protobuf_bs, std::string const& surface_name) = 0;
+    virtual std::shared_ptr<ClientBufferStream> make_consumer_stream(
+        MirConnection* allocating_connection,
+        protobuf::DisplayServer& server,
+        protobuf::BufferStream const& protobuf_bs,
+        std::string const& surface_name) = 0;
+    virtual std::shared_ptr<ClientBufferStream> make_producer_stream(
+       MirConnection* allocating_connection,
+       protobuf::DisplayServer& server,
+       protobuf::BufferStream const& protobuf_bs,
+       std::string const& surface_name) = 0;
 
     // For creating buffer stream owned by client.
-    virtual ClientBufferStream* make_producer_stream(protobuf::DisplayServer& server,
+    virtual ClientBufferStream* make_producer_stream(
+       MirConnection* allocating_connection,
+       protobuf::DisplayServer& server,
        protobuf::BufferStreamParameters const& params,
        mir_buffer_stream_callback callback, void* context) = 0;
 
 
=== modified file 'src/client/connection_configuration.h'
--- src/client/connection_configuration.h	2015-01-21 07:34:50 +0000
+++ src/client/connection_configuration.h	2015-05-14 07:50:26 +0000
@@ -49,13 +49,18 @@
 class EventSink;
 class EventHandlerRegister;
 
+namespace rpc
+{
+class MirBasicRpcChannel;
+}
+
 class ConnectionConfiguration
 {
 public:
     virtual ~ConnectionConfiguration() = default;
 
     virtual std::shared_ptr<ConnectionSurfaceMap> the_surface_map() = 0;
-    virtual std::shared_ptr<google::protobuf::RpcChannel> the_rpc_channel() = 0;
+    virtual std::shared_ptr<rpc::MirBasicRpcChannel> the_rpc_channel() = 0;
     virtual std::shared_ptr<mir::logging::Logger> the_logger() = 0;
     virtual std::shared_ptr<ClientPlatformFactory> the_client_platform_factory() = 0;
     virtual std::shared_ptr<input::receiver::InputPlatform> the_input_platform() = 0;
 
=== modified file 'src/client/default_client_buffer_stream_factory.cpp'
--- src/client/default_client_buffer_stream_factory.cpp	2015-04-01 20:00:19 +0000
+++ src/client/default_client_buffer_stream_factory.cpp	2015-05-14 07:50:26 +0000
@@ -57,21 +57,31 @@
 {
 }
 
-std::shared_ptr<mcl::ClientBufferStream> mcl::DefaultClientBufferStreamFactory::make_consumer_stream(mp::DisplayServer& server,
-    mp::BufferStream const& protobuf_bs, std::string const& surface_name)
-{
-    return std::make_shared<mcl::BufferStream>(server, mcl::BufferStreamMode::Consumer, client_platform, protobuf_bs, make_perf_report(logger), surface_name);
-}
-
-std::shared_ptr<mcl::ClientBufferStream> mcl::DefaultClientBufferStreamFactory::make_producer_stream(mp::DisplayServer& server,
-    mp::BufferStream const& protobuf_bs, std::string const& surface_name)
-{
-    return std::make_shared<mcl::BufferStream>(server, mcl::BufferStreamMode::Producer, client_platform, protobuf_bs, make_perf_report(logger), surface_name);
-}
-
-
-mcl::ClientBufferStream* mcl::DefaultClientBufferStreamFactory::make_producer_stream(mp::DisplayServer& server,
-    mp::BufferStreamParameters const& params, mir_buffer_stream_callback callback, void* context)
-{
-    return new mcl::BufferStream(server, client_platform, params, make_perf_report(logger), callback, context);
+std::shared_ptr<mcl::ClientBufferStream> mcl::DefaultClientBufferStreamFactory::make_consumer_stream(
+    MirConnection* allocating_connection,
+    mp::DisplayServer& server,
+    mp::BufferStream const& protobuf_bs,
+    std::string const& surface_name)
+{
+    return std::make_shared<mcl::BufferStream>(allocating_connection, server, mcl::BufferStreamMode::Consumer, client_platform, protobuf_bs, make_perf_report(logger), surface_name);
+}
+
+std::shared_ptr<mcl::ClientBufferStream> mcl::DefaultClientBufferStreamFactory::make_producer_stream(
+    MirConnection* allocating_connection,
+    mp::DisplayServer& server,
+    mp::BufferStream const& protobuf_bs,
+    std::string const& surface_name)
+{
+    return std::make_shared<mcl::BufferStream>(allocating_connection, server, mcl::BufferStreamMode::Producer, client_platform, protobuf_bs, make_perf_report(logger), surface_name);
+}
+
+
+mcl::ClientBufferStream* mcl::DefaultClientBufferStreamFactory::make_producer_stream(
+    MirConnection* allocating_connection,
+    mp::DisplayServer& server,
+    mp::BufferStreamParameters const& params,
+    mir_buffer_stream_callback callback,
+    void* context)
+{
+    return new mcl::BufferStream(allocating_connection, server, client_platform, params, make_perf_report(logger), callback, context);
 }
 
=== modified file 'src/client/default_client_buffer_stream_factory.h'
--- src/client/default_client_buffer_stream_factory.h	2015-04-01 20:00:19 +0000
+++ src/client/default_client_buffer_stream_factory.h	2015-05-14 07:50:26 +0000
@@ -21,6 +21,8 @@
 
 #include "client_buffer_stream_factory.h"
 
+class MirConnection;
+
 namespace mir
 {
 namespace logging
@@ -40,14 +42,22 @@
         std::shared_ptr<logging::Logger> const& logger);
     virtual ~DefaultClientBufferStreamFactory() = default;
 
-    std::shared_ptr<ClientBufferStream> make_consumer_stream(protobuf::DisplayServer& server,
-       protobuf::BufferStream const& protobuf_bs, std::string const& surface_name);
-    std::shared_ptr<ClientBufferStream> make_producer_stream(protobuf::DisplayServer& server,
-       protobuf::BufferStream const& protobuf_bs, std::string const& surface_name);
+    std::shared_ptr<ClientBufferStream> make_consumer_stream(
+        MirConnection* allocating_connection,
+        protobuf::DisplayServer& server,
+        protobuf::BufferStream const& protobuf_bs,
+        std::string const& surface_name) override;
+    std::shared_ptr<ClientBufferStream> make_producer_stream(
+       MirConnection* allocating_connection,
+       protobuf::DisplayServer& server,
+       protobuf::BufferStream const& protobuf_bs,
+       std::string const& surface_name) override;
 
-    ClientBufferStream* make_producer_stream(protobuf::DisplayServer& server,
+    ClientBufferStream* make_producer_stream(
+       MirConnection* allocating_connection,
+       protobuf::DisplayServer& server,
        protobuf::BufferStreamParameters const& params,
-       mir_buffer_stream_callback callback, void* context);
+       mir_buffer_stream_callback callback, void* context) override;
 
 private:
     std::shared_ptr<ClientPlatform> const client_platform;
 
=== modified file 'src/client/default_connection_configuration.cpp'
--- src/client/default_connection_configuration.cpp	2015-03-31 02:35:42 +0000
+++ src/client/default_connection_configuration.cpp	2015-05-14 07:50:26 +0000
@@ -85,7 +85,7 @@
         });
 }
 
-std::shared_ptr<google::protobuf::RpcChannel>
+std::shared_ptr<mcl::rpc::MirBasicRpcChannel>
 mcl::DefaultConnectionConfiguration::the_rpc_channel()
 {
     return rpc_channel(
 
=== modified file 'src/client/default_connection_configuration.h'
--- src/client/default_connection_configuration.h	2015-01-21 07:34:50 +0000
+++ src/client/default_connection_configuration.h	2015-05-14 07:50:26 +0000
@@ -51,7 +51,7 @@
     DefaultConnectionConfiguration(std::string const& socket_file);
 
     std::shared_ptr<ConnectionSurfaceMap> the_surface_map() override;
-    std::shared_ptr<google::protobuf::RpcChannel> the_rpc_channel() override;
+    std::shared_ptr<rpc::MirBasicRpcChannel> the_rpc_channel() override;
     std::shared_ptr<mir::logging::Logger> the_logger() override;
     std::shared_ptr<ClientPlatformFactory> the_client_platform_factory() override;
     std::shared_ptr<input::receiver::InputPlatform> the_input_platform() override;
@@ -66,7 +66,7 @@
     virtual std::shared_ptr<input::receiver::InputReceiverReport> the_input_receiver_report();
 
 protected:
-    CachedPtr<google::protobuf::RpcChannel> rpc_channel;
+    CachedPtr<rpc::MirBasicRpcChannel> rpc_channel;
     CachedPtr<mir::logging::Logger> logger;
     CachedPtr<ClientPlatformFactory> client_platform_factory;
     CachedPtr<input::receiver::InputPlatform> input_platform;
 
=== modified file 'src/client/mir_buffer_stream_api.cpp'
--- src/client/mir_buffer_stream_api.cpp	2015-03-31 02:35:42 +0000
+++ src/client/mir_buffer_stream_api.cpp	2015-05-14 07:50:26 +0000
@@ -28,6 +28,8 @@
 
 #include "mir/uncaught.h"
 
+#include "synchronous_helper.h"
+
 #include <stdexcept>
 #include <boost/throw_exception.hpp>
 
@@ -49,13 +51,6 @@
     mcl::ClientBufferStream *bs = reinterpret_cast<mcl::ClientBufferStream*>(stream);
     delete bs;
 }
-// assign_result is compatible with all 2-parameter callbacks
-void assign_result(void* result, void** context)
-{
-    if (context)
-        *context = result;
-}
-
 }
 
 
@@ -74,7 +69,7 @@
     params.set_pixel_format(format);
     params.set_buffer_usage(buffer_usage);
 
-    return connection->get_client_buffer_stream_factory()->make_producer_stream(connection->display_server(), params, callback, context)
+    return connection->get_client_buffer_stream_factory()->make_producer_stream(connection, connection->display_server(), params, callback, context)
         ->get_create_wait_handle();
 }
 catch (std::exception const& ex)
@@ -89,10 +84,14 @@
     MirBufferUsage buffer_usage)
 try
 {
-    mcl::BufferStream *stream = nullptr;
-    mir_connection_create_buffer_stream(connection, width, height, format, buffer_usage,
-        reinterpret_cast<mir_buffer_stream_callback>(assign_result), &stream)->wait_for_all();
-    return reinterpret_cast<MirBufferStream*>(dynamic_cast<mcl::ClientBufferStream*>(stream));
+    MirBufferStream* stream = nullptr;
+    make_synchronous_call(connection,
+                          mir_connection_create_buffer_stream,
+                          connection,
+                          width, height, format, buffer_usage,
+                          &assign_result<MirBufferStream>,
+                          &stream);
+    return stream;
 }
 catch (std::exception const& ex)
 {
@@ -113,7 +112,11 @@
 void mir_buffer_stream_release_sync(MirBufferStream *buffer_stream)
 {
     mcl::ClientBufferStream *bs = reinterpret_cast<mcl::ClientBufferStream*>(buffer_stream);
-    bs->release(nullptr, nullptr)->wait_for_all();
+    make_synchronous_call(bs->get_connection(),
+                          std::mem_fn(&mcl::ClientBufferStream::release),
+                          bs,
+                          &assign_result<MirBufferStream>,
+                          static_cast<void*>(nullptr));
     delete bs;
 }
 
@@ -148,9 +151,13 @@
 
 void mir_buffer_stream_swap_buffers_sync(MirBufferStream* buffer_stream)
 {
-    mir_wait_for(mir_buffer_stream_swap_buffers(buffer_stream,
-        reinterpret_cast<mir_buffer_stream_callback>(assign_result),
-        nullptr));
+    mcl::ClientBufferStream *bs = reinterpret_cast<mcl::ClientBufferStream*>(buffer_stream);
+    make_synchronous_call(
+        bs->get_connection(),
+        &mir_buffer_stream_swap_buffers,
+        buffer_stream,
+        &assign_result<MirBufferStream>,
+        static_cast<void*>(nullptr));
 }
 
 void mir_buffer_stream_get_graphics_region(
 
=== modified file 'src/client/mir_connection.cpp'
--- src/client/mir_connection.cpp	2015-04-01 19:39:19 +0000
+++ src/client/mir_connection.cpp	2015-05-14 07:50:26 +0000
@@ -24,8 +24,10 @@
 #include "mir/client_platform.h"
 #include "mir/client_platform_factory.h"
 #include "rpc/mir_basic_rpc_channel.h"
+#include "rpc/mir_protobuf_rpc_channel.h"
 #include "mir/dispatch/dispatchable.h"
-#include "mir/dispatch/simple_dispatch_thread.h"
+#include "mir/dispatch/multiplexing_dispatchable.h"
+#include "mir/dispatch/threaded_dispatcher.h"
 #include "connection_configuration.h"
 #include "display_configuration.h"
 #include "connection_surface_map.h"
@@ -33,6 +35,7 @@
 
 #include "mir/events/event_builders.h"
 #include "mir/logging/logger.h"
+#include "mir/require.h"
 
 #include <algorithm>
 #include <cstddef>
@@ -42,6 +45,7 @@
 #include <boost/exception/diagnostic_information.hpp>
 
 namespace mcl = mir::client;
+namespace mclr = mir::client::rpc;
 namespace md = mir::dispatch;
 namespace mircv = mir::input::receiver;
 namespace mev = mir::events;
@@ -99,10 +103,15 @@
 {
 }
 
+MirConnection::MirConnection(mir::client::ConnectionConfiguration& conf) :
+    MirConnection(conf, DispatchType::automatic)
+{
+}
+
 MirConnection::MirConnection(
-    mir::client::ConnectionConfiguration& conf) :
+    mir::client::ConnectionConfiguration& conf, DispatchType dispatch) :
         deregisterer{this},
-        channel(conf.the_rpc_channel()),
+        channel{conf.the_rpc_channel()},
         server(channel.get(), ::google::protobuf::Service::STUB_DOESNT_OWN_CHANNEL),
         debug(channel.get(), ::google::protobuf::Service::STUB_DOESNT_OWN_CHANNEL),
         logger(conf.the_logger()),
@@ -113,7 +122,11 @@
         lifecycle_control(conf.the_lifecycle_control()),
         surface_map(conf.the_surface_map()),
         event_handler_register(conf.the_event_handler_register()),
-        eventloop{new md::SimpleDispatchThread{std::dynamic_pointer_cast<md::Dispatchable>(channel)}}
+        dispatcher{std::shared_ptr<md::MultiplexingDispatchable>(new md::MultiplexingDispatchable{channel})},
+        eventloop{dispatch == DispatchType::automatic ?
+                      new md::ThreadedDispatcher{"I/O loop", dispatcher} :
+                      nullptr
+                  }
 {
     connect_result.set_error("connect not called");
     {
@@ -325,6 +338,36 @@
     return &disconnect_wait_handle;
 }
 
+mir::Fd MirConnection::watch_fd() const
+{
+    return eventloop ? mir::Fd{} : dispatcher->watch_fd();
+}
+
+void MirConnection::dispatch()
+{
+    mir::require(!eventloop);
+
+    dispatcher->dispatch(md::FdEvent::readable);
+}
+
+void MirConnection::add_dispatchee(std::shared_ptr<mir::dispatch::Dispatchable> const& dispatchee)
+{
+    dispatcher->add_watch(dispatchee);
+    if (eventloop)
+    {
+        eventloop->add_thread();
+    }
+}
+
+void MirConnection::remove_dispatchee(std::shared_ptr<mir::dispatch::Dispatchable> const& dispatchee)
+{
+    dispatcher->remove_watch(dispatchee);
+    if (eventloop)
+    {
+        eventloop->remove_thread();
+    }
+}
+
 void MirConnection::done_platform_operation(
     mir_platform_operation_callback callback, void* context)
 {
@@ -568,3 +611,8 @@
 {
     return logger;
 }
+
+void MirConnection::process_next_request_first()
+{
+    channel->process_next_request_first();
+}
 
=== modified file 'src/client/mir_connection.h'
--- src/client/mir_connection.h	2015-03-31 02:35:42 +0000
+++ src/client/mir_connection.h	2015-05-14 07:50:26 +0000
@@ -26,6 +26,8 @@
 
 #include <mutex>
 
+#include "rpc/mir_basic_rpc_channel.h"
+
 #include "mir_protobuf.pb.h"
 
 #include "mir_toolkit/mir_client_library.h"
@@ -36,6 +38,8 @@
 
 #include "mir_wait_handle.h"
 
+#include "mir/fd.h"
+
 namespace mir
 {
 class SharedLibrary;
@@ -51,10 +55,6 @@
 class LifecycleControl;
 class EventHandlerRegister;
 
-namespace rpc
-{
-class MirBasicRpcChannel;
-}
 }
 
 namespace input
@@ -72,9 +72,17 @@
 
 namespace dispatch
 {
-class SimpleDispatchThread;
-}
-}
+class ThreadedDispatcher;
+class Dispatchable;
+class MultiplexingDispatchable;
+}
+}
+
+enum class DispatchType
+{
+    automatic,
+    manual
+};
 
 struct MirConnection : mir::client::ClientContext
 {
@@ -82,6 +90,7 @@
     MirConnection(std::string const& error_message);
 
     MirConnection(mir::client::ConnectionConfiguration& conf);
+    MirConnection(mir::client::ConnectionConfiguration &conf, DispatchType dispatch);
     ~MirConnection() noexcept;
 
     MirConnection(MirConnection const &) = delete;
@@ -107,6 +116,12 @@
 
     MirWaitHandle* disconnect();
 
+    mir::Fd watch_fd() const;
+    void dispatch();
+
+    void add_dispatchee(std::shared_ptr<mir::dispatch::Dispatchable> const& dispatchee);
+    void remove_dispatchee(std::shared_ptr<mir::dispatch::Dispatchable> const& dispatchee);
+
     MirWaitHandle* platform_operation(
         MirPlatformMessage const* request,
         mir_platform_operation_callback callback, void* context);
@@ -140,6 +155,7 @@
     mir::protobuf::DisplayServer& display_server();
     std::shared_ptr<mir::logging::Logger> const& the_logger() const;
 
+    void process_next_request_first();
 private:
     void populate_server_package(MirPlatformPackage& platform_package) override;
     // MUST be first data member so it is destroyed last.
@@ -152,7 +168,7 @@
 
     std::mutex mutex; // Protects all members of *this (except release_wait_handles)
 
-    std::shared_ptr<google::protobuf::RpcChannel> const channel;
+    std::shared_ptr<mir::client::rpc::MirBasicRpcChannel> const channel;
     mir::protobuf::DisplayServer::Stub server;
     mir::protobuf::Debug::Stub debug;
     std::shared_ptr<mir::logging::Logger> const logger;
@@ -189,7 +205,9 @@
 
     std::shared_ptr<mir::client::EventHandlerRegister> const event_handler_register;
 
-    std::unique_ptr<mir::dispatch::SimpleDispatchThread> const eventloop;
+    std::shared_ptr<mir::dispatch::MultiplexingDispatchable> const dispatcher;
+    std::unique_ptr<mir::dispatch::ThreadedDispatcher> const eventloop;
+
     
     std::shared_ptr<mir::client::ClientBufferStreamFactory> buffer_stream_factory;
 
 
=== modified file 'src/client/mir_connection_api.cpp'
--- src/client/mir_connection_api.cpp	2015-03-31 02:35:42 +0000
+++ src/client/mir_connection_api.cpp	2015-05-14 07:50:26 +0000
@@ -39,6 +39,7 @@
 #include <unordered_set>
 #include <cstddef>
 #include <cstring>
+#include <poll.h>
 
 namespace mcl = mir::client;
 
@@ -98,7 +99,20 @@
             try
             {
                 auto wait_handle = connection->disconnect();
-                wait_handle->wait_for_all();
+                if (connection->watch_fd() != mir::Fd::invalid)
+                {
+                    pollfd fd;
+                    fd.fd = connection->watch_fd();
+                    fd.events = POLLIN;
+                    while(!wait_handle->ready() && (poll(&fd, 1, -1) > 0))
+                    {
+                        connection->dispatch();
+                    }
+                }
+                else
+                {
+                    wait_handle->wait_for_all();
+                }
             }
             catch (std::exception const& ex)
             {
@@ -165,6 +179,41 @@
     return conn;
 }
 
+MirConnection* mir_connect_with_manual_dispatch(
+        char const* server,
+        char const* app_name,
+        mir_connected_callback callback,
+        void* context)
+{
+    try
+    {
+        std::string sock;
+        if (server)
+            sock = server;
+        else
+        {
+            auto socket_env = getenv("MIR_SOCKET");
+            if (socket_env)
+                sock = socket_env;
+            else
+                sock = mir::default_server_socket;
+        }
+
+        mcl::DefaultConnectionConfiguration conf{sock};
+
+        std::unique_ptr<MirConnection> connection{new MirConnection(conf, DispatchType::manual)};
+        connection->connect(app_name, callback, context);
+        return connection.release();
+    }
+    catch (std::exception const& x)
+    {
+        MirConnection* error_connection = new MirConnection(x.what());
+        mcl::ErrorConnections::instance().insert(error_connection);
+        callback(error_connection, context);
+        return error_connection;
+    }
+}
+
 bool mir_connection_is_valid(MirConnection* connection)
 {
     return MirConnection::is_valid(connection);
@@ -187,6 +236,16 @@
     }
 }
 
+int mir_connection_get_event_fd(MirConnection* connection)
+{
+    return connection->watch_fd();
+}
+
+void mir_connection_dispatch(MirConnection* connection)
+{
+    connection->dispatch();
+}
+
 void mir_connection_get_platform(
     MirConnection* connection,
     MirPlatformPackage* platform_package)
 
=== modified file 'src/client/mir_prompt_session_api.cpp'
--- src/client/mir_prompt_session_api.cpp	2015-03-31 02:35:42 +0000
+++ src/client/mir_prompt_session_api.cpp	2015-05-14 07:50:26 +0000
@@ -24,6 +24,8 @@
 
 #include "mir/uncaught.h"
 
+#include "synchronous_helper.h"
+
 #include <stdexcept>
 #include <boost/throw_exception.hpp>
 
@@ -44,9 +46,14 @@
         if (state_change_callback)
             prompt_session->register_prompt_session_state_change_callback(state_change_callback, context);
 
-        mir_wait_for(prompt_session->start(application_pid,
-                     null_callback,
-                     nullptr));
+        make_synchronous_call(
+            connection,
+            std::mem_fn(&MirPromptSession::start),
+            prompt_session,
+            application_pid,
+            &assign_result<MirPromptSession>,
+            static_cast<void*>(nullptr));
+
         return prompt_session;
     }
     catch (std::exception const& ex)
 
=== modified file 'src/client/mir_screencast.cpp'
--- src/client/mir_screencast.cpp	2015-04-25 11:38:31 +0000
+++ src/client/mir_screencast.cpp	2015-05-14 07:50:26 +0000
@@ -30,13 +30,15 @@
 namespace geom = mir::geometry;
 
 MirScreencast::MirScreencast(
+    MirConnection* allocating_connection,
     geom::Rectangle const& region,
     geom::Size const& size,
     MirPixelFormat pixel_format,
     mir::protobuf::DisplayServer& server,
     std::shared_ptr<mcl::ClientBufferStreamFactory> const& buffer_stream_factory,
     mir_screencast_callback callback, void* context)
-    : server(server),
+    : allocating_connection{allocating_connection},
+      server(server),
       output_size{size},
       buffer_stream_factory{buffer_stream_factory}
 {
@@ -102,12 +104,16 @@
 }
 
 void MirScreencast::screencast_created(
-    mir_screencast_callback callback, void* context)
+    mir_screencast_callback callback,
+    void* context)
 {
     if (!protobuf_screencast.has_error())
     {
-        buffer_stream = buffer_stream_factory->make_consumer_stream(server,
-            protobuf_screencast.buffer_stream(), "MirScreencast");
+        buffer_stream = buffer_stream_factory->make_consumer_stream(
+            allocating_connection,
+            server,
+            protobuf_screencast.buffer_stream(),
+            "MirScreencast");
     }
 
     callback(this, context);
@@ -125,3 +131,8 @@
 {
     return buffer_stream.get();
 }
+
+MirConnection* MirScreencast::get_connection()
+{
+    return allocating_connection;
+}
 
=== modified file 'src/client/mir_screencast.h'
--- src/client/mir_screencast.h	2015-03-31 02:35:42 +0000
+++ src/client/mir_screencast.h	2015-05-14 07:50:26 +0000
@@ -41,6 +41,7 @@
 {
 public:
     MirScreencast(
+        MirConnection* allocating_connection,
         mir::geometry::Rectangle const& region,
         mir::geometry::Size const& size,
         MirPixelFormat pixel_format,
@@ -60,12 +61,14 @@
 
     mir::client::ClientBufferStream* get_buffer_stream();
 
+    MirConnection* get_connection();
 private:
     void screencast_created(
         mir_screencast_callback callback, void* context);
     void released(
         mir_screencast_callback callback, void* context);
 
+    MirConnection* const allocating_connection;
     mir::protobuf::DisplayServer& server;
     mir::geometry::Size const output_size;
     std::shared_ptr<mir::client::ClientBufferStreamFactory> const buffer_stream_factory;
 
=== modified file 'src/client/mir_screencast_api.cpp'
--- src/client/mir_screencast_api.cpp	2015-03-31 02:35:42 +0000
+++ src/client/mir_screencast_api.cpp	2015-05-14 07:50:26 +0000
@@ -25,6 +25,8 @@
 
 #include "mir/uncaught.h"
 
+#include "synchronous_helper.h"
+
 #include <stdexcept>
 #include <boost/throw_exception.hpp>
 
@@ -56,6 +58,7 @@
 
         std::unique_ptr<MirScreencast> screencast_uptr{
             new MirScreencast{
+                connection,
                 region,
                 size,
                 parameters->pixel_format,
@@ -63,7 +66,18 @@
                 connection->get_client_buffer_stream_factory(),
                 null_callback, nullptr}};
 
-        screencast_uptr->creation_wait_handle()->wait_for_all();
+
+        if (connection->watch_fd() != mir::Fd::invalid)
+        {
+            dispatch_connection_until(connection, [&screencast_uptr]()
+            {
+                return screencast_uptr->creation_wait_handle()->ready();
+            });
+        }
+        else
+        {
+            screencast_uptr->creation_wait_handle()->wait_for_all();
+        }
 
         if (screencast_uptr->valid())
         {
@@ -82,7 +96,11 @@
 
 void mir_screencast_release_sync(MirScreencast* screencast)
 {
-    screencast->release(null_callback, nullptr)->wait_for_all();
+    make_synchronous_call(screencast->get_connection(),
+                          std::mem_fn(&MirScreencast::release),
+                          screencast,
+                          &assign_result<MirScreencast>,
+                          static_cast<void*>(nullptr));
     delete screencast;
 }
 
 
=== modified file 'src/client/mir_surface.cpp'
--- src/client/mir_surface.cpp	2015-04-28 10:30:44 +0000
+++ src/client/mir_surface.cpp	2015-05-14 07:50:26 +0000
@@ -23,7 +23,7 @@
 #include "cursor_configuration.h"
 #include "client_buffer_stream_factory.h"
 #include "mir_connection.h"
-#include "mir/dispatch/simple_dispatch_thread.h"
+#include "mir/dispatch/threaded_dispatcher.h"
 #include "mir/input/input_platform.h"
 #include "mir/input/xkb_mapper.h"
 
@@ -179,7 +179,10 @@
 
     std::lock_guard<decltype(mutex)> lock(mutex);
 
-    input_thread.reset();
+    if (input_dispatcher)
+    {
+        connection->remove_dispatchee(input_dispatcher);
+    }
 
     for (auto i = 0, end = surface.fd_size(); i != end; ++i)
         close(surface.fd(i));
@@ -220,6 +223,11 @@
     return false;
 }
 
+MirConnection* MirSurface::get_connection() const
+{
+    return connection;
+}
+
 MirWaitHandle* MirSurface::get_create_wait_handle()
 {
     return &create_wait_handle;
@@ -252,7 +260,7 @@
             std::lock_guard<decltype(mutex)> lock(mutex);
 
             buffer_stream = buffer_stream_factory->
-                make_producer_stream(*server, surface.buffer_stream(), name);
+                make_producer_stream(connection, *server, surface.buffer_stream(), name);
 
             for(int i = 0; i < surface.attributes_size(); i++)
             {
@@ -458,7 +466,11 @@
 {
     std::lock_guard<decltype(mutex)> lock(mutex);
 
-    input_thread.reset();
+    if (input_dispatcher)
+    {
+        connection->remove_dispatchee(input_dispatcher);
+        input_dispatcher.reset();
+    }
 
     if (callback)
     {
@@ -468,10 +480,10 @@
 
         if (surface.fd_size() > 0 && handle_event_callback)
         {
-            auto input_dispatcher = input_platform->create_input_receiver(surface.fd(0),
-                                                                          keymapper,
-                                                                          handle_event_callback);
-            input_thread = std::make_shared<md::SimpleDispatchThread>(input_dispatcher);
+            input_dispatcher = input_platform->create_input_receiver(surface.fd(0),
+                                                                     keymapper,
+                                                                     handle_event_callback);
+            connection->add_dispatchee(input_dispatcher);
         }
     }
 }
 
=== modified file 'src/client/mir_surface.h'
--- src/client/mir_surface.h	2015-04-28 08:59:22 +0000
+++ src/client/mir_surface.h	2015-05-14 07:50:26 +0000
@@ -39,7 +39,7 @@
 {
 namespace dispatch
 {
-class SimpleDispatchThread;
+class Dispatchable;
 }
 namespace input
 {
@@ -153,6 +153,8 @@
     MirWaitHandle* modify(MirSurfaceSpec const& changes);
 
     static bool is_valid(MirSurface* query);
+
+    MirConnection* get_connection() const;
 private:
     mutable std::mutex mutex; // Protects all members of *this
 
@@ -191,7 +193,7 @@
     MirOrientation orientation = mir_orientation_normal;
 
     std::function<void(MirEvent const*)> handle_event_callback;
-    std::shared_ptr<mir::dispatch::SimpleDispatchThread> input_thread;
+    std::shared_ptr<mir::dispatch::Dispatchable> input_dispatcher;
 };
 
 #endif /* MIR_CLIENT_PRIVATE_MIR_WAIT_HANDLE_H_ */
 
=== modified file 'src/client/mir_surface_api.cpp'
--- src/client/mir_surface_api.cpp	2015-04-30 12:27:33 +0000
+++ src/client/mir_surface_api.cpp	2015-05-14 07:50:26 +0000
@@ -26,24 +26,13 @@
 #include "mir_surface.h"
 #include "error_connections.h"
 #include "mir/uncaught.h"
+#include "synchronous_helper.h"
 
 #include <boost/exception/diagnostic_information.hpp>
 #include <functional>
 
 namespace mcl = mir::client;
 
-namespace
-{
-
-// assign_result is compatible with all 2-parameter callbacks
-void assign_result(void* result, void** context)
-{
-    if (context)
-        *context = result;
-}
-
-}
-
 MirSurfaceSpec* mir_connection_create_spec_for_normal_surface(MirConnection* connection,
                                                               int width, int height,
                                                               MirPixelFormat format)
@@ -128,9 +117,12 @@
 {
     MirSurface* surface = nullptr;
 
-    mir_wait_for(mir_surface_create(requested_specification,
-        reinterpret_cast<mir_surface_callback>(assign_result),
-        &surface));
+    make_synchronous_call(
+        requested_specification->connection,
+        &mir_surface_create,
+        requested_specification,
+        &assign_result<MirSurface>,
+        &surface);
 
     return surface;
 }
@@ -364,9 +356,11 @@
 
 void mir_surface_release_sync(MirSurface* surface)
 {
-    mir_wait_for(mir_surface_release(surface,
-        reinterpret_cast<mir_surface_callback>(assign_result),
-        nullptr));
+    make_synchronous_call(surface->get_connection(),
+                          mir_surface_release,
+                          surface,
+                          &assign_result<MirSurface>,
+                          static_cast<void*>(nullptr));
 }
 
 int mir_surface_get_id(MirSurface* /*surface*/)
 
=== modified file 'src/client/mir_wait_api.cpp'
--- src/client/mir_wait_api.cpp	2014-03-31 14:36:08 +0000
+++ src/client/mir_wait_api.cpp	2015-05-14 07:50:26 +0000
@@ -30,3 +30,12 @@
     if (wait_handle)
         wait_handle->wait_for_one();
 }
+
+bool mir_wait_handle_ready(MirWaitHandle* wait_handle)
+{
+    if (wait_handle)
+    {
+        return wait_handle->ready();
+    }
+    return true;
+}
 
=== modified file 'src/client/mir_wait_handle.cpp'
--- src/client/mir_wait_handle.cpp	2015-01-21 07:34:50 +0000
+++ src/client/mir_wait_handle.cpp	2015-05-14 07:50:26 +0000
@@ -74,3 +74,9 @@
     --expecting;
 }
 
+bool MirWaitHandle::ready()
+{
+    std::lock_guard<std::mutex> lock{guard};
+
+    return received == expecting;
+}
 
=== modified file 'src/client/mir_wait_handle.h'
--- src/client/mir_wait_handle.h	2013-10-03 03:44:08 +0000
+++ src/client/mir_wait_handle.h	2015-05-14 07:50:26 +0000
@@ -39,6 +39,7 @@
     void wait_for_all();
     void wait_for_one();
     void wait_for_pending(std::chrono::milliseconds limit);
+    bool ready();
 
 private:
     std::mutex guard;
 
=== modified file 'src/client/rpc/make_rpc_channel.h'
--- src/client/rpc/make_rpc_channel.h	2015-01-21 07:34:50 +0000
+++ src/client/rpc/make_rpc_channel.h	2015-05-14 07:50:26 +0000
@@ -20,8 +20,6 @@
 
 #include <memory>
 
-namespace google { namespace protobuf { class RpcChannel; } }
-
 namespace mir
 {
 namespace client
@@ -34,8 +32,9 @@
 namespace rpc
 {
 class RpcReport;
+class MirBasicRpcChannel;
 
-std::shared_ptr<google::protobuf::RpcChannel>
+std::shared_ptr<MirBasicRpcChannel>
 make_rpc_channel(std::string const& name,
                  std::shared_ptr<SurfaceMap> const& map,
                  std::shared_ptr<DisplayConfiguration> const& disp_conf,
 
=== modified file 'src/client/rpc/make_socket_rpc_channel.cpp'
--- src/client/rpc/make_socket_rpc_channel.cpp	2015-03-31 02:35:42 +0000
+++ src/client/rpc/make_socket_rpc_channel.cpp	2015-05-14 07:50:26 +0000
@@ -40,7 +40,7 @@
 } const fd_prefix("fd://");
 }
 
-std::shared_ptr<google::protobuf::RpcChannel>
+std::shared_ptr<mclr::MirBasicRpcChannel>
 mclr::make_rpc_channel(std::string const& name,
                        std::shared_ptr<mcl::SurfaceMap> const& map,
                        std::shared_ptr<mcl::DisplayConfiguration> const& disp_conf,
 
=== modified file 'src/client/rpc/mir_basic_rpc_channel.cpp'
--- src/client/rpc/mir_basic_rpc_channel.cpp	2015-01-21 07:34:50 +0000
+++ src/client/rpc/mir_basic_rpc_channel.cpp	2015-05-14 07:50:26 +0000
@@ -38,13 +38,19 @@
 void mclrd::PendingCallCache::save_completion_details(
     mir::protobuf::wire::Invocation const& invoke,
     google::protobuf::Message* response,
-    std::shared_ptr<google::protobuf::Closure> const& complete)
+    google::protobuf::Closure* complete)
 {
     std::unique_lock<std::mutex> lock(mutex);
 
     pending_calls[invoke.id()] = PendingCall(response, complete);
 }
 
+google::protobuf::Message* mclrd::PendingCallCache::message_for_result(mir::protobuf::wire::Result& result)
+{
+    std::unique_lock<std::mutex> lock(mutex);
+    return pending_calls.at(result.id()).response;
+}
+
 void mclrd::PendingCallCache::complete_response(mir::protobuf::wire::Result& result)
 {
     PendingCall completion;
@@ -66,7 +72,6 @@
     else
     {
         rpc_report->complete_response(result);
-        completion.response->ParseFromString(result.response());
         completion.complete->Run();
     }
 }
 
=== modified file 'src/client/rpc/mir_basic_rpc_channel.h'
--- src/client/rpc/mir_basic_rpc_channel.h	2015-01-21 07:34:50 +0000
+++ src/client/rpc/mir_basic_rpc_channel.h	2015-05-14 07:50:26 +0000
@@ -19,6 +19,8 @@
 #ifndef MIR_CLIENT_RPC_MIR_BASIC_RPC_CHANNEL_H_
 #define MIR_CLIENT_RPC_MIR_BASIC_RPC_CHANNEL_H_
 
+#include "mir/dispatch/dispatchable.h"
+
 #include <google/protobuf/service.h>
 #include <google/protobuf/descriptor.h>
 
@@ -57,8 +59,10 @@
     void save_completion_details(
         mir::protobuf::wire::Invocation const& invoke,
         google::protobuf::Message* response,
-        std::shared_ptr<google::protobuf::Closure> const& complete);
-
+        google::protobuf::Closure* complete);
+
+
+    google::protobuf::Message* message_for_result(mir::protobuf::wire::Result& result);
 
     void complete_response(mir::protobuf::wire::Result& result);
 
@@ -72,14 +76,14 @@
     {
         PendingCall(
             google::protobuf::Message* response,
-            std::shared_ptr<google::protobuf::Closure> const& target)
+            google::protobuf::Closure* target)
         : response(response), complete(target) {}
 
         PendingCall()
         : response(0), complete() {}
 
         google::protobuf::Message* response;
-        std::shared_ptr<google::protobuf::Closure> complete;
+        google::protobuf::Closure* complete;
     };
 
     std::mutex mutable mutex;
@@ -88,12 +92,15 @@
 };
 }
 
-class MirBasicRpcChannel : public google::protobuf::RpcChannel
+class MirBasicRpcChannel
+    : public google::protobuf::RpcChannel,
+      public mir::dispatch::Dispatchable
 {
 public:
     MirBasicRpcChannel();
     ~MirBasicRpcChannel();
 
+    virtual void process_next_request_first() = 0;
 protected:
     mir::protobuf::wire::Invocation invocation_for(
         google::protobuf::MethodDescriptor const* method,
 
=== modified file 'src/client/rpc/mir_protobuf_rpc_channel.cpp'
--- src/client/rpc/mir_protobuf_rpc_channel.cpp	2015-05-01 14:47:55 +0000
+++ src/client/rpc/mir_protobuf_rpc_channel.cpp	2015-05-14 07:50:26 +0000
@@ -60,7 +60,9 @@
     lifecycle_control(lifecycle_control),
     event_sink(event_sink),
     disconnected(false),
-    transport{std::move(transport)}
+    transport{std::move(transport)},
+    delayed_processor{std::make_shared<md::ActionQueue>()},
+    multiplexer{this->transport, delayed_processor}
 {
     class NullDeleter
     {
@@ -105,8 +107,7 @@
     }
 }
 
-void mclr::MirProtobufRpcChannel::receive_file_descriptors(google::protobuf::Message* response,
-    google::protobuf::Closure* complete)
+void mclr::MirProtobufRpcChannel::receive_file_descriptors(google::protobuf::Message* response)
 {
     auto const message_type = response->GetTypeName();
 
@@ -163,7 +164,6 @@
     receive_any_file_descriptors_for(platform);
     receive_any_file_descriptors_for(socket_fd);
     receive_any_file_descriptors_for(platform_operation_message);
-    complete->Run();
 }
 
 void mclr::MirProtobufRpcChannel::CallMethod(
@@ -193,10 +193,13 @@
 
     rpc_report->invocation_requested(invocation);
 
-    std::shared_ptr<google::protobuf::Closure> callback(
-        google::protobuf::NewPermanentCallback(this, &MirProtobufRpcChannel::receive_file_descriptors, response, complete));
+    pending_calls.save_completion_details(invocation, response, complete);
 
-    pending_calls.save_completion_details(invocation, response, callback);
+    if (prioritise_next_request)
+    {
+        id_to_wait_for = invocation.id();
+        prioritise_next_request = false;
+    }
 
     send_message(invocation, invocation, fds);
 }
@@ -317,7 +320,7 @@
      */
     std::lock_guard<decltype(read_mutex)> lock(read_mutex);
 
-    mir::protobuf::wire::Result result;
+    auto result = std::make_unique<mir::protobuf::wire::Result>();
     try
     {
         uint16_t message_size;
@@ -327,9 +330,9 @@
         body_bytes.resize(message_size);
         transport->receive_data(body_bytes.data(), message_size);
 
-        result.ParseFromArray(body_bytes.data(), message_size);
+        result->ParseFromArray(body_bytes.data(), message_size);
 
-        rpc_report->result_receipt_succeeded(result);
+        rpc_report->result_receipt_succeeded(*result);
     }
     catch (std::exception const& x)
     {
@@ -339,14 +342,39 @@
 
     try
     {
-        for (int i = 0; i != result.events_size(); ++i)
+        for (int i = 0; i != result->events_size(); ++i)
         {
-            process_event_sequence(result.events(i));
+            process_event_sequence(result->events(i));
         }
 
-        if (result.has_id())
+        if (result->has_id())
         {
-            pending_calls.complete_response(result);
+            auto result_message = pending_calls.message_for_result(*result);
+            result_message->ParseFromString(result->response());
+            receive_file_descriptors(result_message);
+
+            if (id_to_wait_for)
+            {
+                if (result->id() == id_to_wait_for.value())
+                {
+                    pending_calls.complete_response(*result);
+                    multiplexer.add_watch(delayed_processor);
+                }
+                else
+                {
+                    // It's too difficult to convince C++ to move this lambda everywhere, so
+                    // just give up and let it pretend its a shared_ptr.
+                    std::shared_ptr<mir::protobuf::wire::Result> appeaser{std::move(result)};
+                    delayed_processor->enqueue([delayed_result = std::move(appeaser), this]() mutable
+                    {
+                        pending_calls.complete_response(*delayed_result);
+                    });
+                }
+            }
+            else
+            {
+                pending_calls.complete_response(*result);
+            }
         }
     }
     catch (std::exception const& x)
@@ -354,7 +382,7 @@
         // TODO: This is dangerous as an error in result processing could cause a wait handle
         // to never fire. Could perhaps fix by catching and setting error on the response before invoking
         // callback ~racarr
-        rpc_report->result_processing_failed(result, x);
+        rpc_report->result_processing_failed(*result, x);
     }
 }
 
@@ -365,15 +393,21 @@
 
 mir::Fd mir::client::rpc::MirProtobufRpcChannel::watch_fd() const
 {
-    return transport->watch_fd();
+    return multiplexer.watch_fd();
 }
 
 bool mir::client::rpc::MirProtobufRpcChannel::dispatch(md::FdEvents events)
 {
-    return transport->dispatch(events);
+    return multiplexer.dispatch(events);
 }
 
 md::FdEvents mclr::MirProtobufRpcChannel::relevant_events() const
 {
-    return transport->relevant_events();
+    return multiplexer.relevant_events();
+}
+
+void mclr::MirProtobufRpcChannel::process_next_request_first()
+{
+    prioritise_next_request = true;
+    multiplexer.remove_watch(delayed_processor);
 }
 
=== modified file 'src/client/rpc/mir_protobuf_rpc_channel.h'
--- src/client/rpc/mir_protobuf_rpc_channel.h	2015-03-31 02:35:42 +0000
+++ src/client/rpc/mir_protobuf_rpc_channel.h	2015-05-14 07:50:26 +0000
@@ -22,23 +22,21 @@
 #include "mir_basic_rpc_channel.h"
 #include "stream_transport.h"
 #include "mir/dispatch/dispatchable.h"
+#include "mir/dispatch/multiplexing_dispatchable.h"
+#include "mir/dispatch/action_queue.h"
+
+#include "mir_protobuf_wire.pb.h"
 
 #include <google/protobuf/service.h>
 #include <google/protobuf/descriptor.h>
 
 #include <thread>
 #include <atomic>
+#include <list>
+#include <experimental/optional>
 
 namespace mir
 {
-namespace protobuf
-{
-namespace wire
-{
-class Invocation;
-class Result;
-}
-}
 
 namespace client
 {
@@ -53,8 +51,7 @@
 
 class MirProtobufRpcChannel :
         public MirBasicRpcChannel,
-        public StreamTransport::Observer,
-        public dispatch::Dispatchable
+        public StreamTransport::Observer
 {
 public:
     MirProtobufRpcChannel(std::unique_ptr<StreamTransport> transport,
@@ -74,6 +71,20 @@
     Fd watch_fd() const override;
     bool dispatch(mir::dispatch::FdEvents events) override;
     mir::dispatch::FdEvents relevant_events() const override;
+
+    /**
+     * \brief Switch the RpcChannel into out-of-order mode
+     *
+     * The first CallMethod after this method is called will be processed
+     * out of order - no server responses will be processed until the response
+     * for the next CallMethod is processed.
+     *
+     * After the response for the next CallMethod is processed, normal processing
+     * is resumed.
+     *
+     * No messages are discarded, only delayed.
+     */
+    void process_next_request_first() override;
 private:
     virtual void CallMethod(const google::protobuf::MethodDescriptor* method, google::protobuf::RpcController*,
         const google::protobuf::Message* parameters, google::protobuf::Message* response,
@@ -86,7 +97,7 @@
     detail::SendBuffer header_bytes;
     detail::SendBuffer body_bytes;
 
-    void receive_file_descriptors(google::protobuf::Message* response, google::protobuf::Closure* complete);
+    void receive_file_descriptors(google::protobuf::Message* response);
     template<class MessageType>
     void receive_any_file_descriptors_for(MessageType* response);
     void send_message(mir::protobuf::wire::Invocation const& body,
@@ -106,14 +117,22 @@
     std::mutex read_mutex;
     std::mutex write_mutex;
 
+    bool prioritise_next_request{false};
+    std::experimental::optional<uint32_t> id_to_wait_for;
+
     /* We use the guarantee that the transport's destructor blocks until
      * pending processing has finished to ensure that on_data_available()
      * isn't called after the members it relies on are destroyed.
      *
-     * This means the transport field must appear after any field used
-     * by on_data_available. For simplicity, put it last.
+     * This means that anything that owns a reference to the transport
+     * needs to be after anything that can be accessed from on_data_available().
+     *
+     * For simplicity's sake keep all of the dispatch infrastructure at the
+     * end to guarantee this.
      */
-    std::unique_ptr<StreamTransport> transport;
+    std::shared_ptr<StreamTransport> const transport;
+    std::shared_ptr<mir::dispatch::ActionQueue> const delayed_processor;
+    mir::dispatch::MultiplexingDispatchable multiplexer;
 };
 
 }
 
=== modified file 'src/client/symbols.map'
--- src/client/symbols.map	2015-05-07 19:41:08 +0000
+++ src/client/symbols.map	2015-05-14 07:50:26 +0000
@@ -196,6 +196,10 @@
     mir_buffer_stream_release;
     mir_buffer_stream_release_sync;
     mir_buffer_stream_is_valid;
+    mir_connect_with_manual_dispatch;
+    mir_connection_get_event_fd;
+    mir_connection_dispatch;
+    mir_wait_handle_ready;
 } MIR_CLIENT_8.3;
 
 MIR_CLIENT_DETAIL_8 {
 
=== added file 'src/client/synchronous_helper.cpp'
--- src/client/synchronous_helper.cpp	1970-01-01 00:00:00 +0000
+++ src/client/synchronous_helper.cpp	2015-05-14 07:50:26 +0000
@@ -0,0 +1,33 @@
+/*
+ * 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 "synchronous_helper.h"
+
+#include <poll.h>
+
+void dispatch_connection_until(MirConnection* connection, std::function<bool()> predicate)
+{
+    pollfd fd;
+    fd.fd = connection->watch_fd();
+    fd.events = POLLIN;
+    while(!predicate() && (poll(&fd, 1, -1) > 0))
+    {
+        connection->dispatch();
+    }
+}
+
 
=== added file 'src/client/synchronous_helper.h'
--- src/client/synchronous_helper.h	1970-01-01 00:00:00 +0000
+++ src/client/synchronous_helper.h	2015-05-14 07:50:26 +0000
@@ -0,0 +1,160 @@
+/*
+ * 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_CLIENT_SYNCHRONOUS_H_
+#define MIR_CLIENT_SYNCHRONOUS_H_
+
+#include "mir_connection.h"
+#include "mir_wait_handle.h"
+
+#include <functional>
+#include <type_traits>
+
+template<typename Result>
+void assign_result(Result* result, void* ctx)
+{
+    auto assignee = reinterpret_cast<Result**>(ctx);
+    if (assignee)
+    {
+        *assignee = result;
+    }
+}
+
+template<typename Callback>
+struct SynchronousContext
+{
+    Callback real_callback;
+    bool complete;
+    void* userdata;
+};
+
+template<typename Callable, class Tuple, std::size_t...I>
+void apply_substituting_last_arg(
+    Callable&& function,
+    Tuple&& args,
+    std::index_sequence<I...>,
+    void* context)
+{
+    return std::forward<Callable>(function)(
+        std::get<I>(std::forward<Tuple>(args))...,
+        context);
+}
+
+
+template<typename... Args>
+void synchronous_wrapper(Args... args)
+{
+    std::tuple<Args...> arguments{args...};
+    constexpr std::size_t arg_count = sizeof...(Args);
+
+    auto wrapped_context = reinterpret_cast<SynchronousContext<void(*)(Args...)>*>(std::get<arg_count - 1>(arguments));
+    if (wrapped_context->real_callback)
+    {
+        apply_substituting_last_arg(
+               wrapped_context->real_callback,
+               std::forward_as_tuple(args...),
+               std::make_index_sequence<arg_count - 1>(),
+               wrapped_context->userdata);
+    }
+    wrapped_context->complete = true;
+}
+
+
+template<typename Callable, typename Tuple, std::size_t...I>
+constexpr decltype(auto) call_impl(Callable&& function, Tuple&& args, std::index_sequence<I...>)
+{
+    return std::forward<Callable>(function)(std::get<I>(std::forward<Tuple>((args)))...);
+}
+
+template<typename Callable, typename Tuple>
+constexpr decltype(auto) apply(Callable&& function, Tuple&& args)
+{
+    return call_impl(
+        std::forward<Callable>(function),
+        std::forward<Tuple>(args),
+        std::make_index_sequence<std::tuple_size<std::decay_t<Tuple>>{}>{});
+}
+
+void dispatch_connection_until(MirConnection* connection, std::function<bool()> predicate);
+
+
+/**
+ * \brief Make a synchronous RPC call
+ *
+ * This wrapper takes care of manually dispatching the MirConnection if it is in
+ * manual dispatch mode, or waiting for the operation to complete if in automatic
+ * dispatch mode.
+ *
+ * Given a function mir_foo_do_thing(MirFoo* target, int arg, foo_callback callback, void* context)
+ * the correct way to call this function is
+ * make_synchronous_call(connection,
+ *                       &mir_foo_do_thing,
+ *                       target,
+ *                       arg,
+ *                       callback,
+ *                       static_cast<void*>(context));
+ *
+ * The last two parameters must be the callback function and the void* to pass to that callback.
+ * The last argument must have a pointer type; NULL or nullptr must be explicitly cast to
+ * void*.
+ */
+template<typename Callable, typename... Args>
+void make_synchronous_call(MirConnection* connection,
+                           Callable&& function,
+                           Args&&... args)
+{
+    static_assert(
+        std::is_same<typename std::result_of<Callable(Args...)>::type, MirWaitHandle*>::value,
+        "Second parameter must be a function that returns a MirWaitHandle*");
+
+    if (connection->watch_fd() != mir::Fd::invalid)
+    {
+        std::tuple<Args...> arguments{args...};
+        constexpr int arg_count = sizeof...(Args);
+        auto callback = std::get<arg_count - 2>(arguments);
+        auto context = std::get<arg_count - 1>(arguments);
+
+        static_assert(
+            std::is_pointer<typename std::tuple_element<arg_count - 1, std::tuple<Args...>>::type>::value,
+            "The final argument must be a pointer type");
+        static_assert(
+            std::is_pointer<typename std::tuple_element<arg_count - 2, std::tuple<Args...>>::type>::value,
+            "The second last argument must be a function pointer");
+
+        SynchronousContext<decltype(callback)> wrapper_context {
+            callback,
+            false,
+            context
+        };
+
+        std::get<arg_count - 2>(arguments) = &synchronous_wrapper;
+        std::get<arg_count - 1>(arguments) =
+            reinterpret_cast<typename std::tuple_element<arg_count - 1, std::tuple<Args...>>::type>(&wrapper_context);
+
+        connection->process_next_request_first();
+        apply(std::forward<Callable>(function), arguments);
+
+        dispatch_connection_until(connection, [&wrapper_context](){ return wrapper_context.complete; });
+    }
+    else
+    {
+        mir_wait_for(std::forward<Callable>(function)(std::forward<Args>(args)...));
+    }
+}
+
+#endif // MIR_CLIENT_SYNCHRONOUS_H_
 
=== modified file 'src/common/dispatch/CMakeLists.txt'
--- src/common/dispatch/CMakeLists.txt	2015-04-09 06:20:31 +0000
+++ src/common/dispatch/CMakeLists.txt	2015-05-14 07:50:26 +0000
@@ -17,10 +17,9 @@
 list(
   APPEND MIR_COMMON_SOURCES
   ${CMAKE_CURRENT_SOURCE_DIR}/action_queue.cpp
-  ${CMAKE_CURRENT_SOURCE_DIR}/simple_dispatch_thread.cpp
   ${CMAKE_CURRENT_SOURCE_DIR}/multiplexing_dispatchable.cpp
   ${CMAKE_CURRENT_SOURCE_DIR}/utils.cpp
-  ${CMAKE_CURRENT_SOURCE_DIR}/action_queue.cpp
+  ${CMAKE_CURRENT_SOURCE_DIR}/threaded_dispatcher.cpp
 )
 
 set(MIR_COMMON_SOURCES ${MIR_COMMON_SOURCES} PARENT_SCOPE)
 
=== removed file 'src/common/dispatch/simple_dispatch_thread.cpp'
--- src/common/dispatch/simple_dispatch_thread.cpp	2015-04-30 17:41:03 +0000
+++ src/common/dispatch/simple_dispatch_thread.cpp	1970-01-01 00:00:00 +0000
@@ -1,163 +0,0 @@
-/*
- * 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 "mir/logging/logger.h"
-#include "utils.h"
-#include "mir/signal_blocker.h"
-
-#include <sys/epoll.h>
-#include <unistd.h>
-#include <system_error>
-#include <array>
-#include <signal.h>
-#include <boost/exception/all.hpp>
-
-namespace md = mir::dispatch;
-
-namespace
-{
-
-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 = md::fd_event_to_epoll(dispatchee->relevant_events());
-    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, dispatchee->watch_fd(), &event);
-
-    for (;;)
-    {
-        std::array<epoll_event,2> events;
-        auto const num_available_events =
-            epoll_wait(epoll_fd, events.data(), events.size(), -1);
-
-        if (num_available_events == 1)
-        {
-            if (events[0].data.u32 == fd_names::dispatchee_fd)
-            {
-                if (!dispatchee->dispatch(md::epoll_to_fd_event(events[0])))
-                {
-                    // No need to keep looping, the Dispatchable's not going to produce any more events.
-                    return;
-                }
-            }
-            else if (events[0].data.u32 == fd_names::shutdown)
-            {
-                // The only thing we do with the shutdown fd is to close it.
-                return;
-            }
-        }
-        else if (num_available_events > 1)
-        {
-            // Because we only have two fds in the epoll, if there is more than one
-            // event pending then one of them must be a shutdown event.
-            // So, shutdown.
-            return;
-        }
-        else if (num_available_events < 0)
-        {
-            // Although we have blocked signals in this thread, we can still
-            // get interrupted by SIGSTOP (which is unblockable and non-fatal).
-            if (errno != EINTR)
-            {
-                BOOST_THROW_EXCEPTION((std::system_error{
-                    errno, std::system_category(), "Failed to wait for epoll events"}));
-            }
-        }
-    }
-}
-
-}
-
-md::SimpleDispatchThread::SimpleDispatchThread(std::shared_ptr<md::Dispatchable> const& dispatchee)
-    : SimpleDispatchThread(dispatchee, []{})
-{}
-
-md::SimpleDispatchThread::SimpleDispatchThread(
-    std::shared_ptr<md::Dispatchable> const& dispatchee,
-    std::function<void()> const& exception_handler)
-{
-    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]};
-    {
-        // The newly spawned thread inherits the current signal mask; block everything
-        // before creating the new thread so that there's no race between thread start
-        // and signal blocking.
-        mir::SignalBlocker block_signals;
-        eventloop = std::thread{
-            [exception_handler, dispatchee, terminate_fd]()
-            {
-                try
-                {
-                    wait_for_events_forever(dispatchee, terminate_fd);
-                }
-                catch(...)
-                {
-                    exception_handler();
-                }
-            }};
-    }
-}
-
-md::SimpleDispatchThread::~SimpleDispatchThread() noexcept
-{
-    shutdown_fd = mir::Fd{};
-    if (eventloop.get_id() == std::this_thread::get_id())
-    {
-        // We're being destroyed from within the dispatch callback
-        // Attempting to join the eventloop will result in a trivial deadlock.
-        // 
-        // The std::thread destructor will call std::terminate() for us, let's
-        // leave a useful message.
-        mir::logging::log(mir::logging::Severity::critical,
-                          "Destroying SimpleDispatchThread from within a dispatch callback. This is a programming error.",
-                          "Dispatch");
-    }
-    else if (eventloop.joinable())
-    {
-        eventloop.join();
-    }
-}
 
=== added file 'src/common/dispatch/threaded_dispatcher.cpp'
--- src/common/dispatch/threaded_dispatcher.cpp	1970-01-01 00:00:00 +0000
+++ src/common/dispatch/threaded_dispatcher.cpp	2015-05-14 07:50:26 +0000
@@ -0,0 +1,294 @@
+/*
+ * 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/threaded_dispatcher.h"
+#include "mir/dispatch/dispatchable.h"
+#include "mir/thread_name.h"
+
+#include "mir/raii.h"
+#include "mir/logging/logger.h"
+
+#include <fcntl.h>
+#include <poll.h>
+#include <unistd.h>
+#include <system_error>
+#include <signal.h>
+#include <boost/exception/all.hpp>
+#include <algorithm>
+#include <unordered_map>
+#include <sys/eventfd.h>
+
+namespace md = mir::dispatch;
+
+class md::ThreadedDispatcher::ThreadShutdownRequestHandler : public md::Dispatchable
+{
+public:
+    ThreadShutdownRequestHandler()
+        : event_semaphore{eventfd(0, EFD_CLOEXEC | EFD_SEMAPHORE)},
+          shutting_down{false}
+    {
+        if (event_semaphore == mir::Fd::invalid)
+        {
+            BOOST_THROW_EXCEPTION((std::system_error{errno,
+                                                     std::system_category(),
+                                                     "Failed to create shutdown eventfd"}));
+        }
+    }
+
+    mir::Fd watch_fd() const override
+    {
+        return event_semaphore;
+    }
+
+    bool dispatch(md::FdEvents events) override
+    {
+        if (events & md::FdEvent::error)
+        {
+            return false;
+        }
+
+        eventfd_t dummy;
+        if (eventfd_read(event_semaphore, &dummy) < 0)
+        {
+            BOOST_THROW_EXCEPTION((std::system_error{errno,
+                                                     std::system_category(),
+                                                     "Failed to clear shutdown notification"}));
+        }
+        std::lock_guard<decltype(running_flag_guard)> lock{running_flag_guard};
+        *running_flags.at(std::this_thread::get_id()) = false;
+
+        return true;
+    }
+
+    md::FdEvents relevant_events() const override
+    {
+        return md::FdEvent::readable;
+    }
+
+    std::thread::id terminate_one_thread()
+    {
+        // First we tell a thread to die, any thread...
+        if (eventfd_write(event_semaphore, 1) < 0)
+        {
+            BOOST_THROW_EXCEPTION((std::system_error{errno,
+                                                     std::system_category(),
+                                                     "Failed to trigger thread shutdown"}));
+        }
+
+        // ...now we wait for a thread to die and tell us its ID...
+        // We wait for a surprisingly long time because our threads are potentially blocked
+        // in client code that we don't control.
+        //
+        // If the client is entirely unresponsive for a whole minute, it deserves to die.
+        std::unique_lock<decltype(terminating_thread_mutex)> lock{terminating_thread_mutex};
+        if (!thread_terminating.wait_for (lock,
+                                          std::chrono::seconds{60},
+                                          [this]() { return !terminating_threads.empty(); }))
+        {
+            BOOST_THROW_EXCEPTION((std::runtime_error{"Thread failed to shutdown"}));
+        }
+
+        auto killed_thread_id = terminating_threads.back();
+        terminating_threads.pop_back();
+        return killed_thread_id;
+    }
+
+    void terminate_all_threads_async()
+    {
+        eventfd_t thread_count;
+        {
+            std::lock_guard<std::mutex> lock(running_flag_guard);
+            thread_count = running_flags.size();
+            shutting_down = true;
+        }
+        if (eventfd_write(event_semaphore, thread_count) < 0)
+        {
+            BOOST_THROW_EXCEPTION((std::system_error{errno,
+                                                     std::system_category(),
+                                                     "Failed to trigger thread shutdown"}));
+        }
+    }
+
+    void register_thread(bool& run_flag)
+    {
+        std::lock_guard<decltype(running_flag_guard)> lock{running_flag_guard};
+
+        running_flags[std::this_thread::get_id()] = &run_flag;
+        if (shutting_down)
+        {
+            if (eventfd_write(event_semaphore, 1) < 0)
+            {
+                BOOST_THROW_EXCEPTION((std::system_error{errno,
+                                                         std::system_category(),
+                                                         "Failed to trigger thread shutdown"}));
+            }
+        }
+    }
+
+    void unregister_thread()
+    {
+        {
+            std::lock_guard<decltype(terminating_thread_mutex)> lock{terminating_thread_mutex};
+            terminating_threads.push_back(std::this_thread::get_id());
+        }
+        thread_terminating.notify_one();
+        {
+            std::lock_guard<decltype(running_flag_guard)> lock{running_flag_guard};
+
+            if (running_flags.erase(std::this_thread::get_id()) != 1)
+            {
+                BOOST_THROW_EXCEPTION((std::logic_error{"Attempted to unregister a not-registered thread"}));
+            }
+        }
+    }
+
+private:
+    mir::Fd event_semaphore;
+
+    std::mutex terminating_thread_mutex;
+    std::condition_variable thread_terminating;
+    std::vector<std::thread::id> terminating_threads;
+
+    std::mutex running_flag_guard;
+    std::unordered_map<std::thread::id, bool*> running_flags;
+    bool shutting_down;
+};
+
+md::ThreadedDispatcher::ThreadedDispatcher(std::string const& name, std::shared_ptr<Dispatchable> const& dispatchee)
+    : ThreadedDispatcher(name, dispatchee, [](){ throw; })
+{
+}
+
+md::ThreadedDispatcher::ThreadedDispatcher(std::string const& name,
+                                           std::shared_ptr<md::Dispatchable> const& dispatchee,
+                                           std::function<void()> const& exception_handler)
+    : name_base{name},
+      thread_exiter{std::make_shared<ThreadShutdownRequestHandler>()},
+      dispatcher{std::make_shared<MultiplexingDispatchable>()},
+      exception_handler{exception_handler}
+{
+
+    // We rely on exactly one thread at a time getting a shutdown message
+    dispatcher->add_watch(thread_exiter, md::DispatchReentrancy::sequential);
+
+    // But our target dispatchable is welcome to be dispatched on as many threads
+    // as desired.
+    dispatcher->add_watch(dispatchee, md::DispatchReentrancy::reentrant);
+
+    threadpool.emplace_back(&dispatch_loop, name_base, thread_exiter, dispatcher, exception_handler);
+}
+
+md::ThreadedDispatcher::~ThreadedDispatcher() noexcept
+{
+    std::lock_guard<decltype(thread_pool_mutex)> lock{thread_pool_mutex};
+
+    thread_exiter->terminate_all_threads_async();
+
+    for (auto& thread : threadpool)
+    {
+        if (thread.get_id() == std::this_thread::get_id())
+        {
+            // We're being destroyed from within the dispatch callback
+            // Attempting to join the eventloop will result in a trivial deadlock.
+            //
+            // The std::thread destructor will call std::terminate() for us, let's
+            // leave a useful message.
+            mir::logging::log(mir::logging::Severity::critical,
+                              "Destroying ThreadedDispatcher from within a dispatch callback. This is a programming error.",
+                              "Dispatch");
+        }
+        else
+        {
+            thread.join();
+        }
+    }
+}
+
+void md::ThreadedDispatcher::add_thread()
+{
+    std::lock_guard<decltype(thread_pool_mutex)> lock{thread_pool_mutex};
+    threadpool.emplace_back(&dispatch_loop, name_base, thread_exiter, dispatcher, exception_handler);
+}
+
+void md::ThreadedDispatcher::remove_thread()
+{
+    auto terminated_thread_id = thread_exiter->terminate_one_thread();
+
+    // Find that thread in our vector, join() it, then remove it.
+    std::lock_guard<decltype(thread_pool_mutex)> threadpool_lock{thread_pool_mutex};
+
+    auto dying_thread = std::find_if(threadpool.begin(),
+                                     threadpool.end(),
+                                     [this, &terminated_thread_id](std::thread const& candidate)
+    {
+            return candidate.get_id() == terminated_thread_id;
+    });
+    dying_thread->join();
+    threadpool.erase(dying_thread);
+}
+
+void md::ThreadedDispatcher::dispatch_loop(std::string const& name,
+                                           std::shared_ptr<ThreadShutdownRequestHandler> thread_register,
+                                           std::shared_ptr<Dispatchable> dispatcher,
+                                           std::function<void()> const& exception_handler)
+{
+    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"}));
+
+    mir::set_thread_name(name);
+
+    // This does not have to be std::atomic<bool> because thread_register is guaranteed to
+    // only ever be dispatch()ed from one thread at a time.
+    bool running{true};
+
+    auto thread_registrar = mir::raii::paired_calls(
+    [&running, thread_register]()
+    {
+        thread_register->register_thread(running);
+    },
+    [thread_register]()
+    {
+        thread_register->unregister_thread();
+    });
+
+    try
+    {
+        struct pollfd waiter;
+        waiter.fd = dispatcher->watch_fd();
+        waiter.events = POLL_IN;
+        while (running)
+        {
+            if (poll(&waiter, 1, -1) < 0)
+            {
+                BOOST_THROW_EXCEPTION((std::system_error{errno,
+                                                         std::system_category(),
+                                                         "Failed to wait for event"}));
+            }
+            dispatcher->dispatch(md::FdEvent::readable);
+        }
+    }
+    catch(...)
+    {
+        exception_handler();
+    }
+}
 
=== modified file 'src/common/symbols.map'
--- src/common/symbols.map	2015-04-15 05:07:08 +0000
+++ src/common/symbols.map	2015-05-14 07:50:26 +0000
@@ -24,8 +24,10 @@
     mir::dispatch::MultiplexingDispatchable::relevant_events*;
     mir::dispatch::MultiplexingDispatchable::remove_watch*;
     mir::dispatch::MultiplexingDispatchable::watch_fd*;
-    mir::dispatch::SimpleDispatchThread::?SimpleDispatchThread*;
-    mir::dispatch::SimpleDispatchThread::SimpleDispatchThread*;
+    mir::dispatch::ThreadedDispatcher::?ThreadedDispatcher*;
+    mir::dispatch::ThreadedDispatcher::ThreadedDispatcher*;
+    mir::dispatch::ThreadedDispatcher::add_thread*;
+    mir::dispatch::ThreadedDispatcher::remove_thread*;
     mir::Fd::Fd*;
     mir::Fd::invalid*;
     mir::Fd::operator*;
@@ -73,7 +75,7 @@
     typeinfo?for?mir::dispatch::ActionQueue;
     typeinfo?for?mir::dispatch::Dispatchable;
     typeinfo?for?mir::dispatch::MultiplexingDispatchable;
-    typeinfo?for?mir::dispatch::SimpleDispatchThread;
+    typeinfo?for?mir::dispatch::ThreadedDispatcher;
     typeinfo?for?mir::Fd;
     typeinfo?for?mir::geometry::Displacement;
     typeinfo?for?mir::geometry::Length;
@@ -88,7 +90,7 @@
     vtable?for?mir::dispatch::ActionQueue;
     vtable?for?mir::dispatch::Dispatchable;
     vtable?for?mir::dispatch::MultiplexingDispatchable;
-    vtable?for?mir::dispatch::SimpleDispatchThread;
+    vtable?for?mir::dispatch::ThreadedDispatcher;
     vtable?for?mir::Fd;
     vtable?for?mir::geometry::Displacement;
     vtable?for?mir::geometry::Length;
 
=== modified file 'src/server/input/default_input_manager.cpp'
--- src/server/input/default_input_manager.cpp	2015-05-05 13:46:29 +0000
+++ src/server/input/default_input_manager.cpp	2015-05-14 07:50:26 +0000
@@ -22,7 +22,7 @@
 #include "mir/input/platform.h"
 #include "mir/dispatch/action_queue.h"
 #include "mir/dispatch/multiplexing_dispatchable.h"
-#include "mir/dispatch/simple_dispatch_thread.h"
+#include "mir/dispatch/threaded_dispatcher.h"
 
 #include "mir/main_loop.h"
 #include "mir/thread_name.h"
@@ -86,7 +86,6 @@
 
     queue->enqueue([this,weak_started_promise]()
                    {
-                        mir::set_thread_name("Mir/InputReader");
                         for (auto const& platform : platforms)
                         {
                             platform->start();
@@ -100,7 +99,8 @@
                         started_promise->set_value();
                    });
 
-    input_thread = std::make_unique<dispatch::SimpleDispatchThread>(
+    input_thread = std::make_unique<dispatch::ThreadedDispatcher>(
+        "Mir/InputReader",
         multiplexer,
         [this,weak_started_promise]()
         {
 
=== modified file 'src/server/input/default_input_manager.h'
--- src/server/input/default_input_manager.h	2015-05-01 14:47:55 +0000
+++ src/server/input/default_input_manager.h	2015-05-14 07:50:26 +0000
@@ -37,7 +37,7 @@
 namespace dispatch
 {
 class MultiplexingDispatchable;
-class SimpleDispatchThread;
+class ThreadedDispatcher;
 class ActionQueue;
 }
 namespace input
@@ -65,7 +65,7 @@
     std::shared_ptr<dispatch::MultiplexingDispatchable> const multiplexer;
     std::shared_ptr<input::android::InputReaderDispatchable> const legacy_dispatchable;
     std::shared_ptr<dispatch::ActionQueue> const queue;
-    std::unique_ptr<dispatch::SimpleDispatchThread> input_thread;
+    std::unique_ptr<dispatch::ThreadedDispatcher> input_thread;
 
     enum class State
     {
 
=== modified file 'tests/acceptance-tests/CMakeLists.txt'
--- tests/acceptance-tests/CMakeLists.txt	2015-05-05 06:46:30 +0000
+++ tests/acceptance-tests/CMakeLists.txt	2015-05-14 07:50:26 +0000
@@ -2,6 +2,7 @@
 
 include_directories(
   ${CMAKE_SOURCE_DIR}
+  ${UMOCKDEV_INCLUDE_DIRS}
 )
 
 set(
@@ -84,5 +85,5 @@
   OFF)
 
 if (MIR_RUN_ACCEPTANCE_TESTS)
-  mir_discover_tests(mir_acceptance_tests)
+  mir_discover_tests(mir_acceptance_tests LD_PRELOAD=libumockdev-preload.so.0 G_SLICE=always-malloc G_DEBUG=gc-friendly)
 endif (MIR_RUN_ACCEPTANCE_TESTS)
 
=== modified file 'tests/acceptance-tests/test_client_library.cpp'
--- tests/acceptance-tests/test_client_library.cpp	2015-04-30 12:27:33 +0000
+++ tests/acceptance-tests/test_client_library.cpp	2015-05-14 07:50:26 +0000
@@ -24,6 +24,10 @@
 #include "mir_test_framework/using_stub_client_platform.h"
 #include "mir_test_framework/any_surface.h"
 #include "mir_test/validity_matchers.h"
+#include "mir_test/fd_utils.h"
+#include "mir_test_framework/udev_environment.h"
+#include "mir_test/signal.h"
+#include "mir_test/auto_unblock_thread.h"
 
 #include "src/include/client/mir/client_buffer.h"
 
@@ -37,6 +41,8 @@
 #include <system/window.h>  // for ANativeWindowBuffer AKA MirNativeBuffer
 #endif
 
+#include <boost/throw_exception.hpp>
+
 #include <gtest/gtest.h>
 #include <gmock/gmock.h>
 #include <chrono>
@@ -53,6 +59,8 @@
 namespace mc = mir::compositor;
 namespace mcl = mir::client;
 namespace mtf = mir_test_framework;
+namespace mt = mir::test;
+
 namespace
 {
 struct ClientLibrary : mtf::HeadlessInProcessServer
@@ -61,6 +69,12 @@
     MirConnection* connection = nullptr;
     MirSurface* surface  = nullptr;
     int buffers = 0;
+    mtf::UdevEnvironment mock_devices;
+
+    ClientLibrary()
+    {
+        mock_devices.add_standard_device("laptop-keyboard");
+    }
 
     static void connection_callback(MirConnection* connection, void* context)
     {
@@ -275,111 +289,6 @@
     mir_connection_release(connection);
 }
 
-TEST_F(ClientLibrary, can_set_surface_min_width)
-{
-    connection = mir_connect_sync(new_connection().c_str(), __PRETTY_FUNCTION__);
-
-    int const width = 640;
-    int const height = 480;
-    auto const format = mir_pixel_format_abgr_8888;
-    auto const spec =
-        mir_connection_create_spec_for_normal_surface(connection, width, height, format);
-
-    int const min_width = 480;
-    EXPECT_TRUE(mir_surface_spec_set_min_width(spec, min_width));
-
-    surface = mir_surface_create_sync(spec);
-    mir_surface_spec_release(spec);
-    mir_surface_release_sync(surface);
-    mir_connection_release(connection);
-}
-
-TEST_F(ClientLibrary, can_set_surface_min_height)
-{
-    connection = mir_connect_sync(new_connection().c_str(), __PRETTY_FUNCTION__);
-
-    int const width = 640;
-    int const height = 480;
-    auto const format = mir_pixel_format_abgr_8888;
-    auto const spec =
-        mir_connection_create_spec_for_normal_surface(connection, width, height, format);
-
-    int const min_height = 480;
-    EXPECT_TRUE(mir_surface_spec_set_min_height(spec, min_height));
-
-    surface = mir_surface_create_sync(spec);
-    mir_surface_spec_release(spec);
-    mir_surface_release_sync(surface);
-    mir_connection_release(connection);
-}
-
-TEST_F(ClientLibrary, can_set_surface_max_width)
-{
-    connection = mir_connect_sync(new_connection().c_str(), __PRETTY_FUNCTION__);
-
-    int const width = 640;
-    int const height = 480;
-    auto const format = mir_pixel_format_abgr_8888;
-    auto const spec =
-        mir_connection_create_spec_for_normal_surface(connection, width, height, format);
-
-    int const max_width = 1024;
-    EXPECT_TRUE(mir_surface_spec_set_max_width(spec, max_width));
-
-    surface = mir_surface_create_sync(spec);
-    mir_surface_spec_release(spec);
-    mir_surface_release_sync(surface);
-    mir_connection_release(connection);
-}
-
-TEST_F(ClientLibrary, can_set_surface_max_height)
-{
-    connection = mir_connect_sync(new_connection().c_str(), __PRETTY_FUNCTION__);
-
-    int const width = 640;
-    int const height = 480;
-    auto const format = mir_pixel_format_abgr_8888;
-    auto const spec =
-        mir_connection_create_spec_for_normal_surface(connection, width, height, format);
-
-    int const max_height = 1024;
-    EXPECT_TRUE(mir_surface_spec_set_max_height(spec, max_height));
-
-    surface = mir_surface_create_sync(spec);
-    mir_surface_spec_release(spec);
-    mir_surface_release_sync(surface);
-    mir_connection_release(connection);
-}
-
-TEST_F(ClientLibrary, min_size_respected_when_placing_surface)
-{
-    connection = mir_connect_sync(new_connection().c_str(), __PRETTY_FUNCTION__);
-
-    int const width = 6400;
-    int const height = 4800;
-    auto const format = mir_pixel_format_abgr_8888;
-    auto const spec =
-        mir_connection_create_spec_for_normal_surface(connection, width, height, format);
-
-    int const min_width = 4800;
-    int const min_height = 3200;
-
-    mir_surface_spec_set_min_width(spec, min_width);
-    mir_surface_spec_set_min_height(spec, min_height);
-    surface = mir_surface_create_sync(spec);
-    mir_surface_spec_release(spec);
-
-    auto const buffer_stream = mir_surface_get_buffer_stream(surface);
-
-    MirGraphicsRegion graphics_region;
-    mir_buffer_stream_get_graphics_region(buffer_stream, &graphics_region);
-    EXPECT_THAT(graphics_region.width, Ge(min_width));
-    EXPECT_THAT(graphics_region.height, Ge(min_height));
-
-    mir_surface_release_sync(surface);
-    mir_connection_release(connection);
-}
-
 TEST_F(ClientLibrary, receives_surface_dpi_value)
 {
     connection = mir_connect_sync(new_connection().c_str(), __PRETTY_FUNCTION__);
@@ -930,3 +839,320 @@
     mir_surface_release_sync(surface);
     mir_connection_release(connection);
 }
+
+namespace
+{
+struct ThreadTrackingCallbacks
+{
+    ThreadTrackingCallbacks()
+        : client_thread{pthread_self()}
+    {
+    }
+
+    static void connection_ready(MirConnection* /*connection*/, void* ctx)
+    {
+        auto data = reinterpret_cast<ThreadTrackingCallbacks*>(ctx);
+        EXPECT_EQ(pthread_self(), data->client_thread);
+        data->connection_ready_called.raise();
+    }
+
+    static void event_delegate(MirSurface* /*surf*/, MirEvent const* event, void* ctx)
+    {
+        auto data = reinterpret_cast<ThreadTrackingCallbacks*>(ctx);
+
+        EXPECT_THAT(pthread_self(), Eq(data->client_thread));
+        data->event_received.raise();
+        if (mir_event_get_type(event) == mir_event_type_input)
+        {
+            data->input_event_received.raise();
+        }
+    }
+
+    static void surface_created(MirSurface* surf, void* ctx)
+    {
+        auto data = reinterpret_cast<ThreadTrackingCallbacks*>(ctx);
+        EXPECT_THAT(pthread_self(), Eq(data->client_thread));
+        data->surf = surf;
+
+        mir_surface_set_event_handler(data->surf, &ThreadTrackingCallbacks::event_delegate, data);
+    }
+
+    static void swap_buffers_complete(MirBufferStream* /*stream*/, void* ctx)
+    {
+        auto data = reinterpret_cast<ThreadTrackingCallbacks*>(ctx);
+        EXPECT_EQ(pthread_self(), data->client_thread);
+        data->buffers_swapped.raise();
+    }
+
+    pthread_t const client_thread;
+    MirSurface* surf{nullptr};
+    mt::Signal buffers_swapped;
+    mt::Signal connection_ready_called;
+    mt::Signal event_received;
+    mt::Signal input_event_received;
+};
+
+void pump_eventloop_until(MirConnection* connection, std::function<bool()> predicate, std::chrono::steady_clock::time_point timeout)
+{
+    using namespace std::literals::chrono_literals;
+
+    auto fd = mir::Fd{mir::IntOwnedFd{mir_connection_get_event_fd(connection)}};
+
+    while (!predicate() && (std::chrono::steady_clock::now() < timeout))
+    {
+        if (mt::fd_becomes_readable(fd, 10ms))
+        {
+            mir_connection_dispatch(connection);
+        }
+    }
+    if (!predicate())
+    {
+        BOOST_THROW_EXCEPTION((std::runtime_error{"Timeout waiting for state change"}));
+    }
+}
+}
+
+TEST_F(ClientLibrary, manual_dispatch_handles_callbacks_in_parent_thread)
+{
+    using namespace std::literals::chrono_literals;
+
+    auto const test_timeout = std::chrono::steady_clock::now() + 10min;
+
+    ThreadTrackingCallbacks data;
+
+    auto connection = mir_connect_with_manual_dispatch(new_connection().c_str(), __PRETTY_FUNCTION__, &ThreadTrackingCallbacks::connection_ready, &data);
+
+    ASSERT_THAT(connection, Ne(nullptr));
+
+    pump_eventloop_until(
+        connection,
+        [&data]()
+    {
+        return data.connection_ready_called.raised();
+    },
+        test_timeout);
+
+    ASSERT_THAT(connection, IsValid());
+
+    auto surface_spec = mir_connection_create_spec_for_normal_surface(connection,
+                                                                      233, 355,
+                                                                      mir_pixel_format_argb_8888);
+    auto surf_wh = mir_surface_create(surface_spec,
+                                      &ThreadTrackingCallbacks::surface_created,
+                                      &data);
+    mir_surface_spec_release(surface_spec);
+
+    pump_eventloop_until(
+        connection,
+        [surf_wh]()
+    {
+        return mir_wait_handle_ready(surf_wh);
+    },
+        test_timeout);
+
+    EXPECT_THAT(data.surf, IsValid());
+
+    auto buffer_stream = mir_surface_get_buffer_stream(data.surf);
+    auto swap_wh = mir_buffer_stream_swap_buffers(buffer_stream, ThreadTrackingCallbacks::swap_buffers_complete, &data);
+
+    pump_eventloop_until(
+        connection,
+        [swap_wh]()
+    {
+        return mir_wait_handle_ready(swap_wh);
+    },
+        test_timeout);
+    EXPECT_TRUE(data.buffers_swapped.raised());
+
+    mir_surface_release_sync(data.surf);
+    mir_connection_release(connection);
+}
+
+TEST_F(ClientLibrary, manual_dispatch_handles_events_in_parent_thread)
+{
+    using namespace testing;
+    using namespace std::literals::chrono_literals;
+
+    auto const test_timeout = std::chrono::steady_clock::now() + 10min;
+
+    ThreadTrackingCallbacks data;
+
+    connection = mir_connect_with_manual_dispatch(new_connection().c_str(), __PRETTY_FUNCTION__, &ThreadTrackingCallbacks::connection_ready, &data);
+
+    ASSERT_THAT(connection, Ne(nullptr));
+
+    pump_eventloop_until(
+        connection,
+        [&data]()
+    {
+        return data.connection_ready_called.raised();
+    },
+        test_timeout);
+
+    ASSERT_THAT(connection, IsValid());
+
+    auto surface_spec = mir_connection_create_spec_for_normal_surface(connection,
+                                                                      233, 355,
+                                                                      mir_pixel_format_argb_8888);
+    auto surf_wh = mir_surface_create(surface_spec,
+                                      &ThreadTrackingCallbacks::surface_created,
+                                      &data);
+    mir_surface_spec_release(surface_spec);
+
+
+    pump_eventloop_until(
+        connection,
+        [surf_wh]()
+    {
+        return mir_wait_handle_ready(surf_wh);
+    },
+        test_timeout);
+
+    EXPECT_THAT(data.surf, IsValid());
+
+    // We need to swap buffers so that the surface is fully realised and
+    // will be a valid focus target.
+    //
+    // The shell will not focus a surface with no content.
+    auto buffer_stream = mir_surface_get_buffer_stream(data.surf);
+    mir_buffer_stream_swap_buffers_sync(buffer_stream);
+
+    mir_surface_set_state(data.surf, mir_surface_state_fullscreen);
+
+    pump_eventloop_until(
+        connection,
+        [&data]()
+    {
+        return data.event_received.raised();
+    },
+        test_timeout);
+
+    ASSERT_THAT(mir_surface_get_focus(data.surf), Eq(mir_surface_focused));
+
+    mock_devices.load_device_evemu("laptop-keyboard-hello");
+
+    pump_eventloop_until(
+        connection,
+        [&data]()
+    {
+        return data.input_event_received.raised();
+    },
+        test_timeout);
+
+    mir_surface_release_sync(data.surf);
+    mir_connection_release(connection);
+}
+
+namespace
+{
+struct SignalPair
+{
+    mir::test::Signal now_blocking;
+    mir::test::Signal event_received;
+};
+
+void notifying_event_handler(MirSurface*, MirEvent const* ev, void* ctx)
+{
+    auto signal_pair = *reinterpret_cast<std::shared_ptr<SignalPair>*>(ctx);
+    // We trigger an input event once we've noticed the surface callback is blocking
+    // so we need to only raise the flag on an input event; otherwise we may spuriously
+    // fail if we receive a surface event (like the focus event) before we hit
+    // the wait in blocking_surface_callback()
+    if (mir_event_get_type(ev) == mir_event_type_input)
+    {
+        signal_pair->event_received.raise();
+    }
+}
+
+void blocking_buffer_stream_callback(MirBufferStream*, void* ctx)
+{
+    auto signal_pair = *reinterpret_cast<std::shared_ptr<SignalPair>*>(ctx);
+    signal_pair->now_blocking.raise();
+    EXPECT_TRUE(signal_pair->event_received.wait_for(std::chrono::seconds{5}));
+}
+}
+
+TEST_F(ClientLibrary, rpc_blocking_doesnt_block_event_delivery_with_auto_dispatch)
+{
+    using namespace testing;
+
+    connection = mir_connect_sync(new_connection().c_str(), __PRETTY_FUNCTION__);
+
+    ASSERT_THAT(connection, IsValid());
+
+    auto surface_spec = mir_connection_create_spec_for_normal_surface(connection,
+                                                                      233, 355,
+                                                                      mir_pixel_format_argb_8888);
+
+    auto surf = mir_surface_create_sync(surface_spec);
+    mir_surface_spec_release(surface_spec);
+
+    EXPECT_THAT(surf, IsValid());
+
+    auto signal_pair = std::make_shared<SignalPair>();
+    mir_surface_set_event_handler(surf, &notifying_event_handler, &signal_pair);
+
+    auto buffer_stream = mir_surface_get_buffer_stream(surf);
+    auto wh = mir_buffer_stream_swap_buffers(buffer_stream, &blocking_buffer_stream_callback, &signal_pair);
+
+    EXPECT_TRUE(signal_pair->now_blocking.wait_for(std::chrono::seconds{5}));
+    EXPECT_FALSE(signal_pair->event_received.raised());
+
+    mock_devices.load_device_evemu("laptop-keyboard-hello");
+
+    EXPECT_TRUE(signal_pair->event_received.wait_for(std::chrono::seconds{5}));
+
+    mir_wait_for(wh);
+    mir_surface_release_sync(surf);
+    mir_connection_release(connection);
+}
+
+namespace
+{
+void async_release_completed(MirSurface*, void* ctx)
+{
+    auto called = reinterpret_cast<bool*>(ctx);
+    *called = true;
+}
+
+void async_creation_completed(MirSurface* surf, void* ctx)
+{
+    mir_surface_release(surf, &async_release_completed, ctx);
+}
+}
+
+TEST_F(ClientLibrary, sync_call_completes_before_previous_undispatched_call)
+{
+    using namespace std::literals::chrono_literals;
+    using namespace testing;
+
+    auto timeout = std::chrono::steady_clock::now() + 60s;
+
+    ThreadTrackingCallbacks data;
+
+    auto connection = mir_connect_with_manual_dispatch(new_connection().c_str(), __PRETTY_FUNCTION__, &ThreadTrackingCallbacks::connection_ready, &data);
+    ASSERT_THAT(connection, Ne(nullptr));
+
+    pump_eventloop_until(connection, [connection]() { return mir_connection_is_valid(connection); }, timeout);
+
+    bool async_call_completed{false};
+
+    auto surface_spec = mir_connection_create_spec_for_normal_surface(connection,
+                                                                      233, 355,
+                                                                      mir_pixel_format_argb_8888);
+    mir_surface_create(surface_spec, &async_creation_completed, &async_call_completed);
+
+    EXPECT_FALSE(async_call_completed);
+    auto surf = mir_surface_create_sync(surface_spec);
+    mir_surface_spec_release(surface_spec);
+
+    EXPECT_THAT(surf, IsValid());
+    EXPECT_FALSE(async_call_completed);
+
+    mir_surface_release_sync(surf);
+    EXPECT_FALSE(async_call_completed);
+
+    pump_eventloop_until(connection, [&async_call_completed]() { return async_call_completed;}, timeout);
+
+    mir_connection_release(connection);
+}
 
=== modified file 'tests/include/mir_test/test_protobuf_client.h'
--- tests/include/mir_test/test_protobuf_client.h	2015-03-31 02:35:42 +0000
+++ tests/include/mir_test/test_protobuf_client.h	2015-05-14 07:50:26 +0000
@@ -32,7 +32,7 @@
 {
 namespace dispatch
 {
-class SimpleDispatchThread;
+class ThreadedDispatcher;
 }
 namespace test
 {
@@ -46,7 +46,7 @@
 
     std::shared_ptr<doubles::MockRpcReport> rpc_report;
     std::shared_ptr<google::protobuf::RpcChannel> channel;
-    std::shared_ptr<dispatch::SimpleDispatchThread> eventloop;
+    std::shared_ptr<dispatch::ThreadedDispatcher> eventloop;
     mir::protobuf::DisplayServer::Stub display_server;
     mir::protobuf::ConnectParameters connect_parameters;
     mir::protobuf::SurfaceParameters surface_parameters;
 
=== modified file 'tests/include/mir_test_doubles/mock_client_buffer_stream.h'
--- tests/include/mir_test_doubles/mock_client_buffer_stream.h	2015-04-01 20:00:19 +0000
+++ tests/include/mir_test_doubles/mock_client_buffer_stream.h	2015-05-14 07:50:26 +0000
@@ -45,6 +45,7 @@
     MOCK_METHOD0(get_create_wait_handle, MirWaitHandle*(void));
     MOCK_CONST_METHOD0(rpc_id, frontend::BufferStreamId(void));
     MOCK_CONST_METHOD0(valid, bool(void));
+    MOCK_CONST_METHOD0(get_connection, MirConnection*());
 };
 
 }
 
=== modified file 'tests/include/mir_test_doubles/mock_client_buffer_stream_factory.h'
--- tests/include/mir_test_doubles/mock_client_buffer_stream_factory.h	2015-04-21 17:22:09 +0000
+++ tests/include/mir_test_doubles/mock_client_buffer_stream_factory.h	2015-05-14 07:50:26 +0000
@@ -34,12 +34,22 @@
 
 struct MockClientBufferStreamFactory : public client::ClientBufferStreamFactory
 {
-    MOCK_METHOD3(make_consumer_stream, std::shared_ptr<client::ClientBufferStream>(protobuf::DisplayServer&,
-        protobuf::BufferStream const&, std::string const&));
-    MOCK_METHOD3(make_producer_stream, std::shared_ptr<client::ClientBufferStream>(protobuf::DisplayServer&,
-        protobuf::BufferStream const&, std::string const&));
-    MOCK_METHOD4(make_producer_stream, client::ClientBufferStream*(protobuf::DisplayServer&,
-        protobuf::BufferStreamParameters const&, mir_buffer_stream_callback callback, void* context));
+    MOCK_METHOD4(make_consumer_stream, std::shared_ptr<client::ClientBufferStream>(
+        MirConnection*,
+        protobuf::DisplayServer&,
+        protobuf::BufferStream const&,
+        std::string const&));
+    MOCK_METHOD4(make_producer_stream, std::shared_ptr<client::ClientBufferStream>(
+        MirConnection*,
+        protobuf::DisplayServer&,
+        protobuf::BufferStream const&,
+        std::string const&));
+    MOCK_METHOD5(make_producer_stream, client::ClientBufferStream*(
+        MirConnection*,
+        protobuf::DisplayServer&,
+        protobuf::BufferStreamParameters const&,
+        mir_buffer_stream_callback callback,
+        void* context));
 };
 
 }
 
=== modified file 'tests/include/mir_test_doubles/stub_client_buffer_stream_factory.h'
--- tests/include/mir_test_doubles/stub_client_buffer_stream_factory.h	2015-04-01 20:00:19 +0000
+++ tests/include/mir_test_doubles/stub_client_buffer_stream_factory.h	2015-05-14 07:50:26 +0000
@@ -32,6 +32,7 @@
 struct StubClientBufferStreamFactory : public client::ClientBufferStreamFactory
 {
     std::shared_ptr<client::ClientBufferStream> make_consumer_stream(
+        MirConnection* /* allocating_connection */,
         protobuf::DisplayServer& /* server */,
         protobuf::BufferStream const& /* protobuf_bs */,
         std::string const& /* surface_name */) override
@@ -40,6 +41,7 @@
     }
 
     std::shared_ptr<client::ClientBufferStream> make_producer_stream(
+        MirConnection* /* allocating_connection */,
         protobuf::DisplayServer& /* server */,
         protobuf::BufferStream const& /* protobuf_bs */,
         std::string const& /* surface_name */) override
@@ -47,7 +49,9 @@
         return nullptr;
     }
 
-    client::ClientBufferStream* make_producer_stream(protobuf::DisplayServer& /* server */,
+    client::ClientBufferStream* make_producer_stream(
+       MirConnection* /* allocating_connection */,
+       protobuf::DisplayServer& /* server */,
        protobuf::BufferStreamParameters const& /* params */,
        mir_buffer_stream_callback /* callback */, void* /* context */) override
     {
 
=== modified file 'tests/include/mir_test_framework/udev_environment.h'
--- tests/include/mir_test_framework/udev_environment.h	2015-01-21 07:34:50 +0000
+++ tests/include/mir_test_framework/udev_environment.h	2015-05-14 07:50:26 +0000
@@ -53,6 +53,32 @@
      */
     void add_standard_device(std::string const& name);
 
+    /**
+     * Load an ioctl recording for a UMockdev device
+     *
+     * Looks for a <tt>name</tt>.ioctl file recorded with umockdev-record --ioctl
+     * and adds it to the associated device in the testbed.
+     *
+     * The udev records for the device these ioctl records will be associated with
+     * must already exist in the testbed
+     *
+     * @param name The unadorned filename for the ioctl records to add
+     */
+    void load_device_ioctls(std::string const& name);
+
+    /**
+     * Load an evemu evdev recording for a UMockdev device
+     *
+     * Looks for a <tt>name</tt>.evemu file recorded with umockdev-record --evemu
+     * (or evemu-record) and associates it with the udev device it was recorded from.
+     *
+     * The udev records for the device this recording is associated with
+     * must already exist in the testbed
+     *
+     * @param name The unadorned filename for the ioctl records to add
+     */
+    void load_device_evemu(std::string const& name);
+
     UMockdevTestbed *testbed;
     std::string const recordings_path;
 };
 
=== modified file 'tests/integration-tests/client/test_screencast.cpp'
--- tests/integration-tests/client/test_screencast.cpp	2015-03-31 02:35:42 +0000
+++ tests/integration-tests/client/test_screencast.cpp	2015-05-14 07:50:26 +0000
@@ -18,7 +18,8 @@
 
 #include "mir_protobuf.pb.h"
 #include "src/client/default_connection_configuration.h"
-#include "mir/dispatch/simple_dispatch_thread.h"
+#include "src/client/rpc/mir_basic_rpc_channel.h"
+#include "mir/dispatch/threaded_dispatcher.h"
 #include "mir/dispatch/dispatchable.h"
 
 #include "mir/frontend/connector.h"
@@ -78,16 +79,15 @@
         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));
+            std::make_shared<md::ThreadedDispatcher>("I/O loop", rpc_channel);
     }
 
     char const* const test_socket = "./test_socket_screencast";
     std::shared_ptr<StubScreencastServerTool> const server_tool;
     std::shared_ptr<mt::TestProtobufServer> test_server;
-    std::shared_ptr<google::protobuf::RpcChannel> rpc_channel;
+    std::shared_ptr<mir::client::rpc::MirBasicRpcChannel> rpc_channel;
     std::shared_ptr<mir::protobuf::DisplayServer> protobuf_server;
-    std::shared_ptr<mir::dispatch::SimpleDispatchThread> eventloop;
+    std::shared_ptr<mir::dispatch::ThreadedDispatcher> eventloop;
 };
 
 }
 
=== modified file 'tests/mir_test_doubles/test_protobuf_client.cpp'
--- tests/mir_test_doubles/test_protobuf_client.cpp	2015-04-01 19:39:19 +0000
+++ tests/mir_test_doubles/test_protobuf_client.cpp	2015-05-14 07:50:26 +0000
@@ -26,7 +26,7 @@
 #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 "mir/dispatch/threaded_dispatcher.h"
 #include "mir/events/event_private.h"
 
 #include <thread>
@@ -46,7 +46,7 @@
         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))},
+    eventloop{std::make_shared<md::ThreadedDispatcher>("I/O loop", 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/mir_test_framework/CMakeLists.txt'
--- tests/mir_test_framework/CMakeLists.txt	2015-04-30 11:36:36 +0000
+++ tests/mir_test_framework/CMakeLists.txt	2015-05-14 07:50:26 +0000
@@ -14,6 +14,7 @@
 add_definitions(
     -DMIR_CLIENT_PLATFORM_PATH="${MIR_CLIENT_PLATFORM_PATH}"
     -DMIR_SERVER_PLATFORM_PATH="${MIR_SERVER_PLATFORM_PATH}"
+    -DTEST_RECORDING_PATH="${CMAKE_INSTALL_PREFIX}/share/mir/udev_recordings"
     -DMIR_CLIENT_PLATFORM_ABI_STRING="${MIR_CLIENT_PLATFORM_ABI}"
     -DMIR_SERVER_GRAPHICS_PLATFORM_ABI_STRING="${MIR_SERVER_GRAPHICS_PLATFORM_ABI}"
     )
@@ -158,3 +159,5 @@
 install(TARGETS mirplatformgraphicsstub LIBRARY DESTINATION ${MIR_SERVER_PLATFORM_PATH})
 
 install(TARGETS mirclientplatformstub LIBRARY DESTINATION ${MIR_CLIENT_PLATFORM_PATH})
+
+install(DIRECTORY udev_recordings DESTINATION ${CMAKE_INSTALL_PREFIX}/share/mir)
 
=== modified file 'tests/mir_test_framework/udev_environment.cpp'
--- tests/mir_test_framework/udev_environment.cpp	2015-01-21 07:34:50 +0000
+++ tests/mir_test_framework/udev_environment.cpp	2015-05-14 07:50:26 +0000
@@ -37,9 +37,33 @@
 
 namespace mtf = mir_test_framework;
 
+namespace
+{
+std::string find_recording_path()
+{
+    std::initializer_list<std::string> candidates{mtf::executable_path() + "/udev_recordings", TEST_RECORDING_PATH};
+    for (auto candidate : candidates)
+    {
+        struct stat sb;
+        if (stat(candidate.c_str(), &sb) == 0)
+        {
+            if (S_ISDIR(sb.st_mode))
+            {
+                return candidate;
+            }
+        }
+    }
+    BOOST_THROW_EXCEPTION((std::runtime_error{"Failed to find udev recordings directory"}));
+}
+}
+
 mtf::UdevEnvironment::UdevEnvironment()
-    : recordings_path(mtf::executable_path() + "/udev_recordings")
+    : recordings_path{find_recording_path()}
 {
+    if (!umockdev_in_mock_environment())
+    {
+        BOOST_THROW_EXCEPTION((std::runtime_error{"Test not running under umockdev!"}));
+    }
     testbed = umockdev_testbed_new();
 }
 
@@ -109,4 +133,45 @@
             }
         }
     }
+
+    auto script_filename = recordings_path + "/" + name + ".script";
+    if (stat(script_filename.c_str(), &sb) == 0)
+    {
+        if (S_ISREG(sb.st_mode) || S_ISLNK(sb.st_mode))
+        {
+            if (!umockdev_testbed_load_script(testbed, NULL, script_filename.c_str(), &err))
+            {
+                BOOST_THROW_EXCEPTION(std::runtime_error(std::string("Failed to load device recording: ") +
+                                                         err->message));
+            }
+        }
+    }
+}
+
+void mtf::UdevEnvironment::load_device_ioctls(std::string const& name)
+{
+    auto ioctls_filename = recordings_path + "/" + name + ".ioctl";
+
+    GError* err = nullptr;
+    if (!umockdev_testbed_load_ioctl(testbed, NULL, ioctls_filename.c_str(), &err))
+    {
+        std::runtime_error exception{std::string("Failed to load ioctl recording: ") +
+                                     err->message};
+        g_error_free(err);
+        BOOST_THROW_EXCEPTION(exception);
+    }
+}
+
+void mtf::UdevEnvironment::load_device_evemu(std::string const& name)
+{
+    auto evemu_filename = recordings_path + "/" + name + ".evemu";
+
+    GError* err = nullptr;
+    if (!umockdev_testbed_load_evemu_events(testbed, NULL, evemu_filename.c_str(), &err))
+    {
+        std::runtime_error exception{std::string{"Failed to load evemu recording: "} +
+                                     err->message};
+        g_error_free(err);
+        BOOST_THROW_EXCEPTION(exception);
+    }
 }
 
=== added file 'tests/mir_test_framework/udev_recordings/laptop-keyboard-hello.evemu'
--- tests/mir_test_framework/udev_recordings/laptop-keyboard-hello.evemu	1970-01-01 00:00:00 +0000
+++ tests/mir_test_framework/udev_recordings/laptop-keyboard-hello.evemu	2015-05-14 07:50:26 +0000
@@ -0,0 +1,272 @@
+# device /dev/input/event4
+# EVEMU 1.2
+# Input device name: "AT Translated Set 2 keyboard"
+# Input device ID: bus 0x11 vendor 0x01 product 0x01 version 0xab83
+# Supported events:
+#   Event type 0 (EV_SYN)
+#     Event code 0 (SYN_REPORT)
+#     Event code 1 (SYN_CONFIG)
+#     Event code 4 (FF_STATUS_STOPPED)
+#     Event code 17 ((null))
+#     Event code 20 ((null))
+#   Event type 1 (EV_KEY)
+#     Event code 1 (KEY_ESC)
+#     Event code 2 (KEY_1)
+#     Event code 3 (KEY_2)
+#     Event code 4 (KEY_3)
+#     Event code 5 (KEY_4)
+#     Event code 6 (KEY_5)
+#     Event code 7 (KEY_6)
+#     Event code 8 (KEY_7)
+#     Event code 9 (KEY_8)
+#     Event code 10 (KEY_9)
+#     Event code 11 (KEY_0)
+#     Event code 12 (KEY_MINUS)
+#     Event code 13 (KEY_EQUAL)
+#     Event code 14 (KEY_BACKSPACE)
+#     Event code 15 (KEY_TAB)
+#     Event code 16 (KEY_Q)
+#     Event code 17 (KEY_W)
+#     Event code 18 (KEY_E)
+#     Event code 19 (KEY_R)
+#     Event code 20 (KEY_T)
+#     Event code 21 (KEY_Y)
+#     Event code 22 (KEY_U)
+#     Event code 23 (KEY_I)
+#     Event code 24 (KEY_O)
+#     Event code 25 (KEY_P)
+#     Event code 26 (KEY_LEFTBRACE)
+#     Event code 27 (KEY_RIGHTBRACE)
+#     Event code 28 (KEY_ENTER)
+#     Event code 29 (KEY_LEFTCTRL)
+#     Event code 30 (KEY_A)
+#     Event code 31 (KEY_S)
+#     Event code 32 (KEY_D)
+#     Event code 33 (KEY_F)
+#     Event code 34 (KEY_G)
+#     Event code 35 (KEY_H)
+#     Event code 36 (KEY_J)
+#     Event code 37 (KEY_K)
+#     Event code 38 (KEY_L)
+#     Event code 39 (KEY_SEMICOLON)
+#     Event code 40 (KEY_APOSTROPHE)
+#     Event code 41 (KEY_GRAVE)
+#     Event code 42 (KEY_LEFTSHIFT)
+#     Event code 43 (KEY_BACKSLASH)
+#     Event code 44 (KEY_Z)
+#     Event code 45 (KEY_X)
+#     Event code 46 (KEY_C)
+#     Event code 47 (KEY_V)
+#     Event code 48 (KEY_B)
+#     Event code 49 (KEY_N)
+#     Event code 50 (KEY_M)
+#     Event code 51 (KEY_COMMA)
+#     Event code 52 (KEY_DOT)
+#     Event code 53 (KEY_SLASH)
+#     Event code 54 (KEY_RIGHTSHIFT)
+#     Event code 55 (KEY_KPASTERISK)
+#     Event code 56 (KEY_LEFTALT)
+#     Event code 57 (KEY_SPACE)
+#     Event code 58 (KEY_CAPSLOCK)
+#     Event code 59 (KEY_F1)
+#     Event code 60 (KEY_F2)
+#     Event code 61 (KEY_F3)
+#     Event code 62 (KEY_F4)
+#     Event code 63 (KEY_F5)
+#     Event code 64 (KEY_F6)
+#     Event code 65 (KEY_F7)
+#     Event code 66 (KEY_F8)
+#     Event code 67 (KEY_F9)
+#     Event code 68 (KEY_F10)
+#     Event code 69 (KEY_NUMLOCK)
+#     Event code 70 (KEY_SCROLLLOCK)
+#     Event code 71 (KEY_KP7)
+#     Event code 72 (KEY_KP8)
+#     Event code 73 (KEY_KP9)
+#     Event code 74 (KEY_KPMINUS)
+#     Event code 75 (KEY_KP4)
+#     Event code 76 (KEY_KP5)
+#     Event code 77 (KEY_KP6)
+#     Event code 78 (KEY_KPPLUS)
+#     Event code 79 (KEY_KP1)
+#     Event code 80 (KEY_KP2)
+#     Event code 81 (KEY_KP3)
+#     Event code 82 (KEY_KP0)
+#     Event code 83 (KEY_KPDOT)
+#     Event code 85 (KEY_ZENKAKUHANKAKU)
+#     Event code 86 (KEY_102ND)
+#     Event code 87 (KEY_F11)
+#     Event code 88 (KEY_F12)
+#     Event code 89 (KEY_RO)
+#     Event code 90 (KEY_KATAKANA)
+#     Event code 91 (KEY_HIRAGANA)
+#     Event code 92 (KEY_HENKAN)
+#     Event code 93 (KEY_KATAKANAHIRAGANA)
+#     Event code 94 (KEY_MUHENKAN)
+#     Event code 95 (KEY_KPJPCOMMA)
+#     Event code 96 (KEY_KPENTER)
+#     Event code 97 (KEY_RIGHTCTRL)
+#     Event code 98 (KEY_KPSLASH)
+#     Event code 99 (KEY_SYSRQ)
+#     Event code 100 (KEY_RIGHTALT)
+#     Event code 102 (KEY_HOME)
+#     Event code 103 (KEY_UP)
+#     Event code 104 (KEY_PAGEUP)
+#     Event code 105 (KEY_LEFT)
+#     Event code 106 (KEY_RIGHT)
+#     Event code 107 (KEY_END)
+#     Event code 108 (KEY_DOWN)
+#     Event code 109 (KEY_PAGEDOWN)
+#     Event code 110 (KEY_INSERT)
+#     Event code 111 (KEY_DELETE)
+#     Event code 112 (KEY_MACRO)
+#     Event code 113 (KEY_MUTE)
+#     Event code 114 (KEY_VOLUMEDOWN)
+#     Event code 115 (KEY_VOLUMEUP)
+#     Event code 116 (KEY_POWER)
+#     Event code 117 (KEY_KPEQUAL)
+#     Event code 118 (KEY_KPPLUSMINUS)
+#     Event code 119 (KEY_PAUSE)
+#     Event code 121 (KEY_KPCOMMA)
+#     Event code 122 (KEY_HANGEUL)
+#     Event code 123 (KEY_HANJA)
+#     Event code 124 (KEY_YEN)
+#     Event code 125 (KEY_LEFTMETA)
+#     Event code 126 (KEY_RIGHTMETA)
+#     Event code 127 (KEY_COMPOSE)
+#     Event code 128 (KEY_STOP)
+#     Event code 140 (KEY_CALC)
+#     Event code 142 (KEY_SLEEP)
+#     Event code 143 (KEY_WAKEUP)
+#     Event code 155 (KEY_MAIL)
+#     Event code 156 (KEY_BOOKMARKS)
+#     Event code 157 (KEY_COMPUTER)
+#     Event code 158 (KEY_BACK)
+#     Event code 159 (KEY_FORWARD)
+#     Event code 163 (KEY_NEXTSONG)
+#     Event code 164 (KEY_PLAYPAUSE)
+#     Event code 165 (KEY_PREVIOUSSONG)
+#     Event code 166 (KEY_STOPCD)
+#     Event code 172 (KEY_HOMEPAGE)
+#     Event code 173 (KEY_REFRESH)
+#     Event code 183 (KEY_F13)
+#     Event code 184 (KEY_F14)
+#     Event code 185 (KEY_F15)
+#     Event code 217 (KEY_SEARCH)
+#     Event code 226 (KEY_MEDIA)
+#   Event type 4 (EV_MSC)
+#     Event code 4 (MSC_SCAN)
+#   Event type 17 (EV_LED)
+#     Event code 0 (LED_NUML)
+#     Event code 1 (LED_CAPSL)
+#     Event code 2 (LED_SCROLLL)
+#   Event type 20 (EV_REP)
+# Properties:
+# N: AT Translated Set 2 keyboard
+# I: 0011 0001 0001 ab83
+# P: 00 00 00 00 00 00 00 00
+# B: 00 13 00 12 00 00 00 00 00
+# B: 01 fe ff ff ff ff ff ff ff
+# B: 01 ff ff ef ff df ff ff fe
+# B: 01 01 d0 00 f8 78 30 80 03
+# B: 01 00 00 00 02 04 00 00 00
+# B: 01 00 00 00 00 00 00 00 00
+# B: 01 00 00 00 00 00 00 00 00
+# B: 01 00 00 00 00 00 00 00 00
+# B: 01 00 00 00 00 00 00 00 00
+# B: 01 00 00 00 00 00 00 00 00
+# B: 01 00 00 00 00 00 00 00 00
+# B: 01 00 00 00 00 00 00 00 00
+# B: 01 00 00 00 00 00 00 00 00
+# B: 02 00 00 00 00 00 00 00 00
+# B: 03 00 00 00 00 00 00 00 00
+# B: 04 10 00 00 00 00 00 00 00
+# B: 05 00 00 00 00 00 00 00 00
+# B: 11 07 00 00 00 00 00 00 00
+# B: 12 00 00 00 00 00 00 00 00
+# B: 15 00 00 00 00 00 00 00 00
+# B: 15 00 00 00 00 00 00 00 00
+################################
+#      Waiting for events      #
+################################
+E: 0.000000 0004 0004 0028	# EV_MSC / MSC_SCAN             28
+E: 0.000000 0001 001c 0000	# EV_KEY / KEY_ENTER            0
+E: 0.000000 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 0.825945 0004 0004 0054	# EV_MSC / MSC_SCAN             54
+E: 0.825945 0001 0036 0001	# EV_KEY / KEY_RIGHTSHIFT       1
+E: 0.825945 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.101404 0004 0004 0054	# EV_MSC / MSC_SCAN             54
+E: 1.101404 0001 0036 0002	# EV_KEY / KEY_RIGHTSHIFT       2
+E: 1.101404 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.131449 0004 0004 0054	# EV_MSC / MSC_SCAN             54
+E: 1.131449 0001 0036 0002	# EV_KEY / KEY_RIGHTSHIFT       2
+E: 1.131449 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.161472 0004 0004 0036	# EV_MSC / MSC_SCAN             36
+E: 1.161472 0001 0024 0001	# EV_KEY / KEY_J                1
+E: 1.161472 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.221981 0004 0004 0036	# EV_MSC / MSC_SCAN             36
+E: 1.221981 0001 0024 0000	# EV_KEY / KEY_J                0
+E: 1.221981 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.241906 0004 0004 0054	# EV_MSC / MSC_SCAN             54
+E: 1.241906 0001 0036 0000	# EV_KEY / KEY_RIGHTSHIFT       0
+E: 1.241906 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.346773 0004 0004 0032	# EV_MSC / MSC_SCAN             32
+E: 1.346773 0001 0020 0001	# EV_KEY / KEY_D                1
+E: 1.346773 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.432213 0004 0004 0032	# EV_MSC / MSC_SCAN             32
+E: 1.432213 0001 0020 0000	# EV_KEY / KEY_D                0
+E: 1.432213 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.562060 0004 0004 0025	# EV_MSC / MSC_SCAN             25
+E: 1.562060 0001 0019 0001	# EV_KEY / KEY_P                1
+E: 1.562060 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.622477 0004 0004 0025	# EV_MSC / MSC_SCAN             25
+E: 1.622477 0001 0019 0000	# EV_KEY / KEY_P                0
+E: 1.622477 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.687305 0004 0004 0025	# EV_MSC / MSC_SCAN             25
+E: 1.687305 0001 0019 0001	# EV_KEY / KEY_P                1
+E: 1.687305 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.777807 0004 0004 0025	# EV_MSC / MSC_SCAN             25
+E: 1.777807 0001 0019 0000	# EV_KEY / KEY_P                0
+E: 1.777807 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.867577 0004 0004 0031	# EV_MSC / MSC_SCAN             31
+E: 1.867577 0001 001f 0001	# EV_KEY / KEY_S                1
+E: 1.867577 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 1.953019 0004 0004 0031	# EV_MSC / MSC_SCAN             31
+E: 1.953019 0001 001f 0000	# EV_KEY / KEY_S                0
+E: 1.953019 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 2.118004 0004 0004 0054	# EV_MSC / MSC_SCAN             54
+E: 2.118004 0001 0036 0001	# EV_KEY / KEY_RIGHTSHIFT       1
+E: 2.118004 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 2.348316 0004 0004 0002	# EV_MSC / MSC_SCAN             2
+E: 2.348316 0001 0002 0001	# EV_KEY / KEY_1                1
+E: 2.348316 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 2.489561 0004 0004 0002	# EV_MSC / MSC_SCAN             2
+E: 2.489561 0001 0002 0000	# EV_KEY / KEY_1                0
+E: 2.489561 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 2.523740 0004 0004 0054	# EV_MSC / MSC_SCAN             54
+E: 2.523740 0001 0036 0000	# EV_KEY / KEY_RIGHTSHIFT       0
+E: 2.523740 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 6.714286 0004 0004 0029	# EV_MSC / MSC_SCAN             29
+E: 6.714286 0001 001d 0001	# EV_KEY / KEY_LEFTCTRL         1
+E: 6.714286 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 6.989728 0004 0004 0029	# EV_MSC / MSC_SCAN             29
+E: 6.989728 0001 001d 0002	# EV_KEY / KEY_LEFTCTRL         2
+E: 6.989728 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 7.019767 0004 0004 0029	# EV_MSC / MSC_SCAN             29
+E: 7.019767 0001 001d 0002	# EV_KEY / KEY_LEFTCTRL         2
+E: 7.019767 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 7.049847 0004 0004 0029	# EV_MSC / MSC_SCAN             29
+E: 7.049847 0001 001d 0002	# EV_KEY / KEY_LEFTCTRL         2
+E: 7.049847 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 7.079889 0004 0004 0029	# EV_MSC / MSC_SCAN             29
+E: 7.079889 0001 001d 0002	# EV_KEY / KEY_LEFTCTRL         2
+E: 7.079889 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 7.109934 0004 0004 0029	# EV_MSC / MSC_SCAN             29
+E: 7.109934 0001 001d 0002	# EV_KEY / KEY_LEFTCTRL         2
+E: 7.109934 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 7.139943 0004 0004 0029	# EV_MSC / MSC_SCAN             29
+E: 7.139943 0001 001d 0002	# EV_KEY / KEY_LEFTCTRL         2
+E: 7.139943 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
+E: 7.170005 0004 0004 0023	# EV_MSC / MSC_SCAN             23
+E: 7.170005 0001 0017 0001	# EV_KEY / KEY_I                1
+E: 7.170005 0000 0000 0000	# ------------ SYN_REPORT (0) ----------
 
=== modified file 'tests/unit-tests/client/test_client_buffer_stream.cpp'
--- tests/unit-tests/client/test_client_buffer_stream.cpp	2015-04-16 19:56:08 +0000
+++ tests/unit-tests/client/test_client_buffer_stream.cpp	2015-05-14 07:50:26 +0000
@@ -132,7 +132,7 @@
         mcl::ClientBufferFactory& buffer_factory,
         mcl::BufferStreamMode mode=mcl::BufferStreamMode::Producer)
     {
-        return std::make_shared<mcl::BufferStream>(mock_protobuf_server, mode,
+        return std::make_shared<mcl::BufferStream>(nullptr, mock_protobuf_server, mode,
             std::make_shared<StubClientPlatform>(mt::fake_shared(buffer_factory)), protobuf_bs, perf_report, "");
     }
 };
@@ -448,7 +448,7 @@
 
     EXPECT_CALL(mock_perf_report, name_surface(StrEq(name))).Times(1);
 
-    auto bs = std::make_shared<mcl::BufferStream>(mock_protobuf_server, mcl::BufferStreamMode::Producer,
+    auto bs = std::make_shared<mcl::BufferStream>(nullptr, mock_protobuf_server, mcl::BufferStreamMode::Producer,
         std::make_shared<StubClientPlatform>(mt::fake_shared(stub_client_buffer_factory)),
         protobuf_bs, mt::fake_shared(mock_perf_report), name);
 }
 
=== modified file 'tests/unit-tests/client/test_client_mir_surface.cpp'
--- tests/unit-tests/client/test_client_mir_surface.cpp	2015-04-16 20:03:39 +0000
+++ tests/unit-tests/client/test_client_mir_surface.cpp	2015-05-14 07:50:26 +0000
@@ -30,8 +30,8 @@
 #include "src/client/mir_connection.h"
 #include "src/client/default_connection_configuration.h"
 #include "src/client/rpc/null_rpc_report.h"
-#include "mir/dispatch/simple_dispatch_thread.h"
 #include "mir/dispatch/dispatchable.h"
+#include "mir/dispatch/threaded_dispatcher.h"
 
 #include "mir/frontend/connector.h"
 #include "mir/input/input_platform.h"
 
=== modified file 'tests/unit-tests/client/test_mir_connection.cpp'
--- tests/unit-tests/client/test_mir_connection.cpp	2015-03-31 02:35:42 +0000
+++ tests/unit-tests/client/test_mir_connection.cpp	2015-05-14 07:50:26 +0000
@@ -34,6 +34,10 @@
 #include "mir_test/test_protobuf_server.h"
 #include "mir_test/stub_server_tool.h"
 #include "mir_test_doubles/stub_client_buffer_factory.h"
+#include "mir_test/pipe.h"
+#include "mir_test/signal.h"
+#include "mir_test/fd_utils.h"
+#include "mir_test/test_dispatchable.h"
 
 #include "mir_protobuf.pb.h"
 
@@ -49,16 +53,19 @@
 namespace md = mir::dispatch;
 namespace geom = mir::geometry;
 namespace mtd = mir::test::doubles;
+namespace md = mir::dispatch;
+namespace mt = mir::test;
 
 namespace
 {
 
-struct MockRpcChannel : public mir::client::rpc::MirBasicRpcChannel,
-                        public mir::dispatch::Dispatchable
+struct MockRpcChannel : public mir::client::rpc::MirBasicRpcChannel
 {
     MockRpcChannel()
     {
-        ON_CALL(*this, watch_fd()).WillByDefault(testing::Return(mir::Fd{}));
+        using namespace testing;
+        ON_CALL(*this, watch_fd()).WillByDefault(Return(fake_epoll.read_fd()));
+        ON_CALL(*this, relevant_events()).WillByDefault(Return(md::FdEvent::readable));
     }
 
     void CallMethod(const google::protobuf::MethodDescriptor* method,
@@ -86,14 +93,35 @@
         complete->Run();
     }
 
+    // MirBasicRpcChannel
+    MOCK_METHOD0(process_next_request_first,void());
+
+    // Dispatchable
+    MOCK_CONST_METHOD0(watch_fd, mir::Fd());
+    MOCK_METHOD1(dispatch, bool(md::FdEvents));
+    MOCK_CONST_METHOD0(relevant_events, md::FdEvents());
+
     MOCK_METHOD2(connect, void(mp::ConnectParameters const*,mp::Connection*));
     MOCK_METHOD1(configure_display_sent, void(mp::DisplayConfiguration const*));
     MOCK_METHOD2(platform_operation,
                  void(mp::PlatformOperationMessage const*, mp::PlatformOperationMessage*));
 
-    MOCK_CONST_METHOD0(watch_fd, mir::Fd());
-    MOCK_METHOD1(dispatch, bool(md::FdEvents));
-    MOCK_CONST_METHOD0(relevant_events, md::FdEvents());
+    void trigger()
+    {
+        using namespace testing;
+        int dummy{0};
+        EXPECT_THAT(write(fake_epoll.write_fd(), &dummy, sizeof(dummy)), Eq(sizeof(dummy)));
+    }
+
+    void untrigger()
+    {
+        using namespace testing;
+        int dummy{0};
+        EXPECT_THAT(read(fake_epoll.read_fd(), &dummy, sizeof(dummy)), Eq(sizeof(dummy)));
+    }
+
+private:
+    mt::Pipe fake_epoll;
 };
 
 struct MockClientPlatform : public mcl::ClientPlatform
@@ -167,7 +195,7 @@
     {
     }
 
-    std::shared_ptr<::google::protobuf::RpcChannel> the_rpc_channel() override
+    std::shared_ptr<mcl::rpc::MirBasicRpcChannel> the_rpc_channel() override
     {
         return channel;
     }
@@ -194,22 +222,21 @@
     MirConnectionTest()
         : mock_platform{std::make_shared<testing::NiceMock<MockClientPlatform>>()},
           mock_channel{std::make_shared<testing::NiceMock<MockRpcChannel>>()},
-          conf{mock_platform, mock_channel},
-          connection{std::make_shared<MirConnection>(conf)}
+          conf{mock_platform, mock_channel}
     {
-        mock_platform->set_client_context(connection.get());
     }
 
     std::shared_ptr<testing::NiceMock<MockClientPlatform>> const mock_platform;
     std::shared_ptr<testing::NiceMock<MockRpcChannel>> const mock_channel;
     TestConnectionConfiguration conf;
-    std::shared_ptr<MirConnection> const connection;
 };
 
 TEST_F(MirConnectionTest, returns_correct_egl_native_display)
 {
     using namespace testing;
 
+    auto connection = std::make_shared<MirConnection>(conf);
+
     EGLNativeDisplayType native_display_raw = reinterpret_cast<EGLNativeDisplayType>(0xabcdef);
     auto native_display = std::make_shared<EGLNativeDisplayType>();
     *native_display = native_display_raw;
@@ -278,6 +305,8 @@
 {
     using namespace testing;
 
+    auto connection = std::make_shared<MirConnection>(conf);
+
     EXPECT_CALL(*mock_channel, connect(_,_))
         .WillOnce(Invoke(fill_display_configuration));
 
@@ -317,6 +346,8 @@
 {
     using namespace testing;
 
+    auto connection = std::make_shared<MirConnection>(conf);
+
     EXPECT_CALL(*mock_channel, connect(_,_))
         .WillOnce(Invoke(fill_display_configuration));
 
@@ -362,6 +393,8 @@
 {
     using namespace testing;
 
+    auto connection = std::make_shared<MirConnection>(conf);
+
     EXPECT_CALL(*mock_channel, connect(_,_))
         .WillOnce(Invoke(fill_display_configuration));
 
@@ -383,6 +416,8 @@
 {
     using namespace testing;
 
+    auto connection = std::make_shared<MirConnection>(conf);
+
     EXPECT_CALL(*mock_channel, connect(_,_))
         .WillOnce(Invoke(fill_display_configuration));
 
@@ -420,6 +455,8 @@
 {
     using namespace testing;
 
+    auto connection = std::make_shared<MirConnection>(conf);
+
     EXPECT_CALL(*mock_channel, connect(_,_))
         .WillOnce(Invoke(fill_surface_pixel_formats));
     MirWaitHandle* wait_handle = connection->connect("MirClientSurfaceTest",
@@ -443,6 +480,8 @@
 {
     using namespace testing;
 
+    auto connection = std::make_shared<MirConnection>(conf);
+
     EXPECT_CALL(*mock_channel, connect(_,_))
         .WillOnce(Invoke(fill_display_configuration));
 
@@ -509,6 +548,8 @@
 {
     using namespace testing;
 
+    auto connection = std::make_shared<MirConnection>(conf);
+
     MirSurfaceSpec params{nullptr, 640, 480, mir_pixel_format_abgr_8888};
     params.surface_name = __PRETTY_FUNCTION__;
 
@@ -537,6 +578,8 @@
 {
     using namespace testing;
 
+    auto connection = std::make_shared<MirConnection>(conf);
+
     MirSurfaceSpec params{nullptr, 640, 480, mir_pixel_format_abgr_8888};
     params.surface_name = __PRETTY_FUNCTION__;
 
@@ -588,6 +631,8 @@
         mir_platform_message_create(opcode),
         &mir_platform_message_release);
 
+    auto connection = std::make_shared<MirConnection>(conf);
+
     EXPECT_CALL(*mock_platform, platform_operation(request.get()))
         .WillOnce(Return(response.get()));
     EXPECT_CALL(*mock_channel, platform_operation(_,_))
@@ -620,6 +665,8 @@
     EXPECT_CALL(*mock_channel, platform_operation(_,_))
         .WillOnce(CopyRequestToResponse());
 
+    auto connection = std::make_shared<MirConnection>(conf);
+
     auto connect_wh =
         connection->connect("MirClientSurfaceTest", &connected_callback, nullptr);
     mir_wait_for(connect_wh);
@@ -633,3 +680,107 @@
     EXPECT_THAT(mir_platform_message_get_opcode(returned_response), Eq(opcode));
     mir_platform_message_release(returned_response);
 }
+
+TEST_F(MirConnectionTest, dispatch_works_with_automatic_dispatch)
+{
+    using namespace testing;
+
+    auto channel = std::dynamic_pointer_cast<MockRpcChannel>(conf.the_rpc_channel());
+
+    auto dispatched = std::make_shared<mir::test::Signal>();
+    ON_CALL(*channel, dispatch(_))
+        .WillByDefault(Invoke([dispatched](md::FdEvents) { dispatched->raise(); return true; }));
+
+    auto connection = std::make_shared<MirConnection>(conf, DispatchType::automatic);
+
+    channel->trigger();
+
+    EXPECT_TRUE(dispatched->wait_for(std::chrono::seconds{1}));
+}
+
+TEST_F(MirConnectionTest, manual_dispatch_is_not_automatically_dispatched)
+{
+    using namespace testing;
+
+    auto channel = std::dynamic_pointer_cast<MockRpcChannel>(conf.the_rpc_channel());
+
+    auto dispatched = std::make_shared<mir::test::Signal>();
+    ON_CALL(*channel, dispatch(_))
+        .WillByDefault(Invoke([dispatched](md::FdEvents) { dispatched->raise(); return true; }));
+
+    auto connection = std::make_shared<MirConnection>(conf, DispatchType::manual);
+
+    channel->trigger();
+
+    EXPECT_FALSE(dispatched->wait_for(std::chrono::seconds{1}));
+}
+
+TEST_F(MirConnectionTest, returns_invalid_watch_fd_when_using_automatic_dispatch)
+{
+    using namespace testing;
+
+    auto connection = std::make_shared<MirConnection>(conf, DispatchType::automatic);
+
+    EXPECT_THAT(connection->watch_fd(), Lt(0));
+}
+
+TEST_F(MirConnectionTest, returns_pollable_watch_fd_when_using_manual_dispatch)
+{
+    using namespace testing;
+
+    auto channel = std::dynamic_pointer_cast<MockRpcChannel>(conf.the_rpc_channel());
+
+    auto connection = std::make_shared<MirConnection>(conf, DispatchType::manual);
+
+    pollfd fd_readable;
+    fd_readable.events = POLLIN;
+    fd_readable.fd = connection->watch_fd();
+    EXPECT_TRUE(mt::std_call_succeeded(poll(&fd_readable, 1, 0)));
+}
+
+TEST_F(MirConnectionTest, can_add_dispatchables_to_connection)
+{
+    using namespace testing;
+    auto connection = std::make_shared<MirConnection>(conf, DispatchType::manual);
+
+    auto dispatchee = std::make_shared<mt::TestDispatchable>([](){});
+    connection->add_dispatchee(dispatchee);
+}
+
+TEST_F(MirConnectionTest, added_dispatchees_are_dispatched)
+{
+    using namespace testing;
+    auto connection = std::make_shared<MirConnection>(conf, DispatchType::manual);
+
+    bool dispatched{false};
+    auto dispatchee = std::make_shared<mt::TestDispatchable>([&dispatched](){ dispatched = true; });
+
+    connection->add_dispatchee(dispatchee);
+
+    dispatchee->trigger();
+
+    EXPECT_TRUE(mt::fd_is_readable(connection->watch_fd()));
+    connection->dispatch();
+
+    EXPECT_TRUE(dispatched);
+}
+
+TEST_F(MirConnectionTest, removed_dispatchees_are_no_longer_dispatched)
+{
+    using namespace testing;
+    auto connection = std::make_shared<MirConnection>(conf, DispatchType::manual);
+
+    bool dispatched{false};
+    auto dispatchee = std::make_shared<mt::TestDispatchable>([&dispatched](){ dispatched = true; });
+
+    connection->add_dispatchee(dispatchee);
+    connection->remove_dispatchee(dispatchee);
+
+    dispatchee->trigger();
+
+    EXPECT_FALSE(mir::test::fd_is_readable(connection->watch_fd()));
+
+    connection->dispatch();
+
+    EXPECT_FALSE(dispatched);
+}
 
=== modified file 'tests/unit-tests/client/test_mir_screencast.cpp'
--- tests/unit-tests/client/test_mir_screencast.cpp	2015-04-16 19:56:08 +0000
+++ tests/unit-tests/client/test_mir_screencast.cpp	2015-05-14 07:50:26 +0000
@@ -166,7 +166,7 @@
         using namespace ::testing;
 
         ON_CALL(*mock_buffer_stream_factory,
-        make_consumer_stream(_,_,_)).WillByDefault(
+        make_consumer_stream(_,_,_,_)).WillByDefault(
             Return(mt::fake_shared(mock_bs)));
     }
 
@@ -191,6 +191,7 @@
         .WillOnce(RunClosure());
 
     MirScreencast screencast{
+        nullptr,
         default_region,
         default_size,
         default_pixel_format, mock_server,
@@ -215,6 +216,7 @@
         .WillOnce(RunClosure());
 
     MirScreencast screencast{
+        nullptr,
         default_region,
         default_size,
         default_pixel_format, mock_server,
@@ -232,6 +234,7 @@
     EXPECT_CALL(mock_cb, call(_, &mock_cb));
 
     MirScreencast screencast{
+        nullptr,
         default_region,
         default_size,
         default_pixel_format, stub_server,
@@ -246,6 +249,7 @@
     using namespace testing;
 
     MirScreencast screencast{
+        nullptr,
         default_region,
         default_size,
         default_pixel_format, stub_server,
@@ -268,6 +272,7 @@
 
     EXPECT_THROW({
         MirScreencast screencast(
+            nullptr,
             default_region,
             invalid_size,
             default_pixel_format, stub_server,
@@ -277,6 +282,7 @@
 
     EXPECT_THROW({
         MirScreencast screencast(
+            nullptr,
             invalid_region,
             default_size,
             default_pixel_format, stub_server,
@@ -286,6 +292,7 @@
 
     EXPECT_THROW({
         MirScreencast screencast(
+            nullptr,
             default_region,
             default_size,
             mir_pixel_format_invalid, stub_server,
@@ -302,6 +309,7 @@
         .WillOnce(DoAll(SetCreateError(), RunClosure()));
 
     MirScreencast screencast{
+        nullptr,
         default_region,
         default_size,
         default_pixel_format, mock_server,
@@ -323,6 +331,7 @@
     EXPECT_CALL(mock_cb, call(_,&mock_cb));
 
     MirScreencast screencast{
+        nullptr,
         default_region,
         default_size,
         default_pixel_format, mock_server,
 
=== modified file 'tests/unit-tests/client/test_protobuf_rpc_channel.cpp'
--- tests/unit-tests/client/test_protobuf_rpc_channel.cpp	2015-04-13 05:50:18 +0000
+++ tests/unit-tests/client/test_protobuf_rpc_channel.cpp	2015-05-14 07:50:26 +0000
@@ -27,12 +27,16 @@
 #include "mir_protobuf_wire.pb.h"
 
 #include "mir_test_doubles/null_client_event_sink.h"
+#include "mir_test/fd_utils.h"
 
 #include <list>
 #include <endian.h>
 
+#include <sys/eventfd.h>
 #include <fcntl.h>
 
+#include <boost/throw_exception.hpp>
+
 #include <google/protobuf/descriptor.h>
 
 #include <gtest/gtest.h>
@@ -42,6 +46,9 @@
 namespace mclr = mir::client::rpc;
 namespace mtd = mir::test::doubles;
 namespace md = mir::dispatch;
+namespace mt = mir::test;
+
+namespace mt = mir::test;
 
 namespace
 {
@@ -58,7 +65,15 @@
 {
 public:
     MockStreamTransport()
+        : event_fd{eventfd(0, EFD_CLOEXEC)}
     {
+        if (event_fd == mir::Fd::invalid)
+        {
+            BOOST_THROW_EXCEPTION((std::system_error{errno,
+                                                     std::system_category(),
+                                                     "Failed to create event fd"}));
+        }
+
         using namespace testing;
         ON_CALL(*this, register_observer(_))
             .WillByDefault(Invoke(std::bind(&MockStreamTransport::register_observer_default,
@@ -82,6 +97,8 @@
 
     void add_server_message(std::vector<uint8_t> const& message)
     {
+        eventfd_t data_added{message.size()};
+        eventfd_write(event_fd, data_added);
         received_data.insert(received_data.end(), message.begin(), message.end());
     }
     void add_server_message(std::vector<uint8_t> const& message, std::initializer_list<mir::Fd> fds)
@@ -95,24 +112,30 @@
         return received_data.empty() && received_fds.empty();
     }
 
-    void notify_data_received()
-    {
-        do
-        {
-            for(auto& observer : observers)
-                observer->on_data_available();
-        }
-        while (!all_data_consumed());
-    }
-
     MOCK_METHOD1(register_observer, void(std::shared_ptr<Observer> const&));
     MOCK_METHOD1(unregister_observer, void(std::shared_ptr<Observer> const&));
     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());
+
+    mir::Fd watch_fd() const override
+    {
+        return event_fd;
+    }
+
+    bool dispatch(md::FdEvents /*events*/) override
+    {
+        for (auto& observer : observers)
+        {
+            observer->on_data_available();
+        }
+        return true;
+    }
+
+    md::FdEvents relevant_events() const override
+    {
+        return md::FdEvent::readable;
+    }
 
     // Transport interface
     void register_observer_default(std::shared_ptr<Observer> const& observer)
@@ -128,6 +151,9 @@
 
     void receive_data_default(void* buffer, size_t read_bytes, std::vector<mir::Fd>& fds)
     {
+        if (read_bytes == 0)
+            return;
+
         auto num_fds = fds.size();
         if (read_bytes > received_data.size())
         {
@@ -143,6 +169,11 @@
 
         received_data.erase(received_data.begin(), received_data.begin() + read_bytes);
         received_fds.erase(received_fds.begin(), received_fds.begin() + num_fds);
+
+        eventfd_t remaining_bytes;
+        eventfd_read(event_fd, &remaining_bytes);
+        remaining_bytes -= read_bytes;
+        eventfd_write(event_fd, remaining_bytes);
     }
 
     void send_message_default(std::vector<uint8_t> const& buffer)
@@ -156,6 +187,9 @@
     std::vector<uint8_t> received_data;
     std::vector<mir::Fd> received_fds;
     std::list<std::vector<uint8_t>> sent_messages;
+
+private:
+    mir::Fd event_fd;
 };
 
 class MirProtobufRpcChannelTest : public testing::Test
@@ -176,53 +210,56 @@
 
     MockStreamTransport* transport;
     std::shared_ptr<mcl::LifecycleControl> lifecycle;
-    std::shared_ptr<::google::protobuf::RpcChannel> channel;
+    std::shared_ptr<mclr::MirProtobufRpcChannel> channel;
 };
 
 }
 
-TEST_F(MirProtobufRpcChannelTest, ReadsFullMessages)
-{
-    std::vector<uint8_t> empty_message(sizeof(uint16_t));
-    std::vector<uint8_t> small_message(sizeof(uint16_t) + 8);
-    std::vector<uint8_t> large_message(sizeof(uint16_t) + 4096);
-
-    *reinterpret_cast<uint16_t*>(empty_message.data()) = htobe16(0);
-    *reinterpret_cast<uint16_t*>(small_message.data()) = htobe16(8);
-    *reinterpret_cast<uint16_t*>(large_message.data()) = htobe16(4096);
-
-    transport->add_server_message(empty_message);
-    transport->notify_data_received();
-    EXPECT_TRUE(transport->all_data_consumed());
-
-    transport->add_server_message(small_message);
-    transport->notify_data_received();
-    EXPECT_TRUE(transport->all_data_consumed());
-
-    transport->add_server_message(large_message);
-    transport->notify_data_received();
-    EXPECT_TRUE(transport->all_data_consumed());
-}
-
-TEST_F(MirProtobufRpcChannelTest, ReadsAllQueuedMessages)
-{
-    std::vector<uint8_t> empty_message(sizeof(uint16_t));
-    std::vector<uint8_t> small_message(sizeof(uint16_t) + 8);
-    std::vector<uint8_t> large_message(sizeof(uint16_t) + 4096);
-
-    *reinterpret_cast<uint16_t*>(empty_message.data()) = htobe16(0);
-    *reinterpret_cast<uint16_t*>(small_message.data()) = htobe16(8);
-    *reinterpret_cast<uint16_t*>(large_message.data()) = htobe16(4096);
-
-    transport->add_server_message(empty_message);
-    transport->add_server_message(small_message);
-    transport->add_server_message(large_message);
-
-    transport->notify_data_received();
-    EXPECT_TRUE(transport->all_data_consumed());
-}
-
-TEST_F(MirProtobufRpcChannelTest, SendsMessagesAtomically)
+TEST_F(MirProtobufRpcChannelTest, reads_full_messages)
+{
+    std::vector<uint8_t> empty_message(sizeof(uint16_t));
+    std::vector<uint8_t> small_message(sizeof(uint16_t) + 8);
+    std::vector<uint8_t> large_message(sizeof(uint16_t) + 4096);
+
+    *reinterpret_cast<uint16_t*>(empty_message.data()) = htobe16(0);
+    *reinterpret_cast<uint16_t*>(small_message.data()) = htobe16(8);
+    *reinterpret_cast<uint16_t*>(large_message.data()) = htobe16(4096);
+
+    transport->add_server_message(empty_message);
+    transport->dispatch(md::FdEvent::readable);
+    EXPECT_TRUE(transport->all_data_consumed());
+
+    transport->add_server_message(small_message);
+    transport->dispatch(md::FdEvent::readable);
+    EXPECT_TRUE(transport->all_data_consumed());
+
+    transport->add_server_message(large_message);
+    transport->dispatch(md::FdEvent::readable);
+    EXPECT_TRUE(transport->all_data_consumed());
+}
+
+TEST_F(MirProtobufRpcChannelTest, reads_all_queued_messages)
+{
+    std::vector<uint8_t> empty_message(sizeof(uint16_t));
+    std::vector<uint8_t> small_message(sizeof(uint16_t) + 8);
+    std::vector<uint8_t> large_message(sizeof(uint16_t) + 4096);
+
+    *reinterpret_cast<uint16_t*>(empty_message.data()) = htobe16(0);
+    *reinterpret_cast<uint16_t*>(small_message.data()) = htobe16(8);
+    *reinterpret_cast<uint16_t*>(large_message.data()) = htobe16(4096);
+
+    transport->add_server_message(empty_message);
+    transport->add_server_message(small_message);
+    transport->add_server_message(large_message);
+
+    while(mt::fd_is_readable(channel->watch_fd()))
+    {
+        channel->dispatch(md::FdEvent::readable);
+    }
+    EXPECT_TRUE(transport->all_data_consumed());
+}
+
+TEST_F(MirProtobufRpcChannelTest, sends_messages_atomically)
 {
     mir::protobuf::DisplayServer::Stub channel_user{channel.get(), mir::protobuf::DisplayServer::STUB_DOESNT_OWN_CHANNEL};
     mir::protobuf::ConnectParameters message;
@@ -233,7 +270,7 @@
     EXPECT_EQ(transport->sent_messages.size(), 1);
 }
 
-TEST_F(MirProtobufRpcChannelTest, SetsCorrectSizeWhenSendingMessage)
+TEST_F(MirProtobufRpcChannelTest, sets_correct_size_when_sending_message)
 {
     mir::protobuf::DisplayServer::Stub channel_user{channel.get(), mir::protobuf::DisplayServer::STUB_DOESNT_OWN_CHANNEL};
     mir::protobuf::ConnectParameters message;
@@ -246,7 +283,7 @@
     EXPECT_EQ(transport->sent_messages.front().size() - sizeof(uint16_t), message_header);
 }
 
-TEST_F(MirProtobufRpcChannelTest, ReadsFds)
+TEST_F(MirProtobufRpcChannelTest, reads_fds)
 {
     mir::protobuf::DisplayServer::Stub channel_user{channel.get(), mir::protobuf::DisplayServer::STUB_DOESNT_OWN_CHANNEL};
     mir::protobuf::Buffer reply;
@@ -288,7 +325,10 @@
         std::vector<uint8_t> dummy = {1};
         transport->add_server_message(dummy, fds);
 
-        transport->notify_data_received();
+        while(mt::fd_is_readable(channel->watch_fd()))
+        {
+            channel->dispatch(md::FdEvent::readable);
+        }
     }
 
     ASSERT_EQ(reply.fd_size(), fds.size());
@@ -300,7 +340,7 @@
     }
 }
 
-TEST_F(MirProtobufRpcChannelTest, NotifiesOfDisconnectOnWriteError)
+TEST_F(MirProtobufRpcChannelTest, notifies_of_disconnect_on_write_error)
 {
     using namespace ::testing;
 
@@ -328,7 +368,7 @@
     EXPECT_TRUE(disconnected);
 }
 
-TEST_F(MirProtobufRpcChannelTest, ForwardsDisconnectNotification)
+TEST_F(MirProtobufRpcChannelTest, forwards_disconnect_notification)
 {
     using namespace ::testing;
 
@@ -350,7 +390,7 @@
     EXPECT_TRUE(disconnected);
 }
 
-TEST_F(MirProtobufRpcChannelTest, NotifiesOfDisconnectOnlyOnce)
+TEST_F(MirProtobufRpcChannelTest, notifies_of_disconnect_only_once)
 {
     using namespace ::testing;
 
@@ -388,3 +428,196 @@
 
     EXPECT_TRUE(disconnected);
 }
+
+namespace
+{
+void set_flag(bool* flag)
+{
+    *flag = true;
+}
+}
+
+TEST_F(MirProtobufRpcChannelTest, delays_messages_not_requested)
+{
+    using namespace ::testing;
+
+    auto typed_channel = std::dynamic_pointer_cast<mclr::MirProtobufRpcChannel>(channel);
+
+    mir::protobuf::DisplayServer::Stub channel_user{channel.get(), mir::protobuf::DisplayServer::STUB_DOESNT_OWN_CHANNEL};
+    mir::protobuf::DRMMagic request;
+    mir::protobuf::DRMAuthMagicStatus reply;
+
+    bool first_response_called{false};
+    bool second_response_called{false};
+    channel_user.drm_auth_magic(nullptr,
+                                &request,
+                                &reply,
+                                google::protobuf::NewCallback(&set_flag, &first_response_called));
+
+    typed_channel->process_next_request_first();
+    channel_user.drm_auth_magic(nullptr,
+                                &request,
+                                &reply,
+                                google::protobuf::NewCallback(&set_flag, &second_response_called));
+
+    mir::protobuf::wire::Invocation wire_request;
+    mir::protobuf::wire::Result wire_reply;
+
+    wire_request.ParseFromArray(transport->sent_messages.front().data() + sizeof(uint16_t),
+                                transport->sent_messages.front().size() - sizeof(uint16_t));
+
+    transport->sent_messages.pop_front();
+
+    wire_reply.set_id(wire_request.id());
+    wire_reply.set_response(reply.SerializeAsString());
+
+    std::vector<uint8_t> buffer(wire_reply.ByteSize() + sizeof(uint16_t));
+    *reinterpret_cast<uint16_t*>(buffer.data()) = htobe16(wire_reply.ByteSize());
+    ASSERT_TRUE(wire_reply.SerializeToArray(buffer.data() + sizeof(uint16_t), buffer.size() - sizeof(uint16_t)));
+
+    transport->add_server_message(buffer);
+
+    wire_request.ParseFromArray(transport->sent_messages.front().data() + sizeof(uint16_t),
+                                transport->sent_messages.front().size() - sizeof(uint16_t));
+
+    transport->sent_messages.pop_front();
+
+    wire_reply.set_id(wire_request.id());
+    wire_reply.set_response(reply.SerializeAsString());
+
+    buffer.resize(wire_reply.ByteSize() + sizeof(uint16_t));
+    *reinterpret_cast<uint16_t*>(buffer.data()) = htobe16(wire_reply.ByteSize());
+    ASSERT_TRUE(wire_reply.SerializeToArray(buffer.data() + sizeof(uint16_t), buffer.size() - sizeof(uint16_t)));
+
+    transport->add_server_message(buffer);
+
+    // Read the first message; this should be queued for later processing...
+    EXPECT_TRUE(mt::fd_is_readable(typed_channel->watch_fd()));
+    typed_channel->dispatch(md::FdEvent::readable);
+
+    EXPECT_FALSE(first_response_called);
+    EXPECT_FALSE(second_response_called);
+
+    // Read the second message; this should be processed immediately...
+    EXPECT_TRUE(mt::fd_is_readable(typed_channel->watch_fd()));
+    typed_channel->dispatch(md::FdEvent::readable);
+
+    EXPECT_FALSE(first_response_called);
+    EXPECT_TRUE(second_response_called);
+
+    // Now, the first message should be ready to be processed...
+    EXPECT_TRUE(mt::fd_is_readable(typed_channel->watch_fd()));
+    typed_channel->dispatch(md::FdEvent::readable);
+
+    EXPECT_TRUE(first_response_called);
+    EXPECT_TRUE(second_response_called);
+}
+
+TEST_F(MirProtobufRpcChannelTest, delays_messages_with_fds_not_requested)
+{
+    using namespace ::testing;
+
+    auto typed_channel = std::dynamic_pointer_cast<mclr::MirProtobufRpcChannel>(channel);
+
+    mir::protobuf::DisplayServer::Stub channel_user{channel.get(), mir::protobuf::DisplayServer::STUB_DOESNT_OWN_CHANNEL};
+    mir::protobuf::DRMMagic drm_request;
+    mir::protobuf::DRMAuthMagicStatus drm_reply;
+
+    mir::protobuf::Buffer buffer_reply;
+    mir::protobuf::BufferRequest buffer_request;
+
+    bool first_response_called{false};
+    bool second_response_called{false};
+
+
+    channel_user.exchange_buffer(nullptr,
+                                 &buffer_request,
+                                 &buffer_reply,
+                                 google::protobuf::NewCallback(&set_flag, &first_response_called));
+
+    typed_channel->process_next_request_first();
+    channel_user.drm_auth_magic(nullptr,
+                                &drm_request,
+                                &drm_reply,
+                                google::protobuf::NewCallback(&set_flag, &second_response_called));
+
+
+    std::initializer_list<mir::Fd> fds = {mir::Fd{open("/dev/null", O_RDONLY)},
+                                          mir::Fd{open("/dev/null", O_RDONLY)},
+                                          mir::Fd{open("/dev/null", O_RDONLY)}};
+
+    {
+        mir::protobuf::Buffer reply_message;
+
+        for (auto fd : fds)
+            reply_message.add_fd(fd);
+        reply_message.set_fds_on_side_channel(fds.size());
+
+        mir::protobuf::wire::Invocation request;
+        mir::protobuf::wire::Result reply;
+
+        request.ParseFromArray(transport->sent_messages.front().data() + sizeof(uint16_t),
+                               transport->sent_messages.front().size() - sizeof(uint16_t));
+
+        transport->sent_messages.pop_front();
+
+        reply.set_id(request.id());
+        reply.set_response(reply_message.SerializeAsString());
+
+        ASSERT_TRUE(reply.has_id());
+        ASSERT_TRUE(reply.has_response());
+
+        std::vector<uint8_t> buffer(reply.ByteSize() + sizeof(uint16_t));
+        *reinterpret_cast<uint16_t*>(buffer.data()) = htobe16(reply.ByteSize());
+        ASSERT_TRUE(reply.SerializeToArray(buffer.data() + sizeof(uint16_t), buffer.size() - sizeof(uint16_t)));
+
+        transport->add_server_message(buffer);
+
+        // Because our protocol is a bit silly...
+        std::vector<uint8_t> dummy = {1};
+        transport->add_server_message(dummy, fds);
+    }
+
+    {
+        mir::protobuf::DRMAuthMagicStatus reply;
+
+        mir::protobuf::wire::Invocation wire_request;
+        mir::protobuf::wire::Result wire_reply;
+
+        wire_request.ParseFromArray(transport->sent_messages.front().data() + sizeof(uint16_t),
+                                    transport->sent_messages.front().size() - sizeof(uint16_t));
+
+        transport->sent_messages.pop_front();
+
+        wire_reply.set_id(wire_request.id());
+        wire_reply.set_response(reply.SerializeAsString());
+
+        std::vector<uint8_t> buffer(wire_reply.ByteSize() + sizeof(uint16_t));
+        *reinterpret_cast<uint16_t*>(buffer.data()) = htobe16(wire_reply.ByteSize());
+        ASSERT_TRUE(wire_reply.SerializeToArray(buffer.data() + sizeof(uint16_t), buffer.size() - sizeof(uint16_t)));
+
+        transport->add_server_message(buffer);
+    }
+
+    // Read the first message; this should be queued for later processing...
+    EXPECT_TRUE(mt::fd_is_readable(typed_channel->watch_fd()));
+    typed_channel->dispatch(md::FdEvent::readable);
+
+    EXPECT_FALSE(first_response_called);
+    EXPECT_FALSE(second_response_called);
+
+    // Read the second message; this should be processed immediately...
+    EXPECT_TRUE(mt::fd_is_readable(typed_channel->watch_fd()));
+    typed_channel->dispatch(md::FdEvent::readable);
+
+    EXPECT_FALSE(first_response_called);
+    EXPECT_TRUE(second_response_called);
+
+    // Now, the first message should be ready to be processed...
+    EXPECT_TRUE(mt::fd_is_readable(typed_channel->watch_fd()));
+    typed_channel->dispatch(md::FdEvent::readable);
+
+    EXPECT_TRUE(first_response_called);
+    EXPECT_TRUE(second_response_called);
+}
+
 
=== modified file 'tests/unit-tests/dispatch/CMakeLists.txt'
--- tests/unit-tests/dispatch/CMakeLists.txt	2015-03-31 02:35:42 +0000
+++ tests/unit-tests/dispatch/CMakeLists.txt	2015-05-14 07:50:26 +0000
@@ -1,6 +1,6 @@
 list(APPEND UNIT_TEST_SOURCES
+  ${CMAKE_CURRENT_SOURCE_DIR}/test_threaded_dispatcher.cpp
   ${CMAKE_CURRENT_SOURCE_DIR}/test_action_queue.cpp
-  ${CMAKE_CURRENT_SOURCE_DIR}/test_simple_dispatch_thread.cpp
   ${CMAKE_CURRENT_SOURCE_DIR}/test_multiplexing_dispatchable.cpp
   ${CMAKE_CURRENT_SOURCE_DIR}/test_dispatch_utils.cpp
 )
 
=== modified file 'tests/unit-tests/dispatch/test_multiplexing_dispatchable.cpp'
--- tests/unit-tests/dispatch/test_multiplexing_dispatchable.cpp	2015-02-11 23:15:33 +0000
+++ tests/unit-tests/dispatch/test_multiplexing_dispatchable.cpp	2015-05-14 07:50:26 +0000
@@ -17,7 +17,7 @@
  */
 
 #include "mir/dispatch/multiplexing_dispatchable.h"
-#include "mir/dispatch/simple_dispatch_thread.h"
+#include "mir/dispatch/threaded_dispatcher.h"
 #include "mir/fd.h"
 #include "mir_test/pipe.h"
 #include "mir_test/signal.h"
@@ -217,8 +217,8 @@
     auto dispatcher = std::make_shared<md::MultiplexingDispatchable>();
     dispatcher->add_watch(dispatchee);
 
-    md::SimpleDispatchThread first_loop{dispatcher};
-    md::SimpleDispatchThread second_loop{dispatcher};
+    md::ThreadedDispatcher eventloop{"Fools!", dispatcher};
+    eventloop.add_thread();
 
     EXPECT_TRUE(second_dispatch->wait_for(std::chrono::seconds{5}));
 }
@@ -315,15 +315,15 @@
 
     dispatchee->trigger();
 
-    md::SimpleDispatchThread eventloop{dispatcher};
+    md::ThreadedDispatcher eventloop{"I'll destroy", dispatcher};
 
-    EXPECT_TRUE(in_dispatch->wait_for(std::chrono::seconds{1}));
+    EXPECT_TRUE(in_dispatch->wait_for(std::chrono::seconds{5}));
 
     dispatcher->remove_watch(dispatchee);
     dispatchee.reset();
     canary.reset();
 
-    EXPECT_TRUE(canary_killed->wait_for(std::chrono::seconds{2}));
+    EXPECT_TRUE(canary_killed->wait_for(std::chrono::seconds{5}));
 }
 
 TEST(MultiplexingDispatchableTest, destruction_is_threadsafe)
@@ -348,28 +348,27 @@
 
     mt::AutoJoinThread dispatch_thread{[dispatcher]() { dispatcher->dispatch(md::FdEvent::readable); }};
 
-    EXPECT_TRUE(in_dispatch->wait_for(std::chrono::seconds{1}));
+    EXPECT_TRUE(in_dispatch->wait_for(std::chrono::seconds{5}));
 
     dispatcher->remove_watch(dispatchee);
     dispatcher.reset();
     dispatchee.reset();
     canary.reset();
 
-    EXPECT_TRUE(canary_killed->wait_for(std::chrono::seconds{2}));
+    EXPECT_TRUE(canary_killed->wait_for(std::chrono::seconds{5}));
 }
 
 TEST(MultiplexingDispatchableTest, stress_test_threading)
 {
     using namespace testing;
-
     int const dispatchee_count{20};
 
     auto dispatcher = std::make_shared<md::MultiplexingDispatchable>();
 
-    std::vector<std::shared_ptr<md::SimpleDispatchThread>> eventloops;
+    auto event_dispatcher = std::make_shared<md::ThreadedDispatcher>(__func__, dispatcher);
     for (int i = 0 ; i < dispatchee_count + 5 ; ++i)
     {
-        eventloops.push_back(std::make_shared<md::SimpleDispatchThread>(dispatcher));
+        event_dispatcher->add_thread();
     }
 
     std::vector<std::shared_ptr<mt::Signal>> canary_tomb;
@@ -397,12 +396,12 @@
 
     dispatchees.clear();
     dispatcher.reset();
-    eventloops.clear();
+    event_dispatcher.reset();
 
     for (auto headstone : canary_tomb)
     {
         // Use assert so as to not block for *ages* on failure
-        ASSERT_TRUE(headstone->wait_for(std::chrono::seconds{2}));
+        ASSERT_TRUE(headstone->wait_for(std::chrono::seconds{5}));
     }
 }
 
@@ -456,10 +455,10 @@
 
     first_dispatchee->trigger();
 
-    md::SimpleDispatchThread eventloop_one{dispatcher};
-    md::SimpleDispatchThread eventloop_two{dispatcher};
+    md::ThreadedDispatcher eventloop{"them all!", dispatcher};
+    eventloop.add_thread();
 
-    EXPECT_TRUE(in_dispatch->wait_for(std::chrono::seconds{1}));
+    EXPECT_TRUE(in_dispatch->wait_for(std::chrono::seconds{5}));
 
     dispatcher->remove_watch(dummy_dispatchee);
     dispatcher->remove_watch(first_dispatchee);
@@ -470,7 +469,7 @@
 
     unblock_dispatchee->raise();
 
-    EXPECT_TRUE(canary_killed->wait_for(std::chrono::seconds{2}));
+    EXPECT_TRUE(canary_killed->wait_for(std::chrono::seconds{5}));
 }
 
 TEST(MultiplexingDispatchableTest, automatic_removals_are_threadsafe)
@@ -481,7 +480,11 @@
 
     dispatcher->add_watch(dispatchee, md::DispatchReentrancy::reentrant);
 
-    md::SimpleDispatchThread one{dispatcher}, two{dispatcher}, three{dispatcher}, four{dispatcher};
+    md::ThreadedDispatcher eventloop{"Eeelo", dispatcher};
+
+    eventloop.add_thread();
+    eventloop.add_thread();
+    eventloop.add_thread();
 
     dispatchee->trigger();
 }
 
=== removed file 'tests/unit-tests/dispatch/test_simple_dispatch_thread.cpp'
--- tests/unit-tests/dispatch/test_simple_dispatch_thread.cpp	2015-05-01 14:47:55 +0000
+++ tests/unit-tests/dispatch/test_simple_dispatch_thread.cpp	1970-01-01 00:00:00 +0000
@@ -1,291 +0,0 @@
-/*
- * 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 "mir_test_framework/process.h"
-#include "mir_test/cross_process_action.h"
-
-#include <fcntl.h>
-
-#include <atomic>
-#include <exception>
-#include <thread>
-
-#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;
-    using namespace std::chrono_literals;
-
-    int constexpr expected_count{10};
-    auto const dispatched_more_than_enough = std::make_shared<mt::Signal>();
-
-    auto delegate =
-        [dispatched_more_than_enough, dispatch_count = 0](md::FdEvents) mutable
-        {
-            if (++dispatch_count == expected_count)
-            {
-                return false;
-            }
-            if (dispatch_count > expected_count)
-            {
-                dispatched_more_than_enough->raise();
-            }
-            return true;
-        };
-    auto const 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(1s));
-}
-
-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}));
-}
-
-// Regression test for: lp #1439719
-// The bug involves uninitialized memory and is also sensitive to signal
-// timings, so this test does not always catch the problem. However, repeated
-// runs (~300, YMMV) consistently fail when run against the problematic code.
-TEST_F(SimpleDispatchThreadTest, keeps_dispatching_after_signal_interruption)
-{
-    using namespace std::chrono_literals;
-    mt::CrossProcessAction stop_and_restart_process;
-
-    auto child = mir_test_framework::fork_and_run_in_a_different_process(
-        [&]
-        {
-            auto dispatched = std::make_shared<mt::Signal>();
-            auto dispatchable = std::make_shared<mt::TestDispatchable>(
-                [dispatched]() { dispatched->raise(); });
-
-            md::SimpleDispatchThread dispatcher{dispatchable};
-            // Ensure the dispatcher has started
-            dispatchable->trigger();
-            EXPECT_TRUE(dispatched->wait_for(1s));
-
-            stop_and_restart_process();
-
-            dispatched->reset();
-            // The dispatcher shouldn't have been affected by the signal
-            dispatchable->trigger();
-            EXPECT_TRUE(dispatched->wait_for(1s));
-            exit(HasFailure() ? EXIT_FAILURE : EXIT_SUCCESS);
-        },
-        []{ return 1; });
-
-    stop_and_restart_process.exec(
-        [child]
-        {
-            // Increase chances of interrupting the dispatch mechanism
-            for (int i = 0; i < 100; ++i)
-            {
-                child->stop();
-                child->cont();
-            }
-        });
-
-    auto const result = child->wait_for_termination(10s);
-    EXPECT_TRUE(result.succeeded());
-}
-
-using SimpleDispatchThreadDeathTest = SimpleDispatchThreadTest;
-
-TEST_F(SimpleDispatchThreadDeathTest, destroying_dispatcher_from_a_callback_is_an_error)
-{
-    using namespace testing;
-    using namespace std::literals::chrono_literals;
-
-    EXPECT_EXIT(
-    {
-        std::mutex mutex;
-        md::SimpleDispatchThread* dispatcher;
-    
-        auto dispatchable = std::make_shared<mt::TestDispatchable>([&dispatcher, &mutex]{
-            std::lock_guard<decltype(mutex)> lock{mutex};
-            delete dispatcher;
-        });
-        
-        {
-            std::lock_guard<decltype(mutex)> lock{mutex};
-            dispatchable->trigger();
-            dispatcher = new md::SimpleDispatchThread{dispatchable};
-        }
-        std::this_thread::sleep_for(10s);
-    }, KilledBySignal(SIGABRT), ".*Destroying SimpleDispatchThread.*");
-}
-
-TEST_F(SimpleDispatchThreadTest, executes_exception_handler_with_current_exception)
-{
-    using namespace std::chrono_literals;
-    auto dispatched = std::make_shared<mt::Signal>();
-    std::exception_ptr exception;
-
-    auto dispatchable = std::make_shared<mt::TestDispatchable>(
-        []()
-        {
-            throw std::runtime_error("thrown");
-        });
-
-    md::SimpleDispatchThread dispatcher{dispatchable,
-        [&dispatched,&exception]()
-        {
-            exception = std::current_exception();
-            if (exception)
-                dispatched->raise();
-        }};
-    dispatchable->trigger();
-    EXPECT_TRUE(dispatched->wait_for(10s));
-    EXPECT_TRUE(exception!=nullptr);
-}
 
=== added file 'tests/unit-tests/dispatch/test_threaded_dispatcher.cpp'
--- tests/unit-tests/dispatch/test_threaded_dispatcher.cpp	1970-01-01 00:00:00 +0000
+++ tests/unit-tests/dispatch/test_threaded_dispatcher.cpp	2015-05-14 07:50:26 +0000
@@ -0,0 +1,403 @@
+/*
+ * 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/threaded_dispatcher.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 "mir_test_framework/process.h"
+#include "mir_test/cross_process_action.h"
+
+#include <fcntl.h>
+
+#include <atomic>
+#include <thread>
+
+#include <gtest/gtest.h>
+#include <gmock/gmock.h>
+
+namespace md = mir::dispatch;
+namespace mt = mir::test;
+
+namespace
+{
+class ThreadedDispatcherTest : public ::testing::Test
+{
+public:
+    ThreadedDispatcherTest()
+    {
+        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