Mir

Merge lp:~albaguirre/mir/fix-tsan-findings into lp:mir

fix-tsan-findings
Merge into development-branch

Proposed by Alberto Aguirre on 2015-12-07

Status:

Merged

Approved by:

Alberto Aguirre on 2015-12-16

Approved revision:

no longer in the source branch.

Merged at revision:

3199

Proposed branch:

lp:~albaguirre/mir/fix-tsan-findings

Merge into:

lp:mir

Diff against target:

1121 lines (+232/-340)

16 files modified

cmake/MirCommon.cmake (+3/-1)
src/server/compositor/buffer_stream_surfaces.cpp (+10/-7)
src/server/compositor/multi_threaded_compositor.cpp (+16/-11)
src/server/scene/surface_stack.cpp (+2/-1)
src/server/scene/surface_stack.h (+4/-3)
tests/acceptance-tests/test_unresponsive_client.cpp (+12/-3)
tests/include/mir/test/stub_server_tool.h (+24/-7)
tests/include/mir/test/test_protobuf_client.h (+26/-50)
tests/include/mir_test_framework/stub_input_platform.h (+2/-0)
tests/mir_test_doubles/test_protobuf_client.cpp (+86/-133)
tests/mir_test_framework/stub_input_platform.cpp (+4/-0)
tests/unit-tests/client/test_client_mir_surface.cpp (+5/-1)
tests/unit-tests/frontend/test_protobuf_reports_errors.cpp (+1/-1)
tests/unit-tests/frontend/test_protobuf_surface_apis.cpp (+3/-7)
tests/unit-tests/frontend/test_published_socket_connector.cpp (+32/-114)
tests/unit-tests/thread/test_basic_thread_pool.cpp (+2/-1)

To merge this branch:

bzr merge lp:~albaguirre/mir/fix-tsan-findings

High

Fix Released

Link a bug report

Reviewer	Review Type	Date Requested	Status
Alan Griffiths			Approve on 2015-12-16
PS Jenkins bot (community)	continuous-integration		Approve on 2015-12-14
Daniel van Vugt		2015-12-07	Abstain on 2015-12-11
Review via email: mp+279816@code.launchpad.net

Commit message

Fix TSan findings

Description of the change

Fix TSan findings

Revision history for this message

PS Jenkins bot (ps-jenkins) wrote on 2015-12-07:

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

review: Needs Fixing (continuous-integration)

Revision history for this message

Alberto Aguirre (albaguirre) wrote on 2015-12-07:

Umm unsure if I introduced an issue or uncovered one...I'll check tomorrow.

Revision history for this message

Alan Griffiths (alan-griffiths) wrote on 2015-12-08:

Changes look sensible, but from multiple motivations which makes reviewing harder.

Triggering a rebuild to get another datapoint

Revision history for this message

PS Jenkins bot (ps-jenkins) wrote on 2015-12-08:

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

review: Needs Fixing (continuous-integration)

Revision history for this message

Alberto Aguirre (albaguirre) wrote on 2015-12-08:

^-- the xenial-amd64 failure also happens with lp:mir, it's lp:1523965

Revision history for this message

PS Jenkins bot (ps-jenkins) wrote on 2015-12-08:

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

review: Approve (continuous-integration)

Revision history for this message

Daniel van Vugt (vanvugt) wrote on 2015-12-10:

Sounds good.

Hopefully we'll reach a point where we're clean enough to automate this in CI...

review: Approve

Revision history for this message

Alberto Aguirre (albaguirre) wrote on 2015-12-10:

We had a CI job that ran Thread Sanitizer with a clang build. I believe it got axed after the GCC5/clang fiasco.

So now with Jenkass we should setup a GCC TSan job.

Revision history for this message

PS Jenkins bot (ps-jenkins) wrote on 2015-12-11:

review: Needs Fixing (continuous-integration)

Revision history for this message

Daniel van Vugt (vanvugt) wrote on 2015-12-11:

Failed due to lots of FD leaks. Is that right?

review: Abstain

Revision history for this message

PS Jenkins bot (ps-jenkins) wrote on 2015-12-11:

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

review: Approve (continuous-integration)

Revision history for this message

Alan Griffiths (alan-griffiths) wrote on 2015-12-14:

I'd really like to approve a lot of these changes as they are obviously correct.

But the rework of CompositingFunctor::operator() isn't...

~~~~

+ //Appease TSan, avoid destructor and this thread accessing the same shared_ptr instance
+ auto disp_listener = std::move(display_listener);

As display_listener is const, isn't this simpler as:

auto const disp_listener = display_listener;

Or even better, just make it a value capture in the lines below? Vis:

        auto display_registration = mir::raii::paired_calls(
            [display_listener=display_listener, &group=group]{group.for_each_display_buffer([&](mg::DisplayBuffer& buffer)
                { display_listener->add_display(buffer.view_area()); });},
            [display_listener=display_listener, &group=group]{group.for_each_display_buffer([&](mg::DisplayBuffer& buffer)
                { display_listener->remove_display(buffer.view_area()); });});

~~~~

- auto on_startup_failure = on_unwind(
- [this]
- {
- if (started_future.wait_for(0s) != std::future_status::ready)
- started.set_exception(std::current_exception());
- });
-
...
catch(...)
{
+ try
+ {
+ //Move the promise so that the promise destructor occurs here rather than in the thread
+ //destroying CompositingFunctor, mostly to appease TSan
+ auto promise = std::move(started);
+ promise.set_exception(std::current_exception());
+ }
+ catch(...)
+ {
+ }

So we no longer use "on_unwind()" because the destructor of the moved promise will throw after we set_exception()? And then we eat the exception anyway? That seems an overly complex mechanism to destroy CompositingFunctor::promise.

review: Needs Fixing

Revision history for this message

Alberto Aguirre (albaguirre) wrote on 2015-12-14:

>As display_listener is const, isn't this simpler as:

> auto const disp_listener = display_listener;

The MP code I propose avoids an unnecessary ref count increase, but sure I can make that change.

>Or even better, just make it a value capture in the lines below? Vis:

You cannot capture a member variable directly; you can only through access through "this" (or by creating a local reference that the lambda can then capture).

> So we no longer use "on_unwind()" because the destructor of the moved promise
> will throw after we set_exception()? And then we eat the exception anyway?
> That seems an overly complex mechanism to destroy CompositingFunctor::promise.

No. I didn't see a need for unwind helper since we already have a catch block.

The main motivation is not removing the unwind helper, it's removing the query of std::future since it's not threadsafe (std::future sate would be accessed by two threads). Rather than synchronizing that access, I avoid querying the std::future at all. Since I won't check its status, then set_exception may throw if it already has a value and so we eat the exception.

The move of the std::promise is also to avoid two threads access std::promise state, the thread calling the destructor and the thread accessing the promise (the compositing thread).

Revision history for this message

Alan Griffiths (alan-griffiths) wrote on 2015-12-14:

> >As display_listener is const, isn't this simpler as:
>
> > auto const disp_listener = display_listener;
>
> The MP code I propose avoids an unnecessary ref count increase, but sure I can
> make that change.

Don't we get a single ref count increment in both cases? The proposed version can't move from a const display_listener so an unnamed temporary is created for the move.

> >Or even better, just make it a value capture in the lines below? Vis:
>
> You cannot capture a member variable directly; you can only through access
> through "this" (or by creating a local reference that the lambda can then
> capture).

I gave the syntax I was thinking of. Admittedly, that does create two copies of display_listener.

~~~~

> > So we no longer use "on_unwind()" because the destructor of the moved
> promise
> > will throw after we set_exception()? And then we eat the exception anyway?
> > That seems an overly complex mechanism to destroy
> CompositingFunctor::promise.
>
> No. I didn't see a need for unwind helper since we already have a catch block.
>
> The main motivation is not removing the unwind helper, it's removing the query
> of std::future since it's not threadsafe (std::future sate would be accessed
> by two threads). Rather than synchronizing that access, I avoid querying the
> std::future at all. Since I won't check its status, then set_exception may
> throw if it already has a value and so we eat the exception.
>
> The move of the std::promise is also to avoid two threads access std::promise
> state, the thread calling the destructor and the thread accessing the promise
> (the compositing thread).

I need to think about this - it sounds like the wrong synchronization objects are being used.

review: Needs Information

Revision history for this message

PS Jenkins bot (ps-jenkins) wrote on 2015-12-14:

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

review: Approve (continuous-integration)

Revision history for this message

Alan Griffiths (alan-griffiths) wrote on 2015-12-16:

OK, I still think the logic around started/started_future is expressed poorly, but that's pre-existing and not worth blocking on.

review: Approve

Preview Diff

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

Subscribers

People subscribed via source and target branches

to all changes:

Alberto Aguirre

Emanuele Antonio Faraone

Gerry Boland

Mir development team

 === modified file 'cmake/MirCommon.cmake'
 --- cmake/MirCommon.cmake	2015-12-14 06:21:46 +0000
 +++ cmake/MirCommon.cmake	2015-12-14 15:38:49 +0000
@@ -74,7 +74,9 @@
        endif()
      endif()
      # Space after ${TSAN_EXTRA_OPTIONS} works around bug in TSAN env. variable parsing
--    list(APPEND test_env "TSAN_OPTIONS=\"suppressions=${CMAKE_SOURCE_DIR}/tools/tsan-suppressions second_deadlock_stack=1 halt_on_error=1 history_size=7 ${TSAN_EXTRA_OPTIONS} \"")
++    list(APPEND test_env "TSAN_OPTIONS=\"suppressions=${CMAKE_SOURCE_DIR}/tools/tsan-suppressions second_deadlock_stack=1 halt_on_error=1 history_size=7 die_after_fork=0 ${TSAN_EXTRA_OPTIONS} \"")
++     # TSan does not support starting threads after fork
++    set(test_exclusion_filter "ThreadedDispatcherSignalTest.keeps_dispatching_after_signal_interruption")
    endif()
    if(SYSTEM_SUPPORTS_O_TMPFILE EQUAL 1)
 === modified file 'src/server/compositor/buffer_stream_surfaces.cpp'
 --- src/server/compositor/buffer_stream_surfaces.cpp	2015-11-21 00:23:00 +0000
 +++ src/server/compositor/buffer_stream_surfaces.cpp	2015-12-14 15:38:49 +0000
@@ -59,20 +59,20 @@
+ {
      buffer_bundle->client_release(buf);
+     {
--        std::unique_lock<std::mutex> lk(mutex);
++        std::lock_guard<decltype(mutex)> lk(mutex);
          first_frame_posted = true;
+     }
+ }
  geom::Size mc::BufferStreamSurfaces::stream_size()
+ {
--    std::unique_lock<std::mutex> lk(mutex);
++    std::lock_guard<decltype(mutex)> lk(mutex);
      return logical_size;
+ }
  void mc::BufferStreamSurfaces::resize(geom::Size const& size)
+ {
--    std::unique_lock<std::mutex> lk(mutex);
++    std::lock_guard<decltype(mutex)> lk(mutex);
      logical_size = size;
      buffer_bundle->resize(logical_size * scale);
+ }
@@ -117,14 +117,17 @@
  bool mc::BufferStreamSurfaces::has_submitted_buffer() const
+ {
--    std::unique_lock<std::mutex> lk(mutex);
++    std::lock_guard<decltype(mutex)> lk(mutex);
      return first_frame_posted;
+ }
  void mc::BufferStreamSurfaces::with_most_recent_buffer_do(std::function<void(graphics::Buffer&)> const& exec)
+ {
--    if (!first_frame_posted)
--        BOOST_THROW_EXCEPTION(std::runtime_error("No frame posted yet"));
++    {
++        std::lock_guard<decltype(mutex)> lk(mutex);
++        if (!first_frame_posted)
++            BOOST_THROW_EXCEPTION(std::runtime_error("No frame posted yet"));
++    }
      exec(*std::make_shared<mc::TemporarySnapshotBuffer>(buffer_bundle));
+ }
@@ -164,7 +167,7 @@
      if (new_scale <= 0.0f)
          BOOST_THROW_EXCEPTION(std::logic_error("invalid scale (must be greater than zero)"));
--    std::unique_lock<std::mutex> lk(mutex);
++    std::lock_guard<decltype(mutex)> lk(mutex);
      scale = new_scale;
      buffer_bundle->resize(logical_size * scale);
+ }
 === modified file 'src/server/compositor/multi_threaded_compositor.cpp'
 --- src/server/compositor/multi_threaded_compositor.cpp	2015-09-22 14:53:26 +0000
 +++ src/server/compositor/multi_threaded_compositor.cpp	2015-12-14 15:38:49 +0000
@@ -73,13 +73,6 @@
      void operator()() noexcept  // noexcept is important! (LP: #1237332)
      try
+     {
--        auto on_startup_failure = on_unwind(
--            [this]
--            {
--                if (started_future.wait_for(0s) != std::future_status::ready)
--                    started.set_exception(std::current_exception());
--            });
--
          mir::set_thread_name("Mir/Comp");
          std::vector<std::tuple<mg::DisplayBuffer*, std::unique_ptr<mc::DisplayBufferCompositor>>> compositors;
@@ -96,11 +89,13 @@
                                    CompositorReport::SubCompositorId{comp_id});
          });
++        //Appease TSan, avoid destructor and this thread accessing the same shared_ptr instance
++        auto const disp_listener = display_listener;
          auto display_registration = mir::raii::paired_calls(
--            [this]{group.for_each_display_buffer([this](mg::DisplayBuffer& buffer)
--                { display_listener->add_display(buffer.view_area()); });},
--            [this]{group.for_each_display_buffer([this](mg::DisplayBuffer& buffer)
--                { display_listener->remove_display(buffer.view_area()); });});
++            [this, &disp_listener]{group.for_each_display_buffer([&disp_listener](mg::DisplayBuffer& buffer)
++                { disp_listener->add_display(buffer.view_area()); });},
++            [this, &disp_listener]{group.for_each_display_buffer([&disp_listener](mg::DisplayBuffer& buffer)
++                { disp_listener->remove_display(buffer.view_area()); });});
          auto compositor_registration = mir::raii::paired_calls(
              [this,&compositors]
@@ -179,6 +174,16 @@
+     }
      catch(...)
+     {
++        try
++        {
++            //Move the promise so that the promise destructor occurs here rather than in the thread
++            //destroying CompositingFunctor, mostly to appease TSan
++            auto promise = std::move(started);
++            promise.set_exception(std::current_exception());
++        }
++        catch(...)
++        {
++        }
          mir::terminate_with_current_exception();
+     }
 === modified file 'src/server/scene/surface_stack.cpp'
 --- src/server/scene/surface_stack.cpp	2015-12-03 23:57:09 +0000
 +++ src/server/scene/surface_stack.cpp	2015-12-14 15:38:49 +0000
@@ -121,7 +121,8 @@
  ms::SurfaceStack::SurfaceStack(
      std::shared_ptr<SceneReport> const& report) :
--    report{report}
++    report{report},
++    scene_changed{false}
+ {
+ }
 === modified file 'src/server/scene/surface_stack.h'
 --- src/server/scene/surface_stack.h	2015-12-03 10:28:45 +0000
 +++ src/server/scene/surface_stack.h	2015-12-14 15:38:49 +0000
@@ -28,11 +28,12 @@
  #include "mir/basic_observers.h"
++#include <atomic>
++#include <map>
  #include <memory>
--#include <vector>
  #include <mutex>
--#include <map>
  #include <set>
++#include <vector>
  namespace mir
+ {
@@ -119,7 +120,7 @@
      std::vector<std::shared_ptr<graphics::Renderable>> overlays;
      Observers observers;
--    bool scene_changed = false;
++    std::atomic<bool> scene_changed;
  };
+ }
 === modified file 'tests/acceptance-tests/test_unresponsive_client.cpp'
 --- tests/acceptance-tests/test_unresponsive_client.cpp	2015-07-28 09:09:00 +0000
 +++ tests/acceptance-tests/test_unresponsive_client.cpp	2015-12-14 15:38:49 +0000
@@ -32,8 +32,9 @@
  #include <gtest/gtest.h>
  #include <gmock/gmock.h>
++#include <future>
++#include <mutex>
  #include <string>
--#include <future>
  namespace ms = mir::scene;
  namespace mt = mir::test;
@@ -45,19 +46,27 @@
+ {
  struct SessionListener : ms::NullSessionListener
+ {
++    ~SessionListener()
++    {
++        std::lock_guard<decltype(guard)> lk{guard};
++        sessions.clear();
++    }
++
      void starting(std::shared_ptr<ms::Session> const& session) override
--        { sessions.insert(session); }
++        { std::lock_guard<decltype(guard)> lk{guard}; sessions.insert(session); }
      void stopping(std::shared_ptr<ms::Session> const& session) override
--        { sessions.erase(session); }
++        { std::lock_guard<decltype(guard)> lk{guard}; sessions.erase(session); }
      void for_each(std::function<void(std::shared_ptr<ms::Session> const&)> f) const
+     {
++        std::lock_guard<decltype(guard)> lk{guard};
          for (auto& session : sessions)
              f(session);
+     }
  private:
++    std::mutex mutable guard;
      std::set<std::shared_ptr<ms::Session>> sessions;
  };
+ }
 === modified file 'tests/include/mir/test/stub_server_tool.h'
 --- tests/include/mir/test/stub_server_tool.h	2015-09-18 14:44:59 +0000
 +++ tests/include/mir/test/stub_server_tool.h	2015-12-14 15:38:49 +0000
@@ -44,8 +44,8 @@
          response->set_pixel_format(request->pixel_format());
          response->mutable_buffer()->set_buffer_id(22);
--        std::unique_lock<std::mutex> lock(guard);
--        surface_name = request->surface_name();
++        std::lock_guard<std::mutex> lock(guard);
++        surf_name = request->surface_name();
          wait_condition.notify_one();
          done->Run();
@@ -58,7 +58,8 @@
+     {
          response->set_buffer_id(22);
--        std::unique_lock<std::mutex> lock(guard);
++        std::lock_guard<std::mutex> lock(guard);
++        //FIXME: huh? What's the condition here?
          wait_condition.notify_one();
          done->Run();
+     }
@@ -78,7 +79,10 @@
          mir::protobuf::Connection* connect_msg,
          google::protobuf::Closure* done) override
+     {
--        app_name = request->application_name();
++        {
++            std::lock_guard<std::mutex> lock(guard);
++            app_name = request->application_name();
++        }
          // If you check out MirConnection::connected either the platform and display_configuration
          // have to be set or the error has to be set, otherwise we die and fail to callback.
          //
@@ -95,7 +99,8 @@
          mir::protobuf::Void* /*response*/,
          google::protobuf::Closure* done) override
+     {
--        std::unique_lock<std::mutex> lock(guard);
++        std::lock_guard<std::mutex> lock(guard);
++        //FIXME: huh? What's the condition here?
          wait_condition.notify_one();
          done->Run();
+     }
@@ -108,8 +113,20 @@
          done->Run();
+     }
--    std::mutex guard;
--    std::string surface_name;
++    std::string application_name() const
++    {
++        std::lock_guard<std::mutex> lock(guard);
++        return app_name;
++    }
++
++    std::string surface_name() const
++    {
++        std::lock_guard<std::mutex> lock(guard);
++        return surf_name;
++    }
++
++    std::mutex mutable guard;
++    std::string surf_name;
      std::condition_variable wait_condition;
      std::string app_name;
  };
 === modified file 'tests/include/mir/test/test_protobuf_client.h'
 --- tests/include/mir/test/test_protobuf_client.h	2015-09-18 14:44:59 +0000
 +++ tests/include/mir/test/test_protobuf_client.h	2015-12-14 15:38:49 +0000
@@ -26,8 +26,10 @@
  #include <gmock/gmock.h>
--#include <memory>
--#include <atomic>
++#include <condition_variable>
++#include <functional>
++#include <mutex>
++#include <string>
  namespace mir
+ {
@@ -63,69 +65,43 @@
      MOCK_METHOD0(connect_done, void());
      MOCK_METHOD0(create_surface_done, void());
      MOCK_METHOD0(next_buffer_done, void());
--    MOCK_METHOD0(exchange_buffer_done, void());
--    MOCK_METHOD0(release_surface_done, void());
      MOCK_METHOD0(disconnect_done, void());
      MOCK_METHOD0(display_configure_done, void());
--    MOCK_METHOD0(prompt_session_start_done, void());
--    MOCK_METHOD0(prompt_session_stop_done, void());
      void on_connect_done();
--
      void on_create_surface_done();
--
      void on_next_buffer_done();
--
--    void on_exchange_buffer_done();
--
--    void on_release_surface_done();
--
      void on_disconnect_done();
--
      void on_configure_display_done();
++    void connect();
++    void disconnect();
++    void create_surface();
++    void next_buffer();
++    void configure_display();
++
      void wait_for_connect_done();
--
      void wait_for_create_surface();
--
      void wait_for_next_buffer();
--
--    void wait_for_exchange_buffer();
--
--    void wait_for_release_surface();
--
      void wait_for_disconnect_done();
--
++    void wait_for_configure_display_done();
      void wait_for_surface_count(int count);
--    void wait_for_disconnect_count(int count);
--
--    void tfd_done();
--
--    void wait_for_tfd_done();
--
--    void wait_for_configure_display_done();
--
--    void wait_for_prompt_session_start_done();
--
--    void wait_for_prompt_session_stop_done();
--
--    const int maxwait;
--    std::atomic<bool> connect_done_called;
--    std::atomic<bool> create_surface_called;
--    std::atomic<bool> next_buffer_called;
--    std::atomic<bool> exchange_buffer_called;
--    std::atomic<bool> release_surface_called;
--    std::atomic<bool> disconnect_done_called;
--    std::atomic<bool> configure_display_done_called;
--    std::atomic<bool> tfd_done_called;
--
--    WaitCondition wc_prompt_session_start;
--    WaitCondition wc_prompt_session_stop;
--
--    std::atomic<int> connect_done_count;
--    std::atomic<int> create_surface_done_count;
--    std::atomic<int> disconnect_done_count;
++    void wait_for(std::function<bool()> const& predicate, std::string const& error_message);
++    void signal_condition(bool& condition);
++    void reset_condition(bool& condition);
++
++    int const maxwait;
++    bool connect_done_called;
++    bool create_surface_called;
++    bool next_buffer_called;
++    bool exchange_buffer_called;
++    bool disconnect_done_called;
++    bool configure_display_done_called;
++    int create_surface_done_count;
++
++    std::mutex mutable guard;
++    std::condition_variable cv;
  };
+ }
+ }
 === modified file 'tests/include/mir_test_framework/stub_input_platform.h'
 --- tests/include/mir_test_framework/stub_input_platform.h	2015-10-14 15:22:39 +0000
 +++ tests/include/mir_test_framework/stub_input_platform.h	2015-12-14 15:38:49 +0000
@@ -21,6 +21,7 @@
  #include "mir/input/platform.h"
  #include <atomic>
  #include <memory>
++#include <mutex>
  #include <vector>
  namespace mir
@@ -59,6 +60,7 @@
      std::shared_ptr<mir::input::InputDeviceRegistry> const registry;
      static std::atomic<StubInputPlatform*> stub_input_platform;
      static std::vector<std::weak_ptr<mir::input::InputDevice>> device_store;
++    static std::mutex device_store_guard;
  };
+ }
 === modified file 'tests/mir_test_doubles/test_protobuf_client.cpp'
 --- tests/mir_test_doubles/test_protobuf_client.cpp	2015-09-18 14:44:59 +0000
 +++ tests/mir_test_doubles/test_protobuf_client.cpp	2015-12-14 15:38:49 +0000
@@ -29,15 +29,16 @@
  #include "mir/dispatch/threaded_dispatcher.h"
  #include "mir/events/event_private.h"
++#include <boost/throw_exception.hpp>
++
++#include <stdexcept>
  #include <thread>
  namespace mtd = mir::test::doubles;
  namespace mclr = mir::client::rpc;
  namespace md = mir::dispatch;
--mir::test::TestProtobufClient::TestProtobufClient(
--    std::string socket_file,
--    int timeout_ms) :
++mir::test::TestProtobufClient::TestProtobufClient(std::string socket_file, int timeout_ms) :
      rpc_report(std::make_shared<testing::NiceMock<doubles::MockRpcReport>>()),
      channel(mclr::make_rpc_channel(
          socket_file,
@@ -54,13 +55,9 @@
      create_surface_called(false),
      next_buffer_called(false),
      exchange_buffer_called(false),
--    release_surface_called(false),
      disconnect_done_called(false),
      configure_display_done_called(false),
--    tfd_done_called(false),
--    connect_done_count(0),
--    create_surface_done_count(0),
--    disconnect_done_count(0)
++    create_surface_done_count(0)
+ {
      surface_parameters.set_width(640);
      surface_parameters.set_height(480);
@@ -76,177 +73,133 @@
          .WillByDefault(testing::Invoke(this, &TestProtobufClient::on_create_surface_done));
      ON_CALL(*this, next_buffer_done())
          .WillByDefault(testing::Invoke(this, &TestProtobufClient::on_next_buffer_done));
--    ON_CALL(*this, exchange_buffer_done())
--        .WillByDefault(testing::Invoke(this, &TestProtobufClient::on_exchange_buffer_done));
--    ON_CALL(*this, release_surface_done())
--        .WillByDefault(testing::Invoke(this, &TestProtobufClient::on_release_surface_done));
      ON_CALL(*this, disconnect_done())
          .WillByDefault(testing::Invoke(this, &TestProtobufClient::on_disconnect_done));
      ON_CALL(*this, display_configure_done())
          .WillByDefault(testing::Invoke(this, &TestProtobufClient::on_configure_display_done));
--    ON_CALL(*this, prompt_session_start_done())
--        .WillByDefault(testing::Invoke(&wc_prompt_session_start, &WaitCondition::wake_up_everyone));
--    ON_CALL(*this, prompt_session_stop_done())
--        .WillByDefault(testing::Invoke(&wc_prompt_session_stop, &WaitCondition::wake_up_everyone));
++}
++
++void mir::test::TestProtobufClient::signal_condition(bool& condition)
++{
++    std::lock_guard<decltype(guard)> lk{guard};
++    condition = true;
++    cv.notify_all();
++}
++
++void mir::test::TestProtobufClient::reset_condition(bool& condition)
++{
++    std::lock_guard<decltype(guard)> lk{guard};
++    condition = false;
+ }
  void mir::test::TestProtobufClient::on_connect_done()
+ {
--    connect_done_called.store(true);
--
--    auto old = connect_done_count.load();
--
--    while (!connect_done_count.compare_exchange_weak(old, old+1));
++    signal_condition(connect_done_called);
+ }
  void mir::test::TestProtobufClient::on_create_surface_done()
+ {
--    create_surface_called.store(true);
--
--    auto old = create_surface_done_count.load();
--
--    while (!create_surface_done_count.compare_exchange_weak(old, old+1));
++    std::lock_guard<decltype(guard)> lk{guard};
++    create_surface_called = true;
++    create_surface_done_count++;
++    cv.notify_all();
+ }
  void mir::test::TestProtobufClient::on_next_buffer_done()
+ {
--    next_buffer_called.store(true);
--}
--
--void mir::test::TestProtobufClient::on_exchange_buffer_done()
--{
--    exchange_buffer_called.store(true);
--}
--
--void mir::test::TestProtobufClient::on_release_surface_done()
--{
--    release_surface_called.store(true);
++    signal_condition(next_buffer_called);
+ }
  void mir::test::TestProtobufClient::on_disconnect_done()
+ {
--    disconnect_done_called.store(true);
--
--    auto old = disconnect_done_count.load();
--
--    while (!disconnect_done_count.compare_exchange_weak(old, old+1));
++    signal_condition(disconnect_done_called);
+ }
  void mir::test::TestProtobufClient::on_configure_display_done()
+ {
--    configure_display_done_called.store(true);
++    signal_condition(configure_display_done_called);
++}
++
++void mir::test::TestProtobufClient::connect()
++{
++    reset_condition(connect_done_called);
++    display_server.connect(
++        &connect_parameters,
++        &connection,
++        google::protobuf::NewCallback(this, &TestProtobufClient::connect_done));
++}
++
++void mir::test::TestProtobufClient::disconnect()
++{
++    reset_condition(disconnect_done_called);
++    display_server.disconnect(
++        &ignored,
++        &ignored,
++        google::protobuf::NewCallback(this, &TestProtobufClient::disconnect_done));
++}
++
++void mir::test::TestProtobufClient::create_surface()
++{
++    reset_condition(create_surface_called);
++    display_server.create_surface(
++        &surface_parameters,
++        &surface,
++        google::protobuf::NewCallback(this, &TestProtobufClient::create_surface_done));
++}
++
++void mir::test::TestProtobufClient::next_buffer()
++{
++    reset_condition(next_buffer_called);
++    display_server.next_buffer(
++        &surface.id(),
++        surface.mutable_buffer(),
++        google::protobuf::NewCallback(this, &TestProtobufClient::next_buffer_done));
++}
++
++void mir::test::TestProtobufClient::configure_display()
++{
++    reset_condition(configure_display_done_called);
++    display_server.configure_display(
++        &disp_config,
++        &disp_config_response,
++        google::protobuf::NewCallback(this, &TestProtobufClient::display_configure_done));
+ }
  void mir::test::TestProtobufClient::wait_for_configure_display_done()
+ {
--    for (int i = 0; !configure_display_done_called.load() && i < maxwait; ++i)
--    {
--        std::this_thread::sleep_for(std::chrono::milliseconds(1));
--        std::this_thread::yield();
--    }
--
--    configure_display_done_called.store(false);
++    wait_for([this]{ return configure_display_done_called; }, "Timed out waiting to configure display");
+ }
  void mir::test::TestProtobufClient::wait_for_connect_done()
+ {
--    for (int i = 0; !connect_done_called.load() && i < maxwait; ++i)
--    {
--        std::this_thread::sleep_for(std::chrono::milliseconds(1));
--        std::this_thread::yield();
--    }
--
--    connect_done_called.store(false);
++    wait_for([this]{ return connect_done_called; }, "Timed out waiting to connect");
+ }
  void mir::test::TestProtobufClient::wait_for_create_surface()
+ {
--    for (int i = 0; !create_surface_called.load() && i < maxwait; ++i)
--    {
--        std::this_thread::sleep_for(std::chrono::milliseconds(1));
--        std::this_thread::yield();
--    }
--    create_surface_called.store(false);
++    wait_for([this]{ return create_surface_called; }, "Timed out waiting create surface");
+ }
  void mir::test::TestProtobufClient::wait_for_next_buffer()
+ {
--    for (int i = 0; !next_buffer_called.load() && i < maxwait; ++i)
--    {
--        std::this_thread::sleep_for(std::chrono::milliseconds(1));
--        std::this_thread::yield();
--    }
--    next_buffer_called.store(false);
--}
--
--void mir::test::TestProtobufClient::wait_for_exchange_buffer()
--{
--    for (int i = 0; !exchange_buffer_called.load() && i < maxwait; ++i)
--    {
--        std::this_thread::sleep_for(std::chrono::milliseconds(1));
--        std::this_thread::yield();
--    }
--    exchange_buffer_called.store(false);
--}
--
--void mir::test::TestProtobufClient::wait_for_release_surface()
--{
--    for (int i = 0; !release_surface_called.load() && i < maxwait; ++i)
--    {
--        std::this_thread::sleep_for(std::chrono::milliseconds(1));
--        std::this_thread::yield();
--    }
--    release_surface_called.store(false);
++    wait_for([this] { return next_buffer_called; }, "Timed out waiting for next buffer");
+ }
  void mir::test::TestProtobufClient::wait_for_disconnect_done()
+ {
--    for (int i = 0; !disconnect_done_called.load() && i < maxwait; ++i)
--    {
--        std::this_thread::sleep_for(std::chrono::milliseconds(1));
--        std::this_thread::yield();
--    }
--    disconnect_done_called.store(false);
++    wait_for([this] { return disconnect_done_called; }, "Timed out waiting to disconnect");
+ }
  void mir::test::TestProtobufClient::wait_for_surface_count(int count)
+ {
--    for (int i = 0; count != create_surface_done_count.load() && i < 10000; ++i)
--    {
--        std::this_thread::sleep_for(std::chrono::milliseconds(10));
--        std::this_thread::yield();
--    }
--}
--
--void mir::test::TestProtobufClient::wait_for_disconnect_count(int count)
--{
--    for (int i = 0; count != disconnect_done_count.load() && i < 10000; ++i)
--    {
--        std::this_thread::sleep_for(std::chrono::milliseconds(10));
--        std::this_thread::yield();
--    }
--}
--
--void mir::test::TestProtobufClient::tfd_done()
--{
--    tfd_done_called.store(true);
--}
--
--void mir::test::TestProtobufClient::wait_for_tfd_done()
--{
--    for (int i = 0; !tfd_done_called.load() && i < maxwait; ++i)
--    {
--        std::this_thread::sleep_for(std::chrono::milliseconds(1));
--    }
--    tfd_done_called.store(false);
--}
--
--void mir::test::TestProtobufClient::wait_for_prompt_session_start_done()
--{
--    wc_prompt_session_start.wait_for_at_most_seconds(maxwait);
--}
--
--void mir::test::TestProtobufClient::wait_for_prompt_session_stop_done()
--{
--    wc_prompt_session_stop.wait_for_at_most_seconds(maxwait);
++    wait_for([this, count] { return create_surface_done_count == count; }, "Timed out waiting for surface count");
++}
++
++void mir::test::TestProtobufClient::wait_for(std::function<bool()> const& predicate, std::string const& error_message)
++{
++    std::unique_lock<decltype(guard)> lk{guard};
++    if (!cv.wait_for(lk, std::chrono::milliseconds(maxwait), predicate))
++    {
++        BOOST_THROW_EXCEPTION(std::runtime_error(error_message));
++    }
+ }
 === modified file 'tests/mir_test_framework/stub_input_platform.cpp'
 --- tests/mir_test_framework/stub_input_platform.cpp	2015-10-14 15:22:39 +0000
 +++ tests/mir_test_framework/stub_input_platform.cpp	2015-12-14 15:38:49 +0000
@@ -40,12 +40,14 @@
  mtf::StubInputPlatform::~StubInputPlatform()
+ {
++    std::lock_guard<decltype(device_store_guard)> lk{device_store_guard};
      device_store.clear();
      stub_input_platform = nullptr;
+ }
  void mtf::StubInputPlatform::start()
+ {
++    std::lock_guard<decltype(device_store_guard)> lk{device_store_guard};
      for (auto const& dev : device_store)
+     {
          auto device = dev.lock();
@@ -74,6 +76,7 @@
      auto input_platform = stub_input_platform.load();
      if (!input_platform)
+     {
++        std::lock_guard<decltype(device_store_guard)> lk{device_store_guard};
          device_store.push_back(dev);
          return;
+     }
@@ -118,3 +121,4 @@
  std::atomic<mtf::StubInputPlatform*> mtf::StubInputPlatform::stub_input_platform{nullptr};
  std::vector<std::weak_ptr<mir::input::InputDevice>> mtf::StubInputPlatform::device_store;
++std::mutex mtf::StubInputPlatform::device_store_guard;
 === modified file 'tests/unit-tests/client/test_client_mir_surface.cpp'
 --- tests/unit-tests/client/test_client_mir_surface.cpp	2015-12-08 15:02:49 +0000
 +++ tests/unit-tests/client/test_client_mir_surface.cpp	2015-12-14 15:38:49 +0000
@@ -98,7 +98,11 @@
          google::protobuf::Closure* done) override
+     {
          create_surface_response(response);
--        surface_name = request->surface_name();
++        {
++            std::lock_guard<std::mutex> lock(guard);
++            surf_name = request->surface_name();
++        }
++
          done->Run();
+     }
 === modified file 'tests/unit-tests/frontend/test_protobuf_reports_errors.cpp'
 --- tests/unit-tests/frontend/test_protobuf_reports_errors.cpp	2015-07-28 20:51:25 +0000
 +++ tests/unit-tests/frontend/test_protobuf_reports_errors.cpp	2015-12-14 15:38:49 +0000
@@ -81,7 +81,7 @@
+     {
          stub_services = std::make_shared<mt::ErrorServer>();
          server = std::make_shared<mt::TestProtobufServer>("./test_error_fixture", stub_services);
--        client = std::make_shared<mt::TestProtobufClient>("./test_error_fixture", 100);
++        client = std::make_shared<mt::TestProtobufClient>("./test_error_fixture", 10000);
          server->comm->start();
+     }
 === modified file 'tests/unit-tests/frontend/test_protobuf_surface_apis.cpp'
 --- tests/unit-tests/frontend/test_protobuf_surface_apis.cpp	2015-07-28 20:51:25 +0000
 +++ tests/unit-tests/frontend/test_protobuf_surface_apis.cpp	2015-12-14 15:38:49 +0000
@@ -85,7 +85,7 @@
          stub_server->comm->start();
--        stub_client = std::make_shared<mt::TestProtobufClient>(mtf::test_socket_file(), 100);
++        stub_client = std::make_shared<mt::TestProtobufClient>(mtf::test_socket_file(), 10000);
          stub_client->connect_parameters.set_application_name(__PRETTY_FUNCTION__);
+     }
@@ -136,11 +136,7 @@
      for (int i = 0; i != surface_count; ++i)
+     {
--        stub_client->display_server.create_surface(
--            &stub_client->surface_parameters,
--            &stub_client->surface,
--            google::protobuf::NewCallback(stub_client.get(), &mt::TestProtobufClient::create_surface_done));
--
++        stub_client->create_surface();
          stub_client->wait_for_create_surface();
+     }
@@ -182,7 +178,7 @@
          for(int i=0; i<number_of_clients; i++)
+         {
--            auto client_tmp = std::make_shared<mt::TestProtobufClient>(mtf::test_socket_file(), 100);
++            auto client_tmp = std::make_shared<mt::TestProtobufClient>(mtf::test_socket_file(), 10000);
              clients.push_back(client_tmp);
+         }
+     }
 === modified file 'tests/unit-tests/frontend/test_published_socket_connector.cpp'
 --- tests/unit-tests/frontend/test_published_socket_connector.cpp	2015-09-18 14:44:59 +0000
 +++ tests/unit-tests/frontend/test_published_socket_connector.cpp	2015-12-14 15:38:49 +0000
@@ -29,20 +29,21 @@
  #include "mir_protobuf.pb.h"
--#include "mir/test/doubles/stub_ipc_factory.h"
--#include "mir/test/doubles/stub_session_authorizer.h"
--#include "mir/test/doubles/mock_rpc_report.h"
  #include "mir/test/stub_server_tool.h"
  #include "mir/test/test_protobuf_client.h"
  #include "mir/test/test_protobuf_server.h"
++#include "mir/test/wait_condition.h"
++#include "mir/test/doubles/stub_ipc_factory.h"
++#include "mir/test/doubles/stub_session_authorizer.h"
++#include "mir/test/doubles/mock_rpc_report.h"
  #include "mir/test/doubles/stub_ipc_factory.h"
  #include "mir_test_framework/testing_server_configuration.h"
  #include <gtest/gtest.h>
  #include <gmock/gmock.h>
++#include <memory>
  #include <stdexcept>
--#include <memory>
  #include <string>
  namespace mf = mir::frontend;
@@ -139,11 +140,7 @@
+ {
      EXPECT_CALL(*client, create_surface_done()).Times(1);
--    client->display_server.create_surface(
--        &client->surface_parameters,
--        &client->surface,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::create_surface_done));
--
++    client->create_surface();
      client->wait_for_create_surface();
+ }
@@ -153,14 +150,10 @@
      client->connect_parameters.set_application_name(__PRETTY_FUNCTION__);
--    client->display_server.connect(
--        &client->connect_parameters,
--        &client->connection,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::connect_done));
--
++    client->connect();
      client->wait_for_connect_done();
--    EXPECT_EQ(__PRETTY_FUNCTION__, stub_server_tool->app_name);
++    EXPECT_EQ(__PRETTY_FUNCTION__, stub_server_tool->application_name());
+ }
  TEST_F(PublishedSocketConnector, create_surface_sets_surface_name)
@@ -170,36 +163,23 @@
      client->connect_parameters.set_application_name(__PRETTY_FUNCTION__);
--    client->display_server.connect(
--        &client->connect_parameters,
--        &client->connection,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::connect_done));
--
++    client->connect();
      client->wait_for_connect_done();
      client->surface_parameters.set_surface_name(__PRETTY_FUNCTION__);
--    client->display_server.create_surface(
--        &client->surface_parameters,
--        &client->surface,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::create_surface_done));
--
++    client->create_surface();
      client->wait_for_create_surface();
--    EXPECT_EQ(__PRETTY_FUNCTION__, stub_server_tool->surface_name);
++    EXPECT_EQ(__PRETTY_FUNCTION__, stub_server_tool->surface_name());
+ }
--
  TEST_F(PublishedSocketConnector,
          create_surface_results_in_a_surface_being_created)
+ {
      EXPECT_CALL(*client, create_surface_done()).Times(1);
--    client->display_server.create_surface(
--        &client->surface_parameters,
--        &client->surface,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::create_surface_done));
--
++    client->create_surface();
      client->wait_for_create_surface();
+ }
@@ -218,19 +198,11 @@
      using namespace testing;
      EXPECT_CALL(*client, create_surface_done()).Times(1);
--    client->display_server.create_surface(
--        &client->surface_parameters,
--        &client->surface,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::create_surface_done));
--
++    client->create_surface();
      client->wait_for_create_surface();
      EXPECT_CALL(*client, disconnect_done()).Times(1);
--    client->display_server.disconnect(
--        &client->ignored,
--        &client->ignored,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::disconnect_done));
--
++    client->disconnect();
      client->wait_for_disconnect_done();
      Mock::VerifyAndClearExpectations(client.get());
@@ -247,11 +219,7 @@
      try
+     {
--        client->display_server.disconnect(
--            &client->ignored,
--            &client->ignored,
--            google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::disconnect_done));
--
++        client->disconnect();
          // the write beat the socket closing
+     }
      catch (std::runtime_error const& x)
@@ -268,11 +236,7 @@
      EXPECT_CALL(*client, create_surface_done()).Times(testing::AtLeast(0));
      EXPECT_CALL(*client, disconnect_done()).Times(testing::AtLeast(0));
--    client->display_server.create_surface(
--        &client->surface_parameters,
--        &client->surface,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::create_surface_done));
--
++    client->create_surface();
      client->wait_for_create_surface();
      EXPECT_TRUE(client->surface.has_buffer());
@@ -280,20 +244,12 @@
      for (int i = 0; i != 8; ++i)
+     {
--        client->display_server.next_buffer(
--            &client->surface.id(),
--            client->surface.mutable_buffer(),
--            google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::next_buffer_done));
--
++        client->next_buffer();
          client->wait_for_next_buffer();
          EXPECT_TRUE(client->surface.has_buffer());
+     }
--    client->display_server.disconnect(
--        &client->ignored,
--        &client->ignored,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::disconnect_done));
--
++    client->disconnect();
      client->wait_for_disconnect_done();
+ }
@@ -301,19 +257,11 @@
         connect_create_surface_then_disconnect_a_session)
+ {
      EXPECT_CALL(*client, create_surface_done()).Times(1);
--    client->display_server.create_surface(
--        &client->surface_parameters,
--        &client->surface,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::create_surface_done));
--
++    client->create_surface();
      client->wait_for_create_surface();
      EXPECT_CALL(*client, disconnect_done()).Times(1);
--    client->display_server.disconnect(
--        &client->ignored,
--        &client->ignored,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::disconnect_done));
--
++    client->disconnect();
      client->wait_for_disconnect_done();
+ }
@@ -322,33 +270,22 @@
      using namespace testing;
      EXPECT_CALL(*client, create_surface_done()).Times(AnyNumber());
--    client->display_server.create_surface(
--        &client->surface_parameters,
--        &client->surface,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::create_surface_done));
--
++    client->create_surface();
      client->wait_for_create_surface();
--    std::mutex m;
--    std::condition_variable cv;
--    bool done = false;
++    mt::WaitCondition wc;
--    ON_CALL(*communicator_report, error(_)).WillByDefault(Invoke([&] (std::exception const&)
++    ON_CALL(*communicator_report, error(_)).WillByDefault(Invoke([&wc] (std::exception const&)
+         {
--            std::unique_lock<std::mutex> lock(m);
--
--            done = true;
--            cv.notify_all();
++            wc.wake_up_everyone();
          }));
      EXPECT_CALL(*communicator_report, error(_)).Times(1);
      client.reset();
--    std::unique_lock<std::mutex> lock(m);
--
--    auto const deadline = std::chrono::steady_clock::now() + std::chrono::seconds(1);
--    while (!done && cv.wait_until(lock, deadline) != std::cv_status::timeout);
++    wc.wait_for_at_most_seconds(1);
++    EXPECT_TRUE(wc.woken());
+ }
  TEST_F(PublishedSocketConnector, configure_display)
@@ -356,11 +293,7 @@
      EXPECT_CALL(*client, display_configure_done())
          .Times(1);
--    client->display_server.configure_display(
--        &client->disp_config,
--        &client->disp_config_response,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::display_configure_done));
--
++    client->configure_display();
      client->wait_for_configure_display_done();
+ }
@@ -373,20 +306,13 @@
      client->connect_parameters.set_application_name(__PRETTY_FUNCTION__);
      EXPECT_CALL(*client, connect_done()).Times(1);
--
--    client->display_server.connect(
--        &client->connect_parameters,
--        &client->connection,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::connect_done));
++    client->connect();
++    client->wait_for_connect_done();
      EXPECT_CALL(*client, create_surface_done()).Times(testing::AtLeast(0));
      EXPECT_CALL(*client, disconnect_done()).Times(testing::AtLeast(0));
--    client->display_server.create_surface(
--        &client->surface_parameters,
--        &client->surface,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::create_surface_done));
--
++    client->create_surface();
      client->wait_for_create_surface();
      EXPECT_TRUE(client->surface.has_buffer());
@@ -394,20 +320,12 @@
      for (int i = 0; i != next_buffer_calls; ++i)
+     {
--        client->display_server.next_buffer(
--            &client->surface.id(),
--            client->surface.mutable_buffer(),
--            google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::next_buffer_done));
--
++        client->next_buffer();
          client->wait_for_next_buffer();
          EXPECT_TRUE(client->surface.has_buffer());
+     }
--    client->display_server.disconnect(
--        &client->ignored,
--        &client->ignored,
--        google::protobuf::NewCallback(client.get(), &mt::TestProtobufClient::disconnect_done));
--
++    client->disconnect();
      client->wait_for_disconnect_done();
      EXPECT_EQ(__PRETTY_FUNCTION__, stub_server_tool->app_name);
 === modified file 'tests/unit-tests/thread/test_basic_thread_pool.cpp'
 --- tests/unit-tests/thread/test_basic_thread_pool.cpp	2015-06-25 03:00:08 +0000
 +++ tests/unit-tests/thread/test_basic_thread_pool.cpp	2015-12-14 15:38:49 +0000
@@ -22,6 +22,7 @@
  #include "mir/test/current_thread_name.h"
  #include "mir/test/signal.h"
++#include <atomic>
  #include <memory>
  #include <gmock/gmock.h>
@@ -100,7 +101,7 @@
+     }
  private:
--    bool called;
++    std::atomic<bool> called;
      bool block;
      std::string name;
      mt::Signal signal;