Merge lp:~michihenning/unity-scopes-api/async-create into lp:unity-scopes-api/devel

[Michi Henning]

The scope search, activate, perform_action, and preview methods now return immediately instead of waiting for a response from the server. The returned QueryCtrl is a "fake" one: if cancel() is called on the returned QueryCtrl, the cancellation is remembered and sent once the corresponding create method in the scope has completed.

This change should be transparent to application code (other than that these methods now can no longer block).

Also added interlocks to QueryObject, so cancel() and run() see consistent data.

Fixed a number of race conditions during shutdown.

The scope search, activate, perform_action, and preview methods now return immediately instead of waiting for a response from the server. The returned QueryCtrl is a "fake" one: if cancel() is called on the returned QueryCtrl, the cancellation is remembered and sent once the corresponding create method in the scope has completed.

This change should be transparent to application code (other than that these methods now can no longer block).

Also added interlocks to QueryObject, so cancel() and run() see consistent data.

Fixed a number of race conditions during shutdown.

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:283
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/321/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/322/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/321/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/321/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/321/rebuild

review: Needs Fixing (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:284
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/323/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/324
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/323
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/323/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/323

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/323/rebuild

review: Approve (continuous-integration)

[Paweł Stołowski (stolowski)]

Impressive stuff! I've to admit I don't fully grasp all the changes yet, need to give it another look tomorrow. Just a few minor nitpicks for now:

8 +Changes in version 0.4.1

devel is currently at 0.4.2, you may want to make it 0.4.3.

437 +class Socket final

I think it would be worth marking it as non-copyable; I know it's internal class, but still, makes it harder for us to do silly mistakes.

1939 + * Copyright (C) 2013 Canonical Ltd
2038 + * Copyright (C) 2013 Canonical Ltd

Please also fix it in the header template you use ;)

review: Needs Fixing

[Michi Henning (michihenning)]

> Impressive stuff! I've to admit I don't fully grasp all the changes yet, need
> to give it another look tomorrow.

Hold off with more review just yet please, this isn't ready for prime time as of now.

> Just a few minor nitpicks for now:
>
> 8 +Changes in version 0.4.1
>
> devel is currently at 0.4.2, you may want to make it 0.4.3.

Will do, thanks!

> 437 +class Socket final
>
> I think it would be worth marking it as non-copyable; I know it's internal
> class, but still, makes it harder for us to do silly mistakes.

It's non-copyable already because it defines a move constructor and move assignment operator, which suppresses the default-generated copy constructor and assignment operator.

> 1939 + * Copyright (C) 2013 Canonical Ltd
> 2038 + * Copyright (C) 2013 Canonical Ltd
>
> Please also fix it in the header template you use ;)

Thanks for that! :-) I don't use a header template, as rule. Instead, I just copy some similar header and hack it up. Need to get out of the habit of doing that…

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:287
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/406/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/407/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/406/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/406/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/406/rebuild

review: Needs Fixing (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:288
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/407/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/408/console
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/407
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/407/artifact/work/output/*zip*/output.zip
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/407/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/407/rebuild

review: Needs Fixing (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:289
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/408/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/409/console
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/408
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/408/artifact/work/output/*zip*/output.zip
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/408/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/408/rebuild

review: Needs Fixing (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:290
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/411/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/412
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/411
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/411/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/411

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/411/rebuild

review: Approve (continuous-integration)

[Michi Henning (michihenning)]

I think this now correctly does the async dispatch for search(), preview(), etc. These methods do not fail now. All errors are instead delivered to the finished() callback. The returned QueryCtrl proxy is a "fake" proxy: if cancel() is called on that proxy before the search method in the target scope has returned the real QueryCtrl, the cancel is remembered by the run time and delivered as soon as the real QueryCtrl arrives.

Michal, could you give this a spin please?

This doesn't not yet fix Marcus's problem where a finished() call and overtake a push(). Will work on that next.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:294
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/413/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/414
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/413
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/413/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/413

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/413/rebuild

review: Approve (continuous-integration)

[Marcus Tomlinson (marcustomlinson)]

Hi Michi, I noticed an issue with this after merging with my rebinding branch:

2709 + // If a request times out, we must close the corresponding socket, otherwise
2710 + // zmq gets confused: the reply will never be read, so the socket ends up
2711 + // in a bad state.
2712 + s.close();
2713 + throw TimeoutException("Request timed out after " + std::to_string(timeout) + " milliseconds");

I don't think we can just close the socket and leave it in the connection pool. We need to call "pool.remove(endpoint_)" otherwise we attempt to talk to a closed socket the next time around -which is what I was seeing during rebind: "error: Socket operation on non-socket".

Simply replacing "s.close()" with "pool.remove(endpoint_)" should be sufficient, as removing a socket from the connection pool deletes it, hence closing the socket.

review: Needs Fixing

[Marcus Tomlinson (marcustomlinson)]

117 + if (!valid())

rather than expecting every scope author to have to do this, could we not do this check internally from QueryObject::run() and SSQueryObject::run_query() before we call run()?

review: Needs Fixing

[Michi Henning (michihenning)]

> 117 + if (!valid())
>
> rather than expecting every scope author to have to do this, could we not do
> this check internally from QueryObject::run() and SSQueryObject::run_query()
> before we call run()?

Unfortunately, no. run() and cancel() are called by different threads. That's so I can guarantee that there is always a thread available to deliver a cancel message, even if run() is implemented synchronously and all run threads are currently in use. This also means that the cancel() for a run() invocation can be dispatched *before* run() is called. In the dispatch code for run, I have to do something like this:

unique_lock<mutex> lock(mutex_);
if (not query cancelled already) // Call run() only if the query wasn't cancelled
{
    lock.unlock();
    application_code->run();
}

The problem is that, between the unlock and the call to run(), the scheduler can suspend my thread, run the thread that processes an incoming cancel, and then run my thread again, meaning that, by the time run() gets into the application code, the query was cancelled already.

I can't call run() with the lock held because that would require the application code to unlock, and we can't rely on it doing that. (This would also create all sorts of deadlock scenarios.) Note that, even if I use an atomic_bool or some such, so the sequence is lock-free, the race remains: in between the test and the call to run(), the query can be cancelled.

The upshot of it all for the scope authors is:

- By the time the run() is called, the query may have been cancelled already. If so, valid() will return false. If it does, don't bother running the query; just return.

- The cancel may arrive at any time: before run() is called or while run() is still in progress, and the cancel() will always be delivered by a thread other than the one that called run(). If cancel() relies on state that is established by run(), the scope developer must use a full fence or mutex to make sure that cancel() operates on current state.

- The scope developer must write cancel() such that it finishes promptly.

- Always check the return value from push() and, if push() returns false, stop pushing and clean up. It is benign to ignore the return value from push(), but all the work done once push() returns false is wasted because the run time will simply throw the results that are pushed on the floor.

[lp:~michihenning/unity-scopes-api/async-create]

295. By Michi Henning on 2014-04-17

Fixed bug that would leave a bad socket in the connection pool after a
timeout exception. Added test for that, and improved test coverage
for ScopeExceptions.cpp.

[Michi Henning (michihenning)]

> I don't think we can just close the socket and leave it in the connection
> pool.

Awesome catch, thanks for that!

I fixed it and added a test to make sure it works now.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:295
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/420/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/421
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/420
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/420/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/420

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/420/rebuild

review: Approve (continuous-integration)

[Marcus Tomlinson (marcustomlinson)]

> > 117 + if (!valid())
> >
> > rather than expecting every scope author to have to do this, could we not do
> > this check internally from QueryObject::run() and SSQueryObject::run_query()
> > before we call run()?
>
> Unfortunately, no. run() and cancel() are called by different threads. That's
> so I can guarantee that there is always a thread available to deliver a cancel
> message, even if run() is implemented synchronously and all run threads are
> currently in use. This also means that the cancel() for a run() invocation can
> be dispatched *before* run() is called. In the dispatch code for run, I have
> to do something like this:
>
> unique_lock<mutex> lock(mutex_);
> if (not query cancelled already) // Call run() only if the query wasn't
> cancelled
> {
> lock.unlock();
> application_code->run();
> }
>
> The problem is that, between the unlock and the call to run(), the scheduler
> can suspend my thread, run the thread that processes an incoming cancel, and
> then run my thread again, meaning that, by the time run() gets into the
> application code, the query was cancelled already.

Ok I see what you're saying, but consider these 2 code snippets:
Is there really any difference between them?

QueryObject::run()
{
  ...
  if (scope.valid())
  {
    scope.run()
  }
  ...
}

Scope::run()
{
  if (!valid())
  {
    return;
  }
  ...
}

I'm suggesting the first option is a lot cleaner and give the scope author one less thing to remember. Sorry if I'm still missing the point.

[Michi Henning (michihenning)]

> QueryObject::run()
> {
> ...
> if (scope.valid())
> {
> scope.run()
> }
> ...
> }

This is what's effectively happening already. The problem is that between valid() returning true, and scope.run() being executed, the cancel can arrive. So, the point of valid() on the public API is to give a scope author a way to check whether the query was cancelled already before doing a lot of work setting up and running a query that, in the end, will return false from the first call to push().

Maybe it's me who's missing something? :-)

[lp:~michihenning/unity-scopes-api/async-create]

296. By Michi Henning on 2014-04-17

Improved documentation.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:296
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/423/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/424
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/423
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/423/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/423

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/423/rebuild

review: Approve (continuous-integration)

[Marcus Tomlinson (marcustomlinson)]

Well, I've read through this code a few times now with every effort to understand it all. As far as I can see, this looks really good!

Just a few little things I picked up:

366 + void set_proxy(MWProxy const& p); // Allows a derived proxy to replace mw_proxy_ for aynchronous twoway calls.

"aynchronous" should be "asynchronous"

2126 + stopper_.reset(new StopPublisher(mw_.context(), name_ + "-stoppper"));
4045 + "failed (endpoint: inproc://testscope-stoppper):\n"

"stoppper"? Was this intended?

3587 + // cancel is correcly forwarded to the scope once the real reply proxy arrives

"correcly" should be "correctly"

review: Needs Fixing

[lp:~michihenning/unity-scopes-api/async-create]

297. By Michi Henning on 2014-04-17

Fixed typos.

[Marcus Tomlinson (marcustomlinson)]

4045 + "failed (endpoint: inproc://testscope-stoppper):\n"

forgot this one ;)

review: Needs Fixing

[lp:~michihenning/unity-scopes-api/async-create]

298. By Michi Henning on 2014-04-17

Fixed another typo.

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:297
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/425/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/426/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/425/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/425/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/425/rebuild

review: Needs Fixing (continuous-integration)

[Marcus Tomlinson (marcustomlinson)]

I'm happy with this.

review: Approve

[lp:~michihenning/unity-scopes-api/async-create]

299. By Michi Henning on 2014-04-17

Increased wait time between tests due to "address in use" failures when running on Jenkins.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:298
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/426/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/427
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/426
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/426/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/426

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/426/rebuild

review: Approve (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:299
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/427/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/428
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/427
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/427/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/427

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/427/rebuild

review: Approve (continuous-integration)

[Michal Hruby (mhr3)]

179 + if (--num_children_ == 0)
180 + {
181 + upstream_->finished(); // Send finished once last child finishes
182 + }

Why is this needed suddenly? Wasn't the point of create_subquery to take care of that?

review: Needs Information

[Michal Hruby (mhr3)]

Also, still managed to get it to hang on shutdown by running the client binary (from demo/ dir), here's the state of the threads:

  Id Target Id Frame
  11 Thread 0xb4c21410 (LWP 8173) "client" 0xb6b655f4 in __libc_do_syscall
    () from /lib/arm-linux-gnueabihf/libpthread.so.0
  10 Thread 0xb42ff410 (LWP 8174) "client" 0xb6b655f4 in __libc_do_syscall
    () from /lib/arm-linux-gnueabihf/libpthread.so.0
  9 Thread 0xb3aff410 (LWP 8175) "client" 0xb6b655f4 in __libc_do_syscall
    () from /lib/arm-linux-gnueabihf/libpthread.so.0
  8 Thread 0xb30ff410 (LWP 8176) "client" 0xb6b655f4 in __libc_do_syscall
    () from /lib/arm-linux-gnueabihf/libpthread.so.0
  7 Thread 0xb28ff410 (LWP 8177) "client" 0xb6b655f4 in __libc_do_syscall
    () from /lib/arm-linux-gnueabihf/libpthread.so.0
  6 Thread 0xb20ff410 (LWP 8178) "client" 0xb6d4f032 in epoll_wait ()
   from /lib/arm-linux-gnueabihf/libc.so.6
  5 Thread 0xb18ff410 (LWP 8179) "client" 0xb6d4f032 in epoll_wait ()
   from /lib/arm-linux-gnueabihf/libc.so.6
  4 Thread 0xb08ff410 (LWP 8181) "client" 0xb6b655f4 in __libc_do_syscall
    () from /lib/arm-linux-gnueabihf/libpthread.so.0
  3 Thread 0xb00ff410 (LWP 8182) "client" 0xb6d45d22 in poll ()
   from /lib/arm-linux-gnueabihf/libc.so.6
  2 Thread 0xaf8ff410 (LWP 8183) "client" 0xb6d45d22 in poll ()
   from /lib/arm-linux-gnueabihf/libc.so.6
* 1 Thread 0xb4c59000 (LWP 8172) "client" 0xb6b655f4 in __libc_do_syscall
    () from /lib/arm-linux-gnueabihf/libpthread.so.0

Thread 11 (Thread 0xb4c21410 (LWP 8173)):
#0 0xb6b655f4 in __libc_do_syscall ()
   from /lib/arm-linux-gnueabihf/libpthread.so.0
#1 0xb6b611d8 in pthread_cond_wait@@GLIBC_2.4 ()
   from /lib/arm-linux-gnueabihf/libpthread.so.0
#2 0xb6e425f8 in std::condition_variable::wait(std::unique_lock<std::mutex>&)
    () from /usr/lib/arm-linux-gnueabihf/libstdc++.so.6
#3 0xb6f31e92 in wait<unity::scopes::internal::ThreadSafeQueue<T>::wait_and_pop() [with T = unity::scopes::internal::TaskWrapper]::__lambda1> (__p=...,
    __lock=..., this=0x3b7af0) at /usr/include/c++/4.8/condition_variable:93
#4 unity::scopes::internal::ThreadSafeQueue<unity::scopes::internal::TaskWrapper>::wait_and_pop (this=0x3b7ab0)
    at /home/phablet/build-area/unity-scopes-api-0.4.2+14.04.20140404.2/include/unity/scopes/internal/ThreadSafeQueue.h:135
#5 0xb6f30c42 in unity::scopes::internal::ThreadPool::run (this=0x3b7a00)
    at /home/phablet/build-area/unity-scopes-api-0.4.2+14.04.20140404.2/src/scopes/internal/ThreadPool.cpp:102
#6 0xb6e44a2c in ?? () from /usr/lib/arm-linux-gnueabihf/libstdc++.so.6
Backtrace stopped: previous frame identical to this frame (corrupt stack?)

Thread 10 (Thread 0xb42ff410 (LWP 8174)):
#0 0xb6b655f4 in __libc_do_syscall ()
   from /lib/arm-linux-gnueabihf/libpthread.so.0
#1 0xb6b63268 in __lll_lock_wait ()
   from /lib/arm-linux-gnueabihf/libpthread.so.0
#2 0xb6b5f8d8 in pthread_mutex_lock ()
   from /lib/arm-linux-gnueabihf/libpthread.so.0
#3 0xb6f1944a in __gthread_mutex_lock (__mutex=0x3b055c)
    at /usr/include/arm-linux-gnueabihf/c++/4.8/bits/gthr-default.h:748
#4 lock (this=0x3b055c) at /usr/include/c++/4.8/mutex:134
#5 lock_gua...

[Michal Hruby (mhr3)]

As a note, I was trying this from the all-in-one rebinding-logic branch, so the line numbers are valid for that one.

[Marcus Tomlinson (marcustomlinson)]

> As a note, I was trying this from the all-in-one rebinding-logic branch, so
> the line numbers are valid for that one.

I can confirm that this hang occurs with just this branch on its own (without the serialize-oneway-invoke and rebinding_logic changes) :(

review: Needs Fixing

[Marcus Tomlinson (marcustomlinson)]

Despite the fact that I missed that hang issue, I will stick with my Approved stance on this branch. I probably won't be around when this is completed, so Michal's review will take preference anyway.

review: Approve

[Paweł Stołowski (stolowski)]

I'm having issues with running demo/client against real scopes on the phone with this branch (I've replaced libunity-scopes on the phone with the one from this branch, symlinked system-wide Runtime.ini inside demo/ dir): the client receives metadata only, doesn't receive any results and the query finishes with no errors. Interestingly, Dash on the phone seems to be running just fine with this branch.

Michal tested this branch on the phone and desktop and he had same issue on the phone, but all worked fine on the desktop.

I repeated this test with pristine devel branch (also on the phone) and it worked just fine, so perhaps there is something wrong in the client code that breaks with async change, although I couldn't spot anything.

Could you please have a look at the client code or/and try to reproduce?

review: Needs Information

[Michi Henning (michihenning)]

> 179 + if (--num_children_ == 0)
> 180 + {
> 181 + upstream_->finished(); // Send finished once last child
> finishes
> 182 + }
>
> Why is this needed suddenly? Wasn't the point of create_subquery to take care
> of that?

No, not really. The sub-query thing takes care of transparently passing cancellation down to the child scopes. However, the aggregator still has to make a decision as to when a query is complete. It isn't necessarily the *first* finished() call that arrives. Rather, if the aggregator sends the query to three child scopes, and wants to consider the query finished once the last of these three scopes finishes, it needs to keep track of the number of finished() calls, which is what the above change is doing. (This was broken all along.)

[Michal Hruby (mhr3)]

> > Why is this needed suddenly? Wasn't the point of create_subquery to take
> care
> > of that?
>
> No, not really. The sub-query thing takes care of transparently passing
> cancellation down to the child scopes.

Right, subsearch doesn't magically make it work, but...

> ... However, the aggregator still has to
> make a decision as to when a query is complete.

...this decision was implicit up until now - subsearch() took a SearchListenerBase instance that kept the upstream ReplyProxy alive, and only when the subsearches finished was the last reference to the upstream ReplyProxy destroyed which sent the finished message.

So something must be broke if this is needed now.

[Paweł Stołowski (stolowski)]

> I'm having issues with running demo/client against real scopes on the phone
> with this branch (I've replaced libunity-scopes on the phone with the one from
> this branch, symlinked system-wide Runtime.ini inside demo/ dir): the client
> receives metadata only, doesn't receive any results and the query finishes
> with no errors. Interestingly, Dash on the phone seems to be running just fine
> with this branch.
>
> Michal tested this branch on the phone and desktop and he had same issue on
> the phone, but all worked fine on the desktop.
>
> I repeated this test with pristine devel branch (also on the phone) and it
> worked just fine, so perhaps there is something wrong in the client code that
> breaks with async change, although I couldn't spot anything.
>
> Could you please have a look at the client code or/and try to reproduce?

Some more info about how I tested this:

1) I've compiled a checkout of async-create with dpkg-buildpackage on the phone, then installed the resulting .deb package and rebooted the phone.

2) Inside ~/async-create/obj-arm-linux-gnueabihf/demo (where the source code & build output reside), I've created a symlink to system-wide Runtime.ini.

phablet@ubuntu-phablet:~/async-create/obj-arm-linux-gnueabihf/demo$ ls -l
total 696
drwxrwxr-x 4 phablet phablet 4096 Apr 17 12:27 click
-rwxrwxr-x 1 phablet phablet 667320 Apr 17 12:59 client
drwxrwxr-x 3 phablet phablet 4096 Apr 17 12:27 CMakeFiles
-rw-rw-r-- 1 phablet phablet 1570 Apr 17 12:27 cmake_install.cmake
-rw-rw-r-- 1 phablet phablet 351 Apr 17 12:27 CTestTestfile.cmake
-rw-rw-r-- 1 phablet phablet 7477 Apr 17 12:27 Makefile
-rw-rw-r-- 1 phablet phablet 488 Apr 17 12:27 Registry.ini
lrwxrwxrwx 1 phablet phablet 39 Apr 17 13:11 Runtime.ini -> /usr/share/unity-scopes-api/Runtime.ini
-rw-rw-r-- 1 phablet phablet 236 Apr 17 12:27 Runtime.ini.bak
drwxrwxr-x 9 phablet phablet 4096 Apr 17 12:27 scopes
drwxrwxr-x 3 phablet phablet 4096 Apr 17 12:59 stand-alone
-rw-rw-r-- 1 phablet phablet 59 Apr 17 12:27 Zmq.ini
phablet@ubuntu-phablet:~/async-create/obj-arm-linux-gnueabihf/demo$

3) Now, from within ~/async-create/obj-arm-linux-gnueabihf/demo I run ./client against phone's scopes, e.g.
./client list
./client clickscope a
./client clickscope ""

for any scope and query, I'm only getting scope's metadata and no results:

./client clickscope ""
Scope metadata:
        scope_id: clickscope
        display_name: Apps
        description: Scope for searching the click app store
        art: /usr/share/unity/scopes/clickscope//clickscope-screenshot.jpg
        icon: /usr/share/unity/scopes/clickscope//apps-scope.svg
        search_hint: clickscope.SearchHint
        hot_key: clickscope.HotKey
client: created query
client: wait returned
query complete, status: finished

[Michi Henning (michihenning)]

> Some more info about how I tested this:

Thanks for that! After struggling with building on the phone some more, I could reproduce. (ccache is your friend when it comes to building on the bloody thing...)

The ultimate cause of the problem was this:

120.5.10 <email address hidden> 20140116 | Receiver(int index_to_save)
                                                 | : index_to_save_(index_to_save),
120.6.9 <email address hidden> 20140110 | push_result_count_(0)
                                                 | {
                                                 | }

This constructor was added at some point. The class originally had a default constructor which, after this change, was redundant, but not removed. Unlike the original constructor, the new constructor failed to initialize the query_complete_ member variable. It turns out that, on the device, the corresponding boolean was inititalized to crap (meaning true), causing wait_until_finished() to return immediately. In turn, that caused the client to shut down the run time before the results trickled in from the scope so, unsurprisingly, those results were lost. The finished() call that still seemed to arrive was in fact local and triggered by the shutdown of the run time which, when all remaining ReplyObjects are removed from the object adapter, calls finished(), so it doesn't leave any outstanding unfinished queries behind.

I haven't had a chance to look at the hang during shutdown yet. Will do that next.

[lp:~michihenning/unity-scopes-api/async-create]

300. By Michi Henning on 2014-04-23

Fixed broken wait for completion: member variable was not initialized,
causing the client to shut down the run time before the results arrived
from the scope. Fixed the other receiver classes to match.

[Paweł Stołowski (stolowski)]

Thanks for spotting and fixing the issue! Now, there is just one conflict in release notes that needs fixing.

review: Needs Fixing

[Michal Hruby (mhr3)]

+ the shutdown issue and the ReplyProxy issue.

review: Needs Fixing

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:300
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/430/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/431/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/430/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/430/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/430/rebuild

review: Needs Fixing (continuous-integration)

[lp:~michihenning/unity-scopes-api/async-create]

301. By Michi Henning on 2014-04-23

Merged devel and resolved conflicts. Set version back to 0.4.2
because addition of valid() to QueryBase is ABI compatible.

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:301
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/437/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/438
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/437
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/437/artifact/work/output/*zip*/output.zip
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/437/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/437/rebuild

review: Needs Fixing (continuous-integration)

[lp:~michihenning/unity-scopes-api/async-create]

302. By Michi Henning on 2014-04-24

Debug changes, unfinished.

303. By Michi Henning on 2014-04-28

Fixed incorrect scope of task in ThreadPool::run(): the destructor of the task
was called only once the next task was started, instead of being called
as soon as the task completed.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:303
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/438/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/439
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/438
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/438/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/438

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/438/rebuild

review: Approve (continuous-integration)

[lp:~michihenning/unity-scopes-api/async-create]

304. By Michi Henning on 2014-04-28

Fixed incorrect count of threads in run() method and added a test for throwing tasks.

305. By Michi Henning on 2014-04-28

Tightened ThreadPool exception test.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:304
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/439/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/440
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/439
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/439/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/439

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/439/rebuild

review: Approve (continuous-integration)

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:305
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/440/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/441
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/440
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/440/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/440

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/440/rebuild

review: Approve (continuous-integration)

[lp:~michihenning/unity-scopes-api/async-create]

306. By Michi Henning on 2014-04-28

Changed micro version back to 3 after all because there was mention of that in debian/changelog already.
Better safe than sorry.

[Michi Henning (michihenning)]

> + the shutdown issue and the ReplyProxy issue.

They are both fixed now.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:306
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-ci/442/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-amd64-ci/443
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/442
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-armhf-ci/442/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-devel-trusty-i386-ci/442

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-devel-ci/442/rebuild

review: Approve (continuous-integration)

[Paweł Stołowski (stolowski)]

> > + the shutdown issue and the ReplyProxy issue.
>
> They are both fixed now.

Awesome, thanks for all this! +1 from me.

review: Approve

[Michal Hruby (mhr3)]

+1, seems to work great, although the threading code is getting pretty scary, but I guess we need to live with that.

review: Approve

=== modified file 'RELEASE_NOTES.txt'
--- RELEASE_NOTES.txt	2014-04-11 12:07:36 +0000
+++ RELEASE_NOTES.txt	2014-04-28 05:57:56 +0000
@@ -1,8 +1,15 @@
 Release Notes for unity-scopes-api
 ==================================
 
-Changes in version 0.4.3
+Changes in version 0.4.2
 ========================
+
+  * Made the scope search, activate, perform_action, and preview methods non-blocking.
+    A (fake) QueryCtrl is returned immediately from these methods now. Calling cancel() before
+    the server has finished creating the query remembers the cancel and sends it to the
+    server once the server has returned the real QueryCtrl. This change should be transparent
+    to application code (the only difference being that these methods complete faster now).
+
   * CannedQuery class can now be converted to and from a scopes:// uri with to_uri() and from_uri() methods.
     These methods replace to_string() and from_string() methods that got removed.
 

=== modified file 'debian/libunity-scopes1.symbols'
--- debian/libunity-scopes1.symbols	2014-04-17 10:31:23 +0000
+++ debian/libunity-scopes1.symbols	2014-04-28 05:57:56 +0000
@@ -285,6 +285,7 @@
  (c++)"unity::scopes::Runtime::run_scope(unity::scopes::ScopeBase*, std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&)@Base" 0.4.2+14.04.20140404.2
  (c++)"unity::scopes::Runtime::Runtime(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&)@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Runtime::~Runtime()@Base" 0.4.0+14.04.20140312.1
+ (c++)"unity::scopes::Variant::deserialize_json(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&)@Base" 0replaceme
  (c++)"unity::scopes::Variant::null()@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Variant::swap(unity::scopes::Variant&)@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Variant::Variant(unity::scopes::Variant&&)@Base" 0.4.0+14.04.20140312.1
@@ -454,7 +455,11 @@
  (c++)"unity::scopes::internal::StateReceiverObject::push_state(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unity::scopes::internal::StateReceiverObject::State const&)@Base" 0.4.2+14.04.20140404.2
  (c++)"unity::scopes::internal::StateReceiverObject::StateReceiverObject()@Base" 0.4.2+14.04.20140404.2
  (c++)"unity::scopes::internal::StateReceiverObject::~StateReceiverObject()@Base" 0.4.2+14.04.20140404.2
+ (c++)"unity::scopes::internal::Executor::exec(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::vector<std::basic_string<char, std::char_traits<char>, std::allocator<char> >, std::allocator<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > const&, std::map<std::basic_string<char, std::char_traits<char>, std::allocator<char> >, std::basic_string<char, std::char_traits<char>, std::allocator<char> >, std::less<std::basic_string<char, std::char_traits<char>, std::allocator<char> > >, std::allocator<std::pair<std::basic_string<char, std::char_traits<char>, std::allocator<char> > const, std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > > const&, core::posix::StandardStream const&)@Base" 0replaceme
+ (c++)"unity::scopes::internal::Executor::Executor()@Base" 0replaceme
+ (c++)"unity::scopes::internal::Executor::~Executor()@Base" 0replaceme
  (c++)"unity::scopes::internal::ScopeImpl::perform_action(unity::scopes::Result const&, unity::scopes::ActionMetadata const&, std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::shared_ptr<unity::scopes::ActivationListenerBase> const&)@Base" 0.4.0+14.04.20140312.1
+ (c++)"unity::scopes::internal::ScopeImpl::fwd()@Base" 0replaceme
  (c++)"unity::scopes::internal::ScopeImpl::create(std::shared_ptr<unity::scopes::internal::MWScope> const&, unity::scopes::internal::RuntimeImpl*, std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&)@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::internal::ScopeImpl::search(unity::scopes::CannedQuery const&, unity::scopes::SearchMetadata const&, std::shared_ptr<unity::scopes::SearchListenerBase> const&)@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::internal::ScopeImpl::search(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, unity::scopes::FilterState const&, unity::scopes::SearchMetadata const&, std::shared_ptr<unity::scopes::SearchListenerBase> const&)@Base" 0.4.0+14.04.20140312.1
@@ -575,14 +580,13 @@
  (c++)"unity::scopes::Runtime::registry() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Variant::get_double() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Variant::get_string() const@Base" 0.4.0+14.04.20140312.1
+ (c++)"unity::scopes::Variant::serialize_json() const@Base" 0replaceme
  (c++)"unity::scopes::Variant::which() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Variant::get_int() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Variant::is_null() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Variant::get_bool() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Variant::get_dict() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Variant::get_array() const@Base" 0.4.0+14.04.20140312.1
- (c++)"unity::scopes::Variant::deserialize_json(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&)@Base" 0.4.2+14.04.20140404.2
- (c++)"unity::scopes::Variant::serialize_json() const@Base" 0.4.2+14.04.20140404.2
  (c++)"unity::scopes::Variant::operator==(unity::scopes::Variant const&) const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Variant::operator<(unity::scopes::Variant const&) const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::testing::ScopeMetadataBuilder::operator()() const@Base" 0.4.0+14.04.20140312.1
@@ -599,11 +603,11 @@
  (c++)"unity::scopes::Category::icon() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Category::title() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::Category::serialize() const@Base" 0.4.0+14.04.20140312.1
- (c++)"unity::scopes::internal::Executor::exec(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::vector<std::basic_string<char, std::char_traits<char>, std::allocator<char> >, std::allocator<std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > const&, std::map<std::basic_string<char, std::char_traits<char>, std::allocator<char> >, std::basic_string<char, std::char_traits<char>, std::allocator<char> >, std::less<std::basic_string<char, std::char_traits<char>, std::allocator<char> > >, std::allocator<std::pair<std::basic_string<char, std::char_traits<char>, std::allocator<char> > const, std::basic_string<char, std::char_traits<char>, std::allocator<char> > > > > const&, core::posix::StandardStream const&)@Base" 0replaceme
- (c++)"unity::scopes::internal::Executor::Executor()@Base" 0replaceme
- (c++)"unity::scopes::internal::Executor::~Executor()@Base" 0replaceme
+ (c++)"unity::scopes::internal::RuntimeImpl::async_pool() const@Base" 0replaceme
  (c++)"unity::scopes::internal::RuntimeImpl::configfile() const@Base" 0.4.2+14.04.20140404.2
+ (c++)"unity::scopes::internal::RuntimeImpl::future_queue() const@Base" 0replaceme
  (c++)"unity::scopes::internal::RuntimeImpl::reply_reaper() const@Base" 0.4.0+14.04.20140312.1
+ (c++)"unity::scopes::internal::RuntimeImpl::waiter_thread(std::shared_ptr<unity::scopes::internal::ThreadSafeQueue<std::future<void> > > const&) const@Base" 0replaceme
  (c++)"unity::scopes::internal::RuntimeImpl::proxy_to_string(std::shared_ptr<unity::scopes::Object> const&) const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::internal::RuntimeImpl::string_to_proxy(std::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::internal::RuntimeImpl::registry_endpoint() const@Base" 0.4.0+14.04.20140312.1
@@ -637,10 +641,10 @@
  (c++)"unity::scopes::internal::RuntimeConfig::registry_configfile() const@Base" 0.4.2+14.04.20140404.2
  (c++)"unity::scopes::internal::RuntimeConfig::default_middleware_configfile() const@Base" 0.4.2+14.04.20140404.2
  (c++)"unity::scopes::internal::MiddlewareBase::runtime() const@Base" 0.4.0+14.04.20140312.1
- (c++)"unity::scopes::internal::RegistryConfig::click_installdir() const@Base" 0replaceme
  (c++)"unity::scopes::internal::RegistryConfig::endpointdir() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::internal::RegistryConfig::mw_configfile() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::internal::RegistryConfig::oem_installdir() const@Base" 0.4.0+14.04.20140312.1
+ (c++)"unity::scopes::internal::RegistryConfig::click_installdir() const@Base" 0replaceme
  (c++)"unity::scopes::internal::RegistryConfig::scope_installdir() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::internal::RegistryConfig::scoperunner_path() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::internal::RegistryConfig::ss_registry_endpoint() const@Base" 0.4.0+14.04.20140312.1
@@ -671,7 +675,7 @@
  (c++)"unity::scopes::internal::ScopeMetadataImpl::invisible() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::internal::ScopeMetadataImpl::serialize() const@Base" 0.4.0+14.04.20140312.1
  (c++)"unity::scopes::internal::StateReceiverObject::state_received() const@Base" 0.4.2+14.04.20140404.2
- (c++)"unity::scopes::internal::ScopeImpl::fwd() const@Base" 0.4.0+14.04.20140312.1
+ (c++)"unity::scopes::QueryBase::valid() const@Base" 0replaceme
  (c++)"unity::scopes::ScopeBase::scope_directory() const@Base" 0.4.2+14.04.20140404.2
  (c++)"typeinfo for unity::scopes::Annotation@Base" 0.4.0+14.04.20140312.1
  (c++)"typeinfo for unity::scopes::FilterBase@Base" 0.4.0+14.04.20140312.1

=== modified file 'demo/client.cpp'
--- demo/client.cpp	2014-04-23 11:33:49 +0000
+++ demo/client.cpp	2014-04-28 05:57:56 +0000
@@ -122,8 +122,9 @@
 {
 public:
     Receiver(int index_to_save)
-        : index_to_save_(index_to_save),
-          push_result_count_(0)
+        : query_complete_(false),
+          push_result_count_(0),
+          index_to_save_(index_to_save)
     {
     }
 
@@ -212,23 +213,23 @@
         return push_result_count_;
     }
 
-    Receiver() :
-        query_complete_(false)
-    {
-    }
-
 private:
     bool query_complete_;
+    int push_result_count_ = 0;
     int index_to_save_;
-    mutex mutex_;
-    condition_variable condvar_;
-    int push_result_count_ = 0;
     std::shared_ptr<Result> saved_result_;
+    mutex mutex_;
+    condition_variable condvar_;
 };
 
 class ActivationReceiver : public ActivationListenerBase
 {
 public:
+    ActivationReceiver()
+        : query_complete_(false)
+    {
+    }
+
     void activated(ActivationResponse const& response) override
     {
         cout << "\tGot activation response: " << response.status() << endl;
@@ -237,16 +238,19 @@
     void finished(Reason r, std::string const& error_message)
     {
         cout << "\tActivation finished, reason: " << r << ", error_message: " << error_message << endl;
+        lock_guard<decltype(mutex_)> lock(mutex_);
+        query_complete_ = true;
         condvar_.notify_one();
     }
 
     void wait_until_finished()
     {
         unique_lock<decltype(mutex_)> lock(mutex_);
-        condvar_.wait(lock);
+        condvar_.wait(lock, [this](){ return query_complete_; });
     }
 
 private:
+    bool query_complete_;
     mutex mutex_;
     condition_variable condvar_;
 };
@@ -254,6 +258,11 @@
 class PreviewReceiver : public PreviewListenerBase
 {
 public:
+    PreviewReceiver()
+        : query_complete_(false)
+    {
+    }
+
     void push(ColumnLayoutList const& columns) override
     {
         cout << "\tGot column layouts:" << endl;
@@ -297,17 +306,20 @@
 
     void finished(Reason r, std::string const& error_message) override
     {
+        lock_guard<decltype(mutex_)> lock(mutex_);
         cout << "\tPreview finished, reason: " << r << ", error_message: " << error_message << endl;
+        query_complete_ = true;
         condvar_.notify_one();
     }
 
     void wait_until_finished()
     {
         unique_lock<decltype(mutex_)> lock(mutex_);
-        condvar_.wait(lock);
+        condvar_.wait(lock, [this](){ return query_complete_; });
     }
 
 private:
+    bool query_complete_;
     mutex mutex_;
     condition_variable condvar_;
 };
@@ -434,9 +446,7 @@
         SearchMetadata metadata("C", "desktop");
         metadata.set_cardinality(10);
         auto ctrl = meta.proxy()->search(search_string, metadata, reply); // May raise TimeoutException
-        cout << "client: created query" << endl;
         reply->wait_until_finished();
-        cout << "client: wait returned" << endl;
 
         // handle activation
         if (result_index > 0)

=== modified file 'demo/scopes/scope-A/scope-A.cpp'
--- demo/scopes/scope-A/scope-A.cpp	2014-02-27 23:20:56 +0000
+++ demo/scopes/scope-A/scope-A.cpp	2014-04-28 05:57:56 +0000
@@ -31,13 +31,16 @@
 class MyQuery : public SearchQueryBase
 {
 public:
-    MyQuery(CannedQuery const& query) :
+    MyQuery(string const& scope_id, CannedQuery const& query) :
+        scope_id_(scope_id),
         query_(query)
     {
+        cerr << scope_id_ << ": query instance for \"" << query.query_string() << "\" created" << endl;
     }
 
     ~MyQuery()
     {
+        cerr << scope_id_ << ": query instance for \"" << query_.query_string() << "\" destroyed" << endl;
     }
 
     virtual void cancelled() override
@@ -46,6 +49,11 @@
 
     virtual void run(SearchReplyProxy const& reply) override
     {
+        if (!valid())
+        {
+            return;  // Query was cancelled
+        }
+
         DepartmentList departments({{"news", query_, "News", {{"news-world", query_, "World"}, {"news-europe", query_, "Europe"}}},
                                     {"sport", query_, "Sport"}});
         reply->register_departments(departments);
@@ -56,7 +64,10 @@
         filter->add_option("2", "Option 2");
         filters.push_back(filter);
         FilterState filter_state; // TODO: push real state from query obj
-        reply->push(filters, filter_state);
+        if (!reply->push(filters, filter_state))
+        {
+            return;  // Query was cancelled
+        }
 
         CategoryRenderer rdr;
         auto cat = reply->register_category("cat1", "Category 1", "", rdr);
@@ -65,30 +76,37 @@
         res.set_title("scope-A: result 1 for query \"" + query_.query_string() + "\"");
         res.set_art("icon");
         res.set_dnd_uri("dnd_uri");
-        reply->push(res);
+        if (!reply->push(res))
+        {
+            return;  // Query was cancelled
+        }
 
         CannedQuery q("scope-A", query_.query_string(), "");
         Annotation annotation(Annotation::Type::Link);
         annotation.add_link("More...", q);
         reply->register_annotation(annotation);
 
-        cout << "scope-A: query \"" << query_.query_string() << "\" complete" << endl;
+        cerr << "scope-A: query \"" << query_.query_string() << "\" complete" << endl;
     }
 
 private:
+    string scope_id_;
     CannedQuery query_;
 };
 
 class MyPreview : public PreviewQueryBase
 {
 public:
-    MyPreview(string const& uri) :
+    MyPreview(string const& scope_id, string const& uri) :
+        scope_id_(scope_id),
         uri_(uri)
     {
+        cerr << scope_id_ << ": preview instance for \"" << uri << "\" created" << endl;
     }
 
     ~MyPreview()
     {
+        cerr << scope_id_ << ": preview instance for \"" << uri_ << "\" destroyed" << endl;
     }
 
     virtual void cancelled() override
@@ -119,21 +137,32 @@
         layout3col.add_column({"rating"});
 
         reply->register_layout({layout1col, layout2col, layout3col});
-        reply->push(widgets);
-        reply->push("author", Variant("Foo"));
-        reply->push("rating", Variant("4 blah"));
-        cout << "scope-A: preview for \"" << uri_ << "\" complete" << endl;
+        if (!reply->push(widgets))
+        {
+            return;  // Query was cancelled
+        }
+        if (!reply->push("author", Variant("Foo")))
+        {
+            return;  // Query was cancelled
+        }
+        if (!reply->push("rating", Variant("4 blah")))
+        {
+            return;  // Query was cancelled
+        }
+        cerr << "scope-A: preview for \"" << uri_ << "\" complete" << endl;
     }
 
 private:
+    string scope_id_;
     string uri_;
 };
 
 class MyScope : public ScopeBase
 {
 public:
-    virtual int start(string const&, RegistryProxy const&) override
+    virtual int start(string const& scope_id, RegistryProxy const&) override
     {
+        scope_id_ = scope_id;
         return VERSION;
     }
 
@@ -141,17 +170,19 @@
 
     virtual SearchQueryBase::UPtr search(CannedQuery const& q, SearchMetadata const&) override
     {
-        SearchQueryBase::UPtr query(new MyQuery(q));
-        cout << "scope-A: created query: \"" << q.query_string() << "\"" << endl;
+        SearchQueryBase::UPtr query(new MyQuery(scope_id_, q));
         return query;
     }
 
     virtual PreviewQueryBase::UPtr preview(Result const& result, ActionMetadata const&) override
     {
-        PreviewQueryBase::UPtr preview(new MyPreview(result.uri()));
-        cout << "scope-A: created previewer: \"" << result.uri() << "\"" << endl;
+        PreviewQueryBase::UPtr preview(new MyPreview(scope_id_, result.uri()));
+        cerr << "scope-A: created previewer: \"" << result.uri() << "\"" << endl;
         return preview;
     }
+
+private:
+    string scope_id_;
 };
 
 extern "C"

=== modified file 'demo/scopes/scope-B/scope-B.cpp'
--- demo/scopes/scope-B/scope-B.cpp	2014-04-03 12:57:25 +0000
+++ demo/scopes/scope-B/scope-B.cpp	2014-04-28 05:57:56 +0000
@@ -47,12 +47,12 @@
 public:
     virtual void push(Category::SCPtr category) override
     {
-        cout << "received category: id=" << category->id() << endl;
+        cerr << scope_id_ << ": received category: id=" << category->id() << endl;
     }
 
     virtual void push(CategorisedResult result) override
     {
-        cout << "received result from " << scope_id_ << ": " << result.uri() << ", " << result.title() << endl;
+        cerr << scope_id_ << ": received result: " << result.uri() << ", " << result.title() << endl;
         try
         {
             result.set_category(upstream_->lookup_category("catB"));
@@ -60,18 +60,18 @@
         }
         catch (const unity::InvalidArgumentException &e)
         {
-            cerr << "error pushing result: " << e.what() << endl;
+            cerr << scope_id_ << ": error pushing result: " << e.what() << endl;
         }
     }
 
     virtual void finished(Reason reason, string const& error_message) override
     {
-        cout << "query to " << scope_id_ << " complete, status: " << to_string(reason);
+        cerr << scope_id_ << ": subquery complete, status: " << to_string(reason);
         if (reason == ListenerBase::Error)
         {
-            cout << ": " << error_message;
+            cerr << ": " << error_message;
         }
-        cout << endl;
+        cerr << endl;
     }
 
     Receiver(string const& scope_id, SearchReplyProxy const& upstream) :
@@ -101,11 +101,16 @@
 
     virtual void cancelled()
     {
-        cout << "query to " << scope_id_ << " was cancelled" << endl;
+        cerr << scope_id_ << ": query " << query_.query_string() << " was cancelled" << endl;
     }
 
     virtual void run(SearchReplyProxy const& upstream_reply)
     {
+        if (!valid())
+        {
+            return;  // Query was cancelled
+        }
+
         // note, category id must mach categories received from scope C and D, otherwise result pushing will fail.
         try
         {
@@ -157,13 +162,13 @@
     virtual SearchQueryBase::UPtr search(CannedQuery const& q, SearchMetadata const&) override
     {
         SearchQueryBase::UPtr query(new MyQuery(scope_id_, q, scope_c_, scope_d_));
-        cout << "scope-B: created query: \"" << q.query_string() << "\"" << endl;
+        cerr << "scope-B: created query: \"" << q.query_string() << "\"" << endl;
         return query;
     }
 
     virtual PreviewQueryBase::UPtr preview(Result const& result, ActionMetadata const&) override
     {
-        cout << "scope-B: preview: \"" << result.uri() << "\"" << endl;
+        cerr << "scope-B: preview: \"" << result.uri() << "\"" << endl;
         return nullptr;
     }
 

=== modified file 'demo/scopes/scope-C/scope-C.cpp'
--- demo/scopes/scope-C/scope-C.cpp	2014-04-11 12:04:41 +0000
+++ demo/scopes/scope-C/scope-C.cpp	2014-04-28 05:57:56 +0000
@@ -129,16 +129,17 @@
 class MyQuery : public SearchQueryBase
 {
 public:
-    MyQuery(CannedQuery const& query, Queue& queue) :
+    MyQuery(string const& scope_id, CannedQuery const& query, Queue& queue) :
+        scope_id_(scope_id),
         query_(query),
         queue_(queue)
     {
-        cerr << "My Query created" << endl;
+        cerr << scope_id_ << ": query instance for \"" << query.query_string() << "\" created" << endl;
     }
 
     ~MyQuery()
     {
-        cerr << "My Query destroyed" << endl;
+        cerr << scope_id_ << ": query instance for \"" << query_.query_string() << "\" destroyed" << endl;
     }
 
     virtual void cancelled() override
@@ -148,16 +149,21 @@
         // query, the worker thread's next call to push() will return false,
         // causing the worker thread to stop working on this query.
         queue_.remove(this);
-        cerr << "scope-C: \"" + query_.to_uri() + "\" cancelled" << endl;
+        cerr << scope_id_ << ": \"" + query_.to_uri() + "\" cancelled" << endl;
     }
 
     virtual void run(SearchReplyProxy const& reply) override
     {
+        if (!valid())
+        {
+            return;  // Query was cancelled
+        }
+
         queue_.put(this, query_.query_string(), reply);
-        cerr << "scope-C: run() returning" << endl;
     }
 
 private:
+    string scope_id_;
     CannedQuery query_;
     Queue& queue_;
 };
@@ -205,7 +211,7 @@
             {
                 CategorisedResult result(cat);
                 result.set_uri("uri");
-                result.set_title("scope-C: result " + to_string(i) + " for query \"" + query + "\"");
+                result.set_title(scope_id_ + ": result " + to_string(i) + " for query \"" + query + "\"");
                 result.set_art("icon");
                 result.set_dnd_uri("dnd_uri");
                 result.set_intercept_activation();
@@ -215,24 +221,24 @@
                 }
                 sleep(1);
             }
+            cerr << scope_id_ << ": query \"" << query << "\" complete" << endl;
         }
     }
 
     virtual SearchQueryBase::UPtr search(CannedQuery const& q, SearchMetadata const&) override
     {
-        cout << scope_id_ << ": created query: \"" << q.query_string() << "\"" << endl;
-        return SearchQueryBase::UPtr(new MyQuery(q, queue));
+        return SearchQueryBase::UPtr(new MyQuery(scope_id_, q, queue));
     }
 
     virtual ActivationQueryBase::UPtr activate(Result const& result, ActionMetadata const& /* hints */) override
     {
-        cout << scope_id_ << ": activate: \"" << result.uri() << "\"" << endl;
+        cerr << scope_id_ << ": activate: \"" << result.uri() << "\"" << endl;
         return ActivationQueryBase::UPtr(new MyActivation());
     }
 
     virtual PreviewQueryBase::UPtr preview(Result const& result, ActionMetadata const&) override
     {
-        cout << "scope-C: preview: \"" << result.uri() << "\"" << endl;
+        cerr << scope_id_ << ": preview: \"" << result.uri() << "\"" << endl;
         return nullptr;
     }
 

=== modified file 'demo/scopes/scope-D/scope-D.cpp'
--- demo/scopes/scope-D/scope-D.cpp	2014-04-03 12:57:25 +0000
+++ demo/scopes/scope-D/scope-D.cpp	2014-04-28 05:57:56 +0000
@@ -132,12 +132,12 @@
         query_(query),
         queue_(queue)
     {
-        cerr << "query instance for \"" << scope_id_ << ":" << query.query_string() << "\" created" << endl;
+        cerr << scope_id_ << ": query instance for \"" << query.query_string() << "\" created" << endl;
     }
 
     virtual ~MyQuery()
     {
-        cerr << "query instance for \"" << scope_id_ << ":" << query_.query_string() << "\" destroyed" << endl;
+        cerr << scope_id_ << ": query instance for \"" << query_.query_string() << "\" destroyed" << endl;
     }
 
     virtual void cancelled() override
@@ -151,11 +151,16 @@
         // work may still be in progress on a query. Note that cancellations are frequent;
         // not responding to cancelled() correctly causes loss of performance.
 
-        cerr << "query for \"" << scope_id_ << ":" << query_.query_string() << "\" cancelled" << endl;
+        cerr << scope_id_ << ": query for \"" << query_.query_string() << "\" cancelled" << endl;
     }
 
     virtual void run(SearchReplyProxy const& reply) override
     {
+        if (!valid())
+        {
+            return;  // Query was cancelled
+        }
+
         // The query can do anything it likes with this method, that is, run() can push results
         // directly on the provided reply, or it can save the reply for later use and return from
         // run(). It is OK to push results on the reply from a different thread.
@@ -226,13 +231,12 @@
     virtual SearchQueryBase::UPtr search(CannedQuery const& q, SearchMetadata const&) override
     {
         SearchQueryBase::UPtr query(new MyQuery(scope_id_, q, queue_));
-        cerr << scope_id_ << ": created query: \"" << q.query_string() << "\"" << endl;
         return query;
     }
 
     virtual PreviewQueryBase::UPtr preview(Result const& result, ActionMetadata const&) override
     {
-        cout << scope_id_ << ": preview: \"" << result.uri() << "\"" << endl;
+        cerr << scope_id_ << ": preview: \"" << result.uri() << "\"" << endl;
         return nullptr;
     }
 

=== modified file 'demo/scopes/scope-N/scope-N.cpp'
--- demo/scopes/scope-N/scope-N.cpp	2014-02-27 23:20:56 +0000
+++ demo/scopes/scope-N/scope-N.cpp	2014-04-28 05:57:56 +0000
@@ -32,32 +32,44 @@
 class MyQuery : public SearchQueryBase
 {
 public:
-    MyQuery()
+    MyQuery(string const& scope_id)
+        : scope_id_(scope_id)
     {
+        cerr << scope_id_ << ": query instance created" << endl;
     }
 
     ~MyQuery()
     {
+        cerr << scope_id_ << ": query instance destroyed" << endl;
     }
 
     virtual void cancelled() override
     {
-        cerr << "scope-no-op: received cancel request" << endl;
+        cerr << scope_id_ << ": received cancel request" << endl;
     }
 
     virtual void run(SearchReplyProxy const&) override
     {
-        cerr << "scope-no-op: received query" << endl;
+        if (!valid())
+        {
+            return;  // Query was cancelled
+        }
+
+        cerr << scope_id_ << ": received query" << endl;
         this_thread::sleep_for(chrono::seconds(3));
-        cerr << "scope-no-op: query complete" << endl;
+        cerr << scope_id_ << ": query complete" << endl;
     }
+
+private:
+    string scope_id_;
 };
 
 class MyScope : public ScopeBase
 {
 public:
-    virtual int start(string const&, RegistryProxy const&) override
+    virtual int start(string const& scope_id, RegistryProxy const&) override
     {
+        scope_id_ = scope_id;
         return VERSION;
     }
 
@@ -65,13 +77,16 @@
 
     virtual SearchQueryBase::UPtr search(CannedQuery const&, SearchMetadata const&) override
     {
-        return SearchQueryBase::UPtr(new MyQuery);
+        return SearchQueryBase::UPtr(new MyQuery(scope_id_));
     }
 
     virtual PreviewQueryBase::UPtr preview(Result const&, ActionMetadata const&) override
     {
         return nullptr;
     }
+
+private:
+    string scope_id_;
 };
 
 extern "C"

=== modified file 'demo/scopes/scope-S/scope-S.cpp'
--- demo/scopes/scope-S/scope-S.cpp	2014-03-04 03:15:11 +0000
+++ demo/scopes/scope-S/scope-S.cpp	2014-04-28 05:57:56 +0000
@@ -36,36 +36,49 @@
 class MyQuery : public SearchQueryBase
 {
 public:
-    MyQuery(CannedQuery const& query, CategoryRenderer const& renderer) :
+    MyQuery(string const& scope_id, CannedQuery const& query, CategoryRenderer const& renderer) :
+        scope_id_(scope_id),
         query_(query),
         renderer_(renderer)
     {
-        cerr << "MyQuery/" << query.query_string() << " created" << endl;
+        cerr << scope_id_ << ": query instance for \"" << query.query_string() << "\" created" << endl;
     }
 
     ~MyQuery()
     {
-        cerr << "MyQuery/" << query_.query_string() << " destroyed" << endl;
+        cerr << scope_id_ << ": query instance for \"" << query_.query_string() << "\" destroyed" << endl;
     }
 
     virtual void cancelled() override
     {
-        cerr << "MyQuery/" << query_.query_string() << " cancelled" << endl;
+        cerr << scope_id_ << ": \"" + query_.to_uri() + "\" cancelled" << endl;
     }
 
     virtual void run(SearchReplyProxy const& reply) override
     {
-        cerr << "scope-slow: run called for \"" << query_.query_string() << "\"" << endl;
-        this_thread::sleep_for(chrono::seconds(20));
+        if (!valid())
+        {
+            return;  // Query was cancelled
+        }
+
+        cerr << scope_id_ << ": run called for \"" << query_.query_string() << "\"" << endl;
+        int const short_secs = 5;
+        cerr << scope_id_ << ": sleeping for " << short_secs << " seconds" << endl;
+        this_thread::sleep_for(chrono::seconds(short_secs));
         auto cat = reply->register_category("cat1", "Category 1", "", renderer_);
         CategorisedResult result(cat);
         result.set_uri("uri");
-        result.set_title("scope-slow: result 1 for query \"" + query_.query_string() + "\"");
+        result.set_title(scope_id_ + ": result 1 for query \"" + query_.query_string() + "\"");
+        cerr << scope_id_ << ": pushing result" << endl;
         reply->push(result);
-        cout << "scope-slow: query \"" << query_.query_string() << "\" complete" << endl;
+        int const long_secs = 50;
+        cerr << scope_id_ << ": sleeping for " << long_secs << " seconds" << endl;
+        this_thread::sleep_for(chrono::seconds(long_secs));
+        cout << scope_id_ << ": query \"" << query_.query_string() << "\" complete" << endl;
     }
 
 private:
+    string scope_id_;
     CannedQuery query_;
     CategoryRenderer renderer_;
 };
@@ -73,35 +86,29 @@
 class MyScope : public ScopeBase
 {
 public:
-    virtual int start(string const&, RegistryProxy const&) override
+    virtual int start(string const& scope_id, RegistryProxy const&) override
     {
+        scope_id_ = scope_id;
         return VERSION;
     }
 
     virtual void stop() override {}
 
-    virtual SearchQueryBase::UPtr search(CannedQuery const& q, SearchMetadata const& hints) override
+    virtual SearchQueryBase::UPtr search(CannedQuery const& q, SearchMetadata const& /* hints */) override
     {
-        SearchQueryBase::UPtr query(new MyQuery(q, renderer_));
-        cout << "scope-slow: created query: \"" << q.query_string() << "\"" << endl;
-
-        if (hints.cardinality() > 0)
-        {
-            cerr << "result cardinality: " << hints.cardinality() << endl;
-        }
-
-        cerr << "locale: " << hints.locale() << endl;
-
+        SearchQueryBase::UPtr query(new MyQuery(scope_id_, q, renderer_));
+        cout << scope_id_ << ": created query: \"" << q.query_string() << "\"" << endl;
         return query;
     }
 
     virtual PreviewQueryBase::UPtr preview(Result const& result, ActionMetadata const&) override
     {
-        cout << "scope-S: preview: \"" << result.uri() << "\"" << endl;
+        cout << scope_id_ << ": preview: \"" << result.uri() << "\"" << endl;
         return nullptr;
     }
 
 private:
+    string scope_id_;
     CategoryRenderer renderer_;
 };
 

=== modified file 'include/unity/scopes/PreviewQueryBase.h'
--- include/unity/scopes/PreviewQueryBase.h	2014-03-05 02:42:49 +0000
+++ include/unity/scopes/PreviewQueryBase.h	2014-04-28 05:57:56 +0000
@@ -76,7 +76,7 @@
 
     \param reply The proxy on which to push results for the preview.
     */
-    virtual void run(PreviewReplyProxy const& reply) = 0;         // Called by the run time to start this query
+    virtual void run(PreviewReplyProxy const& reply) = 0;         // Called by the run time to start this preview
 
     // TODO: Add a method for subpreview request?
 

=== modified file 'include/unity/scopes/QueryBase.h'
--- include/unity/scopes/QueryBase.h	2014-03-10 05:51:39 +0000
+++ include/unity/scopes/QueryBase.h	2014-04-28 05:57:56 +0000
@@ -72,10 +72,28 @@
     Your implementation of this method should ensure that the scope stops
     processing the current query as soon as possible. Any calls to a `push()` method
     once a query is cancelled are ignored, so continuing to push after cancellation
-    only wastes CPU cycles.
+    only wastes CPU cycles. (`push()` returns `false` once a query is cancelled or
+    exceeds its cardinality limit.)
     */
     virtual void cancelled() = 0;                          // Originator cancelled the query
 
+    /**
+    \brief Check whether this query is still valid.
+
+    valid() returns false if this query is finished or was cancelled earlier. Note that it is possible
+    that the run time may call SearchQueryBase::run(), ActivationQueryBase::activate(), or PreviewQueryBase::run()
+    \a after cancelled() was called. Your implementation of these methods should check whether the query is still
+    valid and, if not, do nothing.
+
+    This method is provided mainly for convenience: it can be used in your s `run()` or `activate()` implementation to
+    avoid doing a lot of work setting up a query that was cancelled earlier. Note that, because cancellation
+    can happen at any time during query execution, your implementation should always test the return value
+    of `push()`. If `push()` returns `false`, the query was either cancelled or exceeded its cardinality limit.
+    Either way, there is no point in continuing to push more results because, once `push()` returns `false`,
+    the scopes run time discards all subsequent results for the query.
+    */
+    bool valid() const;
+
     /// @cond
     virtual ~QueryBase();
     /// @endcond

=== modified file 'include/unity/scopes/internal/ObjectImpl.h'
--- include/unity/scopes/internal/ObjectImpl.h	2014-04-03 12:57:25 +0000
+++ include/unity/scopes/internal/ObjectImpl.h	2014-04-28 05:57:56 +0000
@@ -22,6 +22,8 @@
 #include<unity/scopes/internal/MWObjectProxyFwd.h>
 #include<unity/scopes/Object.h>
 
+#include <mutex>
+
 namespace unity
 {
 
@@ -31,7 +33,7 @@
 namespace internal
 {
 
-class ObjectImpl : public virtual Object
+class ObjectImpl : public virtual Object, public virtual std::enable_shared_from_this<ObjectImpl>
 {
 public:
     ObjectImpl(MWProxy const& mw_proxy);
@@ -48,10 +50,17 @@
     virtual void ping();
 
 protected:
-    MWProxy proxy() const;                  // Non-virtual because we cannot use covariance with incomplete types.
-                                            // Each derived proxy implements a non-virtual fwd() method
-                                            // that is called from within each operation to down-cast the MWProxy.
+    MWProxy proxy();                   // Non-virtual because we cannot use covariance with incomplete types.
+                                       // Each derived proxy implements a non-virtual fwd() method
+                                       // that is called from within each operation to down-cast the MWProxy.
+
+    void set_proxy(MWProxy const& p);  // Allows a derived proxy to replace mw_proxy_ for asynchronous twoway calls.
+
     MWProxy mw_proxy_;
+    std::mutex proxy_mutex_;           // Protects mw_proxy_
+
+private:
+    void check_proxy();                // Throws from operations if mw_proxy_ is null
 };
 
 } // namespace internal

=== modified file 'include/unity/scopes/internal/QueryBaseImpl.h'
--- include/unity/scopes/internal/QueryBaseImpl.h	2014-03-10 04:17:09 +0000
+++ include/unity/scopes/internal/QueryBaseImpl.h	2014-04-28 05:57:56 +0000
@@ -63,10 +63,13 @@
 
     void cancel();
     void set_metadata(SearchMetadata const& metadata);
+    bool valid() const;
 
 private:
     SearchMetadata::UPtr search_metadata_;
     std::vector<QueryCtrlProxy> subqueries_;
+    bool valid_;
+    mutable std::mutex mutex_;
 };
 
 } // namespace internal

=== modified file 'include/unity/scopes/internal/QueryCtrlImpl.h'
--- include/unity/scopes/internal/QueryCtrlImpl.h	2014-03-07 02:18:16 +0000
+++ include/unity/scopes/internal/QueryCtrlImpl.h	2014-04-28 05:57:56 +0000
@@ -33,11 +33,7 @@
 namespace internal
 {
 
-// Proxy used by a scope to push results for a query.
-// To indicate that it has sent the last result, the scope must call finished().
-// Subsequent calls to finished() are ignored.
-// Calls to push() after finished() was called are ignored.
-// If the proxy goes out of scope before finished was called, it implicitly calls finished().
+class ScopeImpl;
 
 class QueryCtrlImpl : public virtual unity::scopes::QueryCtrl, public virtual ObjectImpl
 {
@@ -47,10 +43,17 @@
 
     virtual void cancel() override;
 
+    void set_proxy(MWQueryCtrlProxy const& p);
+
 private:
-    MWQueryCtrlProxy fwd() const;
+    MWQueryCtrlProxy fwd();
 
     MWReplyProxy reply_proxy_;
+    bool ready_;                // True once ObjectImpl::set_proxy() was called
+    bool cancelled_;            // True if cancel() is called before set_proxy() was called
+    std::mutex mutex_;          // Protects ready_ and cancelled_
+
+    friend class ScopeImpl;     // Allows ScopeImpl to call set_proxy()
 };
 
 } // namespace internal

=== modified file 'include/unity/scopes/internal/QueryObject.h'
--- include/unity/scopes/internal/QueryObject.h	2014-03-04 06:26:21 +0000
+++ include/unity/scopes/internal/QueryObject.h	2014-04-28 05:57:56 +0000
@@ -24,8 +24,6 @@
 #include <unity/scopes/internal/MWQueryCtrlProxyFwd.h>
 #include <unity/scopes/ReplyProxyFwd.h>
 
-#include <atomic>
-
 namespace unity
 {
 
@@ -71,9 +69,10 @@
     MWReplyProxy reply_;
     std::weak_ptr<unity::scopes::Reply> reply_proxy_;
     MWQueryCtrlProxy const ctrl_;
-    std::atomic_bool pushable_;
+    bool pushable_;
     QueryObjectBase::SPtr self_;
-    std::atomic_int cardinality_;
+    int cardinality_;
+    mutable std::mutex mutex_;
 };
 
 } // namespace internal

=== modified file 'include/unity/scopes/internal/RegistryImpl.h'
--- include/unity/scopes/internal/RegistryImpl.h	2014-04-03 12:57:25 +0000
+++ include/unity/scopes/internal/RegistryImpl.h	2014-04-28 05:57:56 +0000
@@ -48,7 +48,7 @@
     ObjectProxy locate(std::string const& identity);
 
 private:
-    MWRegistryProxy fwd() const;
+    MWRegistryProxy fwd();
 };
 
 } // namespace internal

=== modified file 'include/unity/scopes/internal/ReplyImpl.h'
--- include/unity/scopes/internal/ReplyImpl.h	2014-03-07 05:56:22 +0000
+++ include/unity/scopes/internal/ReplyImpl.h	2014-04-28 05:57:56 +0000
@@ -54,7 +54,7 @@
 protected:
     bool push(VariantMap const& variant_map);
 
-    MWReplyProxy fwd() const;
+    MWReplyProxy fwd();
 
 private:
     std::shared_ptr<QueryObjectBase> qo_;

=== modified file 'include/unity/scopes/internal/ReplyObject.h'
--- include/unity/scopes/internal/ReplyObject.h	2014-03-07 01:07:28 +0000
+++ include/unity/scopes/internal/ReplyObject.h	2014-04-28 05:57:56 +0000
@@ -59,7 +59,7 @@
     void finished(ListenerBase::Reason reason, std::string const& error_message) noexcept override;
 
 private:
-    ListenerBase::SPtr const listener_base_;
+    ListenerBase::SPtr listener_base_;
     ReapItem::SPtr reap_item_;
     std::atomic_bool finished_;
     std::mutex mutex_;

=== modified file 'include/unity/scopes/internal/RuntimeImpl.h'
--- include/unity/scopes/internal/RuntimeImpl.h	2014-04-03 14:22:02 +0000
+++ include/unity/scopes/internal/RuntimeImpl.h	2014-04-28 05:57:56 +0000
@@ -22,10 +22,9 @@
 #include <unity/scopes/internal/MiddlewareBase.h>
 #include <unity/scopes/internal/MiddlewareFactory.h>
 #include <unity/scopes/internal/Reaper.h>
+#include <unity/scopes/internal/ThreadPool.h>
 #include <unity/scopes/Runtime.h>
 
-#include <atomic>
-
 namespace unity
 {
 
@@ -53,6 +52,8 @@
     std::string registry_endpointdir() const;
     std::string registry_endpoint() const;
     Reaper::SPtr reply_reaper() const;
+    ThreadPool::SPtr async_pool() const;
+    ThreadSafeQueue<std::future<void>>::SPtr future_queue() const;
     void run_scope(ScopeBase *const scope_base, std::string const &scope_ini_file);
 
     ObjectProxy string_to_proxy(std::string const& s) const;
@@ -62,8 +63,9 @@
 
 private:
     RuntimeImpl(std::string const& scope_id, std::string const& configfile);
+    void waiter_thread(ThreadSafeQueue<std::future<void>>::SPtr const& queue) const noexcept;
 
-    std::atomic_bool destroyed_;
+    bool destroyed_;
     std::string scope_id_;
     std::string configfile_;
     MiddlewareFactory::UPtr middleware_factory_;
@@ -74,7 +76,10 @@
     mutable std::string registry_endpointdir_;
     mutable std::string registry_endpoint_;
     mutable Reaper::SPtr reply_reaper_;
-    mutable std::mutex mutex_;                          // For lazy initialization of reply_reaper_
+    mutable ThreadPool::SPtr async_pool_;  // Pool of invocation threads for async query creation
+    mutable ThreadSafeQueue<std::future<void>>::SPtr future_queue_;
+    mutable std::thread waiter_thread_;
+    mutable std::mutex mutex_;  // For lazy initialization of reply_reaper_, async_pool_, and queue_
 };
 
 } // namespace internal

=== modified file 'include/unity/scopes/internal/ScopeImpl.h'
--- include/unity/scopes/internal/ScopeImpl.h	2014-03-07 04:19:32 +0000
+++ include/unity/scopes/internal/ScopeImpl.h	2014-04-28 05:57:56 +0000
@@ -66,7 +66,7 @@
 
     QueryCtrlProxy search(CannedQuery const& query,
                           SearchMetadata const& metadata,
-                          SearchListenerBase::SPtr const& reply); // Not remote operation, hence not override
+                          SearchListenerBase::SPtr const& reply); // Not remote, hence not override
 
     virtual QueryCtrlProxy activate(Result const& result,
                                     ActionMetadata const& metadata,
@@ -85,7 +85,7 @@
     static ScopeProxy create(MWScopeProxy const& mw_proxy, RuntimeImpl* runtime, std::string const& scope_id);
 
 private:
-    MWScopeProxy fwd() const;
+    MWScopeProxy fwd();
 
     RuntimeImpl* const runtime_;
     std::string scope_id_;

=== modified file 'include/unity/scopes/internal/SearchReplyImpl.h'
--- include/unity/scopes/internal/SearchReplyImpl.h	2014-03-07 10:54:15 +0000
+++ include/unity/scopes/internal/SearchReplyImpl.h	2014-04-28 05:57:56 +0000
@@ -47,7 +47,7 @@
 class SearchReplyImpl : public virtual unity::scopes::SearchReply, public virtual ReplyImpl
 {
 public:
-    SearchReplyImpl(MWReplyProxy const& mw_proxy, std::shared_ptr<QueryObjectBase>const & qo);
+    SearchReplyImpl(MWReplyProxy const& mw_proxy, std::shared_ptr<QueryObjectBase>const & qo, int cardinality);
     virtual ~SearchReplyImpl();
 
     virtual void register_departments(DepartmentList const& departments, std::string current_department_id) override;

=== modified file 'include/unity/scopes/internal/TaskWrapper.h'
--- include/unity/scopes/internal/TaskWrapper.h	2014-01-23 11:28:34 +0000
+++ include/unity/scopes/internal/TaskWrapper.h	2014-04-28 05:57:56 +0000
@@ -31,7 +31,7 @@
 {
 
 // Simple wrapper for tasks. Allows us to use a std::packaged_task as the functor
-// for a task in a thread pool, without having to know the task's return type in advance.
+// for a task in a thread pool without having to know the task's return type in advance.
 
 class TaskWrapper final
 {
@@ -54,7 +54,6 @@
 
     void operator()()
     {
-
         task_->call();
     }
 

=== modified file 'include/unity/scopes/internal/ThreadPool.h'
--- include/unity/scopes/internal/ThreadPool.h	2014-01-23 11:28:34 +0000
+++ include/unity/scopes/internal/ThreadPool.h	2014-04-28 05:57:56 +0000
@@ -22,6 +22,7 @@
 #include <unity/scopes/internal/ThreadSafeQueue.h>
 #include <unity/scopes/internal/TaskWrapper.h>
 #include <unity/UnityExceptions.h>
+#include <unity/util/DefinesPtrs.h>
 
 #include <future>
 
@@ -42,6 +43,7 @@
 {
 public:
     NONCOPYABLE(ThreadPool);
+    UNITY_DEFINES_PTRS(ThreadPool);
 
     ThreadPool(int size);
     ~ThreadPool();

=== modified file 'include/unity/scopes/internal/ThreadSafeQueue.h'
--- include/unity/scopes/internal/ThreadSafeQueue.h	2014-01-23 11:28:34 +0000
+++ include/unity/scopes/internal/ThreadSafeQueue.h	2014-04-28 05:57:56 +0000
@@ -19,9 +19,9 @@
 #ifndef UNITY_SCOPES_INTERNAL_THREADSAFEQUEUE_H
 #define UNITY_SCOPES_INTERNAL_THREADSAFEQUEUE_H
 
+#include <unity/util/DefinesPtrs.h>
 #include <unity/util/NonCopyable.h>
 
-#include <atomic>
 #include <condition_variable>
 #include <mutex>
 #include <queue>
@@ -45,30 +45,34 @@
 {
 public:
     NONCOPYABLE(ThreadSafeQueue);
+    UNITY_DEFINES_PTRS(ThreadSafeQueue);
+
     typedef T value_type;
 
     ThreadSafeQueue();
     ~ThreadSafeQueue();
 
     void destroy() noexcept;
+    void wait_for_destroy() noexcept;
     void push(T const& item);
     void push(T&& item);
     T wait_and_pop();
     bool try_pop(T& item);
     bool empty() const noexcept;
+    void wait_until_empty() const noexcept;
     size_t size() const noexcept;
 
 private:
     std::queue<T> queue_;
     mutable std::mutex mutex_;
-    std::condition_variable cond_;
-    bool done_;
-    std::atomic<int> num_waiters_;
+    mutable std::condition_variable cond_;
+    bool destroyed_;
+    int num_waiters_;
 };
 
 template<typename T>
 ThreadSafeQueue<T>::ThreadSafeQueue() :
-    done_(false),
+    destroyed_(false),
     num_waiters_(0)
 {
 }
@@ -77,40 +81,50 @@
 ThreadSafeQueue<T>::~ThreadSafeQueue()
 {
     destroy();
-
-    // Don't destroy the object while there are still threads in wait_and_pop(), otherwise
-    // a thread that wakes up in wait_and_pop() will try to re-lock the already-destroyed
-    // mutex.
-    while (num_waiters_.load() > 0)
-        std::this_thread::yield();  // LCOV_EXCL_LINE (impossible to reliably hit with a test)
+    wait_for_destroy();
 }
 
 template<typename T>
 void ThreadSafeQueue<T>::destroy() noexcept
 {
     std::lock_guard<std::mutex> lock(mutex_);
-    if (done_)
+    if (destroyed_)
     {
         return;
     }
-    done_ = true;
-    cond_.notify_all(); // Wake up anyone asleep in wait_and_pop()
+    destroyed_ = true;
+    cond_.notify_all(); // Wake up anyone asleep in wait_and_pop() or wait_for_destroy()
+}
+
+template<typename T>
+void ThreadSafeQueue<T>::wait_for_destroy() noexcept
+{
+    std::unique_lock<std::mutex> lock(mutex_);
+    cond_.wait(lock, [this] { return destroyed_ && num_waiters_ == 0; });
 }
 
 template<typename T>
 void ThreadSafeQueue<T>::push(T const& item)
 {
     std::lock_guard<std::mutex> lock(mutex_);
+    if (destroyed_)
+    {
+        throw std::runtime_error("ThreadSafeQueue: cannot push onto destroyed queue");
+    }
     queue_.push(item);
-    cond_.notify_one();
+    cond_.notify_all();
 }
 
 template<typename T>
 void ThreadSafeQueue<T>::push(T&& item)
 {
     std::lock_guard<std::mutex> lock(mutex_);
+    if (destroyed_)
+    {
+        throw std::runtime_error("ThreadSafeQueue: cannot push onto destroyed queue");
+    }
     queue_.emplace(std::move(item));
-    cond_.notify_one();
+    cond_.notify_all();
 }
 
 template<typename T>
@@ -118,16 +132,21 @@
 {
     std::unique_lock<std::mutex> lock(mutex_);
     ++num_waiters_;
-    cond_.wait(lock, [this] { return done_ || queue_.size() != 0; });
-    if (done_)
+    cond_.wait(lock, [this] { return destroyed_ || queue_.size() != 0; });
+    if (destroyed_)
     {
-        lock.unlock();
-        --num_waiters_;
+        if (--num_waiters_ == 0)
+        {
+            cond_.notify_all();
+        }
         throw std::runtime_error("ThreadSafeQueue: queue destroyed while thread was blocked in wait_and_pop()");
     }
     T item = std::move(queue_.front());
     queue_.pop();
-    --num_waiters_;
+    if (--num_waiters_ == 0 || queue_.empty())
+    {
+        cond_.notify_all();
+    }
     return item;
 }
 
@@ -141,6 +160,10 @@
     }
     item = std::move(queue_.front());
     queue_.pop();
+    if (queue_.empty())
+    {
+        cond_.notify_all();
+    }
     return true;
 }
 
@@ -152,6 +175,13 @@
 }
 
 template<typename T>
+void ThreadSafeQueue<T>::wait_until_empty() const noexcept
+{
+    std::unique_lock<std::mutex> lock(mutex_);
+    cond_.wait(lock, [this] { return queue_.empty(); });
+}
+
+template<typename T>
 size_t ThreadSafeQueue<T>::size() const noexcept
 {
     std::lock_guard<std::mutex> lock(mutex_);

=== modified file 'include/unity/scopes/internal/zmq_middleware/ConnectionPool.h'
--- include/unity/scopes/internal/zmq_middleware/ConnectionPool.h	2014-01-22 03:29:48 +0000
+++ include/unity/scopes/internal/zmq_middleware/ConnectionPool.h	2014-04-28 05:57:56 +0000
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2013 Canonical Ltd
+ * Copyright (C) 2014 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
@@ -24,7 +24,6 @@
 
 #include <zmqpp/socket.hpp>
 
-#include <mutex>
 #include <string>
 #include <unordered_map>
 
@@ -33,6 +32,8 @@
 // different threads to crash.
 // So, we maintain a pool of invocation threads, with each thread keeping its own cache of sockets.
 // Sockets are indexed by adapter name and created lazily.
+//
+// WARNING: No locking anywhere here. The pool is intended for us as a thread_local static member only.
 
 namespace unity
 {
@@ -57,17 +58,18 @@
     void register_socket(std::string const& endpoint, zmqpp::socket socket, RequestMode m);
 
 private:
-    struct Connection
+    struct SocketData
     {
         zmqpp::socket socket;
         RequestMode mode;
     };
 
-    typedef std::unordered_map<std::string, Connection> CPool;
+    typedef std::unordered_map<std::string, SocketData> CPool;
+
+    CPool::value_type create_connection(std::string const& endpoint, RequestMode m, int64_t timeout);
 
     zmqpp::context& context_;
     CPool pool_;
-    std::mutex mutex_;
 };
 
 } // namespace zmq_middleware

=== modified file 'include/unity/scopes/internal/zmq_middleware/ObjectAdapter.h'
--- include/unity/scopes/internal/zmq_middleware/ObjectAdapter.h	2014-01-24 00:26:43 +0000
+++ include/unity/scopes/internal/zmq_middleware/ObjectAdapter.h	2014-04-28 05:57:56 +0000
@@ -46,6 +46,7 @@
 {
 
 class ServantBase;
+class StopPublisher;
 class ZmqMiddleware;
 
 class ObjectAdapter final
@@ -89,12 +90,12 @@
     // The Failed state is reachable from any of the other states and indicates
     // a fatal error condition.
     enum AdapterState { Inactive, Activating, Active, Deactivating, Destroyed, Failed };
-    void throw_bad_state(AdapterState state) const;
+    void throw_bad_state(std::string const& label, AdapterState state) const;
 
     void run_workers();
-    void init_ctrl_socket();
-    zmqpp::socket subscribe_to_ctrl_socket();
-    void stop_workers() noexcept;
+    // void init_ctrl_socket();
+    // zmqpp::socket subscribe_to_ctrl_socket();
+    // void stop_workers() noexcept;
 
     void safe_bind(zmqpp::socket& s, std::string const& endpoint);
 
@@ -114,8 +115,9 @@
     std::string endpoint_;
     RequestMode mode_;
     int pool_size_;
-    std::unique_ptr<zmqpp::socket> ctrl_;       // PUB socket to signal when to deactivate
-    std::mutex ctrl_mutex_;                     // Synchronizes access to ctrl_ when sending
+    // std::unique_ptr<zmqpp::socket> ctrl_;       // PUB socket to signal when to deactivate
+    // std::mutex ctrl_mutex_;                     // Synchronizes access to ctrl_ when sending
+    std::unique_ptr<StopPublisher> stopper_;    // Used to signal threads when it's time to terminate
     std::thread broker_;                        // Connects router with dealer
     std::vector<std::thread> workers_;          // Threads for incoming invocations
     std::atomic_int num_workers_;               // For handshake with parent

=== added file 'include/unity/scopes/internal/zmq_middleware/StopPublisher.h'
--- include/unity/scopes/internal/zmq_middleware/StopPublisher.h	1970-01-01 00:00:00 +0000
+++ include/unity/scopes/internal/zmq_middleware/StopPublisher.h	2014-04-28 05:57:56 +0000
@@ -0,0 +1,85 @@
+/*
+ * Copyright (C) 2014 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: Michi Henning <[e-mail]>
+ */
+
+#ifndef UNITY_SCOPES_INTERNAL_ZMQMIDDLEWARE_STOPPUBLISHER_H
+#define UNITY_SCOPES_INTERNAL_ZMQMIDDLEWARE_STOPPUBLISHER_H
+
+#include <unity/util/NonCopyable.h>
+
+#include <zmqpp/context.hpp>
+#include <zmqpp/socket.hpp>
+
+#include <condition_variable>
+#include <mutex>
+#include <thread>
+
+namespace unity
+{
+
+namespace scopes
+{
+
+namespace internal
+{
+
+namespace zmq_middleware
+{
+
+// StopPublisher provides a way to use Zmq pub/sub without falling prey
+// to the slow joiner syndrome. The provided name is used to create an inproc
+// endpoint by prefixing the name with "inproc://".
+// Subscribers create a socket by calling subscribe(). The returned
+// socket becomes ready for reading (for a zero-length message) once
+// some thread calls stop(). Multiple calls to stop() are benign.
+
+class StopPublisher final
+{
+public:
+    NONCOPYABLE(StopPublisher);
+
+    StopPublisher(zmqpp::context* context, std::string const& inproc_name);
+    ~StopPublisher();
+
+    std::string endpoint() const;
+    zmqpp::socket subscribe();
+    void stop();
+
+private:
+    zmqpp::context* context_;
+    std::string endpoint_;
+    std::thread thread_;
+    std::exception_ptr ex_;
+
+    enum State { Starting, Started, Stopping, Stopped, Failed };
+    State state_;
+
+    mutable std::mutex m_;
+    std::condition_variable cond_;
+
+    void stopper_thread() noexcept;
+};
+
+} // namespace zmq_middleware
+
+} // namespace internal
+
+} // namespace scopes
+
+} // namespace unity
+
+#endif

=== modified file 'include/unity/scopes/internal/zmq_middleware/ZmqObjectProxy.h'
--- include/unity/scopes/internal/zmq_middleware/ZmqObjectProxy.h	2014-02-03 02:53:33 +0000
+++ include/unity/scopes/internal/zmq_middleware/ZmqObjectProxy.h	2014-04-28 05:57:56 +0000
@@ -21,6 +21,7 @@
 
 #include <unity/scopes/internal/MWObjectProxy.h>
 #include <scopes/internal/zmq_middleware/capnproto/Message.capnp.h>
+#include <unity/scopes/internal/zmq_middleware/ConnectionPool.h>
 #include <unity/scopes/internal/zmq_middleware/RequestMode.h>
 #include <unity/scopes/internal/zmq_middleware/ZmqMiddleware.h>
 #include <unity/scopes/internal/zmq_middleware/ZmqObjectProxyFwd.h>
@@ -71,8 +72,11 @@
 
 protected:
     capnproto::Request::Builder make_request_(capnp::MessageBuilder& b, std::string const& operation_name) const;
-    ZmqReceiver invoke_(capnp::MessageBuilder& out_params);
-    ZmqReceiver invoke_(capnp::MessageBuilder& out_params, int64_t timeout);
+
+    void invoke_oneway_(capnp::MessageBuilder& out_params);
+
+    ZmqReceiver invoke_twoway_(capnp::MessageBuilder& out_params);
+    ZmqReceiver invoke_twoway_(capnp::MessageBuilder& out_params, int64_t timeout);
 
 private:
     std::string endpoint_;

=== modified file 'src/scopes/QueryBase.cpp'
--- src/scopes/QueryBase.cpp	2014-03-10 04:17:09 +0000
+++ src/scopes/QueryBase.cpp	2014-04-28 05:57:56 +0000
@@ -50,6 +50,11 @@
     p->set_metadata(metadata);
 }
 
+bool QueryBase::valid() const
+{
+    return p->valid();
+}
+
 } // namespace scopes
 
 } // namespace unity

=== modified file 'src/scopes/internal/ObjectImpl.cpp'
--- src/scopes/internal/ObjectImpl.cpp	2014-04-03 12:57:25 +0000
+++ src/scopes/internal/ObjectImpl.cpp	2014-04-28 05:57:56 +0000
@@ -19,8 +19,12 @@
 #include <unity/scopes/internal/ObjectImpl.h>
 
 #include <unity/scopes/internal/MWObjectProxy.h>
+#include <unity/scopes/ScopeExceptions.h>
+
+#include <cassert>
 
 using namespace std;
+using namespace unity::scopes;
 
 namespace unity
 {
@@ -31,9 +35,10 @@
 namespace internal
 {
 
-ObjectImpl::ObjectImpl(MWProxy const& mw_proxy) :
-    mw_proxy_(mw_proxy)
+ObjectImpl::ObjectImpl(MWProxy const& mw_proxy)
 {
+    lock_guard<mutex> lock(proxy_mutex_);
+    mw_proxy_ = mw_proxy;
 }
 
 ObjectImpl::~ObjectImpl()
@@ -42,6 +47,7 @@
 
 string ObjectImpl::identity()
 {
+    check_proxy();
     return mw_proxy_->identity();
 }
 
@@ -52,29 +58,51 @@
 
 string ObjectImpl::endpoint()
 {
+    check_proxy();
     return mw_proxy_->endpoint();
 }
 
 int64_t ObjectImpl::timeout()
 {
+    check_proxy();
     return mw_proxy_->timeout();
 }
 
 string ObjectImpl::to_string()
 {
+    check_proxy();
     return mw_proxy_->to_string();
 }
 
 void ObjectImpl::ping()
 {
+    check_proxy();
     mw_proxy_->ping();
 }
 
-MWProxy ObjectImpl::proxy() const
+MWProxy ObjectImpl::proxy()
 {
+    lock_guard<mutex> lock(proxy_mutex_);
     return mw_proxy_;
 }
 
+void ObjectImpl::set_proxy(MWProxy const& p)
+{
+    assert(p);
+    lock_guard<mutex> lock(proxy_mutex_);
+    assert(!mw_proxy_);
+    mw_proxy_ = p;
+}
+
+void ObjectImpl::check_proxy()
+{
+    lock_guard<mutex> lock(proxy_mutex_);
+    if (!mw_proxy_)
+    {
+        throw MiddlewareException("Cannot invoke on null proxy");
+    }
+}
+
 } // namespace internal
 
 } // namespace scopes

=== modified file 'src/scopes/internal/QueryBaseImpl.cpp'
--- src/scopes/internal/QueryBaseImpl.cpp	2014-03-10 04:17:09 +0000
+++ src/scopes/internal/QueryBaseImpl.cpp	2014-04-28 05:57:56 +0000
@@ -37,6 +37,7 @@
 {
 
 QueryBaseImpl::QueryBaseImpl()
+    : valid_(true)
 {
 }
 
@@ -54,6 +55,8 @@
     // This allows cancel() to forward incoming cancellations to subqueries
     // without intervention from the scope application code.
     QueryCtrlProxy qcp = scope->search(query_string, *search_metadata_, reply);
+
+    lock_guard<mutex> lock(mutex_);
     subqueries_.push_back(qcp);
     return qcp;
 }
@@ -66,6 +69,8 @@
     assert(search_metadata_);
 
     QueryCtrlProxy qcp = scope->search(query_string, filter_state, *search_metadata_, reply);
+
+    lock_guard<mutex> lock(mutex_);
     subqueries_.push_back(qcp);
     return qcp;
 }
@@ -79,6 +84,8 @@
     assert(search_metadata_);
 
     QueryCtrlProxy qcp = scope->search(query_string, department_id, filter_state, *search_metadata_, reply);
+
+    lock_guard<mutex> lock(mutex_);
     subqueries_.push_back(qcp);
     return qcp;
 }
@@ -91,6 +98,8 @@
                                               SearchListenerBase::SPtr const& reply)
 {
     QueryCtrlProxy qcp = scope->search(query_string, department_id, filter_state, metadata, reply);
+
+    lock_guard<mutex> lock(mutex_);
     subqueries_.push_back(qcp);
     return qcp;
 }
@@ -102,16 +111,29 @@
 
 void QueryBaseImpl::cancel()
 {
+    lock_guard<mutex> lock(mutex_);
+
+    if (!valid_)
+    {
+        return;
+    }
+    valid_ = false;
     for (auto& ctrl : subqueries_)
     {
         ctrl->cancel(); // Forward the cancellation to any subqueries that might be active
     }
     // We release the memory for the subquery controls here. That's just a micro-optimization
-    // because this QueryBase will be destroyed shortly anyway, once the cancelled() method
-    // of the application returns. (Not deallocating here would work too.)
+    // because this QueryBase will be destroyed shortly anyway, once the cancelled() (and possibly
+    // run()) methods of the application return. (Not deallocating here would work too.)
     vector<QueryCtrlProxy>().swap(subqueries_);
 }
 
+bool QueryBaseImpl::valid() const
+{
+    lock_guard<mutex> lock(mutex_);
+    return valid_;
+}
+
 } // namespace internal
 
 } // namespace scopes

=== modified file 'src/scopes/internal/QueryCtrlImpl.cpp'
--- src/scopes/internal/QueryCtrlImpl.cpp	2014-03-07 02:01:47 +0000
+++ src/scopes/internal/QueryCtrlImpl.cpp	2014-04-28 05:57:56 +0000
@@ -43,10 +43,13 @@
     ObjectImpl(ctrl_proxy),
     reply_proxy_(reply_proxy)
 {
-    assert(ctrl_proxy);
-    assert(reply_proxy);
     // We remember the reply proxy so, when the query is cancelled, we can
     // inform the reply object belonging to this query that the query is finished.
+    assert(reply_proxy);
+
+    lock_guard<mutex> lock(mutex_);
+    ready_ = ctrl_proxy != nullptr;
+    cancelled_ = false;
 }
 
 QueryCtrlImpl::~QueryCtrlImpl()
@@ -55,13 +58,24 @@
 
 void QueryCtrlImpl::cancel()
 {
+    {
+        lock_guard<mutex> lock(mutex_);
+        if (!ready_)
+        {
+            // Remember that query was cancelled, so we can call
+            // cancel() once set_proxy() is called.
+            cancelled_ = true;
+            return;
+        }
+    }
+
     try
     {
         // Forward cancellation down-stream to the query. This does not block.
         fwd()->cancel();
 
         // Indicate (to ourselves) that this query is complete. Calling via the MWReplyProxy ensures
-        // the finished() call will be processed by a seperate server-side thread,
+        // the finished() call will be processed by a separate server-side thread,
         // so we cannot block here.
         reply_proxy_->finished(ListenerBase::Cancelled, "");
     }
@@ -72,7 +86,28 @@
     }
 }
 
-MWQueryCtrlProxy QueryCtrlImpl::fwd() const
+void QueryCtrlImpl::set_proxy(MWQueryCtrlProxy const& p)
+{
+    assert(proxy() == nullptr);
+    ObjectImpl::set_proxy(p);
+
+    bool need_cancel;
+
+    {
+        lock_guard<mutex> lock(mutex_);
+        ready_ = true;
+        need_cancel = cancelled_;
+    } // Unlock
+
+    if (need_cancel)
+    {
+        // If cancel() was called earlier, do the actual cancellation now
+        // that we have the middleware proxy.
+        cancel();
+    }
+}
+
+MWQueryCtrlProxy QueryCtrlImpl::fwd()
 {
     return dynamic_pointer_cast<MWQueryCtrl>(proxy());
 }

=== modified file 'src/scopes/internal/QueryObject.cpp'
--- src/scopes/internal/QueryObject.cpp	2014-03-07 03:49:03 +0000
+++ src/scopes/internal/QueryObject.cpp	2014-04-28 05:57:56 +0000
@@ -77,6 +77,8 @@
 
 void QueryObject::run(MWReplyProxy const& reply, InvokeInfo const& /* info */) noexcept
 {
+    unique_lock<mutex> lock(mutex_);
+
     // It is possible for a run() to be dispatched by the middleware *after* the query
     // was cancelled. This can happen because run() and cancel() are dispatched by different
     // thread pools. If the scope implementation uses a synchronous run(), a later run()
@@ -84,16 +86,18 @@
     // completes, by which time the query for the later run() call may have been
     // cancelled already.
     // If the query was cancelled by the client before we even receive the
-    // run invocation, we never forward the run() to the implementation.
+    // run invocation, we never forward the run() call to the implementation.
     if (!pushable_)
     {
+        self_ = nullptr;
+        disconnect();
         return;
     }
 
     // Create the reply proxy to pass to query_base_ and keep a weak_ptr, which we will need
     // if cancel() is called later.
     assert(self_);
-    auto reply_proxy = make_shared<SearchReplyImpl>(reply, self_);
+    auto reply_proxy = make_shared<SearchReplyImpl>(reply, self_, cardinality_);
     assert(reply_proxy);
     reply_proxy_ = reply_proxy;
 
@@ -102,12 +106,15 @@
     self_ = nullptr;
     disconnect();
 
-    // Synchronous call into scope implementation.
-    // On return, replies for the query may still be outstanding.
     try
     {
         auto search_query = dynamic_pointer_cast<SearchQueryBase>(query_base_);
         assert(search_query);
+
+        lock.unlock();
+
+        // Synchronous call into scope implementation.
+        // On return, replies for the query may still be outstanding.
         search_query->run(reply_proxy);
     }
     catch (std::exception const& e)
@@ -128,15 +135,24 @@
 
 void QueryObject::cancel(InvokeInfo const& /* info */)
 {
-    pushable_ = false;
-    auto rp = reply_proxy_.lock();
-    if (rp)
     {
-        // Send finished() to up-stream client to tell him the query is done.
-        // We send via the MWReplyProxy here because that allows passing
-        // a ListenerBase::Reason (whereas the public ReplyProxy does not).
-        reply_->finished(ListenerBase::Cancelled, "");     // Oneway, can't block
-    }
+        lock_guard<mutex> lock(mutex_);
+
+        if (!pushable_)
+        {
+            return;
+        }
+        pushable_ = false;
+
+        auto rp = reply_proxy_.lock();
+        if (rp)
+        {
+            // Send finished() to up-stream client to tell him the query is done.
+            // We send via the MWReplyProxy here because that allows passing
+            // a ListenerBase::Reason (whereas the public ReplyProxy does not).
+            reply_->finished(ListenerBase::Cancelled, "");     // Oneway, can't block
+        }
+    }  // Release lock
 
     // Forward the cancellation to the query base (which in turn will forward it to any subqueries).
     // The query base also calls the cancelled() callback to inform the application code.
@@ -145,11 +161,13 @@
 
 bool QueryObject::pushable(InvokeInfo const& /* info */) const noexcept
 {
+    lock_guard<mutex> lock(mutex_);
     return pushable_;
 }
 
 int QueryObject::cardinality(InvokeInfo const& /* info */) const noexcept
 {
+    lock_guard<mutex> lock(mutex_);
     return cardinality_;
 }
 
@@ -169,6 +187,8 @@
 
 void QueryObject::set_self(QueryObjectBase::SPtr const& self) noexcept
 {
+    lock_guard<mutex> lock(mutex_);
+
     assert(self);
     assert(!self_);
     self_ = self;

=== modified file 'src/scopes/internal/RegistryImpl.cpp'
--- src/scopes/internal/RegistryImpl.cpp	2014-04-03 12:57:25 +0000
+++ src/scopes/internal/RegistryImpl.cpp	2014-04-28 05:57:56 +0000
@@ -71,7 +71,7 @@
     return matching_entries;
 }
 
-MWRegistryProxy RegistryImpl::fwd() const
+MWRegistryProxy RegistryImpl::fwd()
 {
     return dynamic_pointer_cast<MWRegistry>(proxy());
 }

=== modified file 'src/scopes/internal/ReplyImpl.cpp'
--- src/scopes/internal/ReplyImpl.cpp	2014-03-07 05:56:22 +0000
+++ src/scopes/internal/ReplyImpl.cpp	2014-04-28 05:57:56 +0000
@@ -143,7 +143,7 @@
     }
 }
 
-MWReplyProxy ReplyImpl::fwd() const
+MWReplyProxy ReplyImpl::fwd()
 {
     return dynamic_pointer_cast<MWReply>(proxy());
 }

=== modified file 'src/scopes/internal/ReplyObject.cpp'
--- src/scopes/internal/ReplyObject.cpp	2014-03-10 10:16:13 +0000
+++ src/scopes/internal/ReplyObject.cpp	2014-04-28 05:57:56 +0000
@@ -50,9 +50,8 @@
     assert(receiver_base);
     assert(runtime);
     reap_item_ = runtime->reply_reaper()->add([this] {
-        cerr << "No activity on ReplyObject for scope " << this->origin_proxy_
-             << ": ReplyObject destroyed" << endl;
-        this->disconnect();
+        string msg = "No activity on ReplyObject for scope " + this->origin_proxy_ + ": ReplyObject destroyed";
+        this->finished(ListenerBase::Error, msg);
     });
 }
 

=== modified file 'src/scopes/internal/RuntimeImpl.cpp'
--- src/scopes/internal/RuntimeImpl.cpp	2014-04-11 08:22:17 +0000
+++ src/scopes/internal/RuntimeImpl.cpp	2014-04-28 05:57:56 +0000
@@ -51,11 +51,14 @@
 namespace internal
 {
 
-RuntimeImpl::RuntimeImpl(string const& scope_id, string const& configfile) :
-    destroyed_(false),
-    scope_id_(scope_id)
+RuntimeImpl::RuntimeImpl(string const& scope_id, string const& configfile)
 {
-    if (scope_id.empty())
+    lock_guard<mutex> lock(mutex_);
+
+    destroyed_ = false;
+
+    scope_id_ = scope_id;
+    if (scope_id_.empty())
     {
         UniqueID id;
         scope_id_ = "c-" + id.gen();
@@ -77,6 +80,10 @@
         middleware_ = middleware_factory_->create(scope_id_, default_middleware, middleware_configfile);
         middleware_->start();
 
+        async_pool_ = make_shared<ThreadPool>(1); // TODO: configurable pool size
+        future_queue_ = make_shared<ThreadSafeQueue<future<void>>>();
+        waiter_thread_ = std::thread([this]{ waiter_thread(future_queue_); });
+
         if (registry_configfile_.empty() || registry_identity_.empty())
         {
             cerr << "Warning: no registry configured" << endl;
@@ -124,21 +131,51 @@
 
 void RuntimeImpl::destroy()
 {
-    if (!destroyed_.exchange(true))
-    {
-        // TODO: not good enough. Need to wait for the middleware to stop and for the reaper
-        // to terminate. Otherwise, it's possible that we exit while threads are still running
-        // with undefined behavior.
-        registry_ = nullptr;
-        middleware_->stop();
-        middleware_ = nullptr;
-        middleware_factory_.reset(nullptr);
+    lock_guard<mutex> lock(mutex_);
+
+    if (destroyed_)
+    {
+        return;
+    }
+    destroyed_ = true;
+
+    // Stop the reaper. Any ReplyObject instances that may still
+    // be hanging around will be cleaned up when the server side
+    // is shut down.
+    if (reply_reaper_)
+    {
+        reply_reaper_->destroy();
         reply_reaper_ = nullptr;
     }
+
+    // No more outgoing invocations.
+    async_pool_ = nullptr;
+
+    // Wait for any twoway operations that were invoked asynchronously to complete.
+    if (future_queue_)
+    {
+        future_queue_->wait_until_empty();
+        future_queue_->destroy();
+        future_queue_->wait_for_destroy();
+    }
+
+    if (waiter_thread_.joinable())
+    {
+        waiter_thread_.join();
+    }
+
+    // Shut down server-side.
+    middleware_->stop();
+    middleware_->wait_for_shutdown();
+    middleware_ = nullptr;
+    middleware_factory_.reset(nullptr);
+
+    registry_ = nullptr;
 }
 
 string RuntimeImpl::scope_id() const
 {
+    lock_guard<mutex> lock(mutex_);
     return scope_id_;
 }
 
@@ -149,7 +186,9 @@
 
 MiddlewareFactory const* RuntimeImpl::factory() const
 {
-    if (destroyed_.load())
+    lock_guard<mutex> lock(mutex_);
+
+    if (destroyed_)
     {
         throw LogicException("factory(): Cannot obtain factory for already destroyed run time");
     }
@@ -158,7 +197,9 @@
 
 RegistryProxy RuntimeImpl::registry() const
 {
-    if (destroyed_.load())
+    lock_guard<mutex> lock(mutex_);
+
+    if (destroyed_)
     {
         throw LogicException("registry(): Cannot obtain registry for already destroyed run time");
     }
@@ -171,21 +212,25 @@
 
 string RuntimeImpl::registry_configfile() const
 {
+    lock_guard<mutex> lock(mutex_);
     return registry_configfile_;
 }
 
 string RuntimeImpl::registry_identity() const
 {
+    lock_guard<mutex> lock(mutex_);
     return registry_identity_;
 }
 
 string RuntimeImpl::registry_endpointdir() const
 {
+    lock_guard<mutex> lock(mutex_);
     return registry_endpointdir_;
 }
 
 string RuntimeImpl::registry_endpoint() const
 {
+    lock_guard<mutex> lock(mutex_);
     return registry_endpoint_;
 }
 
@@ -200,6 +245,35 @@
     return reply_reaper_;
 }
 
+void RuntimeImpl::waiter_thread(ThreadSafeQueue<std::future<void>>::SPtr const& queue) const noexcept
+{
+    for (;;)
+    {
+        try
+        {
+            // Wait on the future from an async invocation.
+            // wait_and_pop() throws runtime_error when queue is destroyed
+            queue->wait_and_pop().get();
+        }
+        catch (std::runtime_error const&)
+        {
+            break;
+        }
+    }
+}
+
+ThreadPool::SPtr RuntimeImpl::async_pool() const
+{
+    lock_guard<mutex> lock(mutex_);
+    return async_pool_;
+}
+
+ThreadSafeQueue<future<void>>::SPtr RuntimeImpl::future_queue() const
+{
+    lock_guard<mutex> lock(mutex_);
+    return future_queue_;
+}
+
 void RuntimeImpl::run_scope(ScopeBase *const scope_base, std::string const& scope_ini_file)
 {
     // Retrieve the registry middleware and create a proxy to its state receiver

=== modified file 'src/scopes/internal/ScopeImpl.cpp'
--- src/scopes/internal/ScopeImpl.cpp	2014-03-07 04:19:32 +0000
+++ src/scopes/internal/ScopeImpl.cpp	2014-04-28 05:57:56 +0000
@@ -21,6 +21,7 @@
 #include <unity/scopes/ActionMetadata.h>
 #include <unity/scopes/internal/ActivationReplyObject.h>
 #include <unity/scopes/internal/MiddlewareBase.h>
+#include <unity/scopes/internal/MWQueryCtrl.h>
 #include <unity/scopes/internal/MWScope.h>
 #include <unity/scopes/internal/PreviewReplyObject.h>
 #include <unity/scopes/internal/QueryCtrlImpl.h>
@@ -30,7 +31,6 @@
 #include <unity/UnityExceptions.h>
 
 #include <cassert>
-#include <iostream> // TODO: remove this once logging is added
 
 using namespace std;
 
@@ -95,30 +95,48 @@
         throw unity::InvalidArgumentException("Scope::search(): invalid SearchListenerBase (nullptr)");
     }
 
-    QueryCtrlProxy ctrl;
     ReplyObject::SPtr ro(make_shared<ResultReplyObject>(reply, runtime_, to_string(), metadata.cardinality()));
-    try
-    {
-        MWReplyProxy rp = fwd()->mw_base()->add_reply_object(ro);
-
-        // Forward the search() method across the bus. This is a
-        // synchronous twoway interaction with the scope, so it can return
-        // the QueryCtrlProxy. This may block for some time, for example, if
-        // the scope is not running and needs to be activated by the registry first.
-        ctrl = fwd()->search(query, metadata.serialize(), rp);
-        assert(ctrl);
-    }
-    catch (std::exception const& e)
+    MWReplyProxy rp = fwd()->mw_base()->add_reply_object(ro);
+
+    // "Fake" QueryCtrlProxy that doesn't have a real MWQueryCtrlProxy yet.
+    shared_ptr<QueryCtrlImpl> ctrl = make_shared<QueryCtrlImpl>(nullptr, rp);
+
+    // We pass a shared pointer to the lambda (instead of the this pointer)
+    // to keep ourselves alive until after the lambda fires.
+    // Otherwise, if the ScopeProxy for this invocation goes out of scope before the invocation is
+    // sent in the new thread, the lambda will call into this by-now-destroyed instance.
+    auto impl = dynamic_pointer_cast<ScopeImpl>(shared_from_this());
+
+    auto send_search = [impl, query, metadata, rp, ro, ctrl]() -> void
     {
         try
         {
-            ro->finished(ListenerBase::Error, e.what());
+            // Forward the (synchronous) search() method across the bus.
+            auto real_ctrl = dynamic_pointer_cast<QueryCtrlImpl>(impl->fwd()->search(query, metadata.serialize(), rp));
+            assert(real_ctrl);
+
+            // Call has completed now, so we update the MWQueryCtrlProxy for the fake proxy
+            // with the real proxy that was returned.
+            auto new_proxy = dynamic_pointer_cast<MWQueryCtrl>(real_ctrl->proxy());
+            assert(new_proxy);
+            ctrl->set_proxy(new_proxy);
         }
-        catch (...)
+        catch (std::exception const& e)
         {
+            try
+            {
+                ro->finished(ListenerBase::Error, e.what());
+            }
+            catch (...)
+            {
+            }
         }
-        throw;
-    }
+    };
+
+    // Send the blocking twoway request asynchronously via the async invocation pool. The waiter thread
+    // waits on the future, so it gets cleaned up.
+    auto future = runtime_->async_pool()->submit(send_search);
+    runtime_->future_queue()->push(move(future));
     return ctrl;
 }
 
@@ -131,31 +149,40 @@
         throw unity::InvalidArgumentException("Scope::activate(): invalid ActivationListenerBase (nullptr)");
     }
 
-    QueryCtrlProxy ctrl;
     ActivationReplyObject::SPtr ro(make_shared<ActivationReplyObject>(reply, runtime_, to_string()));
-    try
-    {
-        MWReplyProxy rp = fwd()->mw_base()->add_reply_object(ro);
-
-        // Forward the activate() method across the bus.
-        ctrl = fwd()->activate(result.p->activation_target(), metadata.serialize(), rp);
-        assert(ctrl);
-    }
-    catch (std::exception const& e)
-    {
-        // TODO: log error
-        cerr << "activate(): " << e.what() << endl;
+    MWReplyProxy rp = fwd()->mw_base()->add_reply_object(ro);
+
+    shared_ptr<QueryCtrlImpl> ctrl = make_shared<QueryCtrlImpl>(nullptr, rp);
+
+    auto impl = dynamic_pointer_cast<ScopeImpl>(shared_from_this());
+
+    auto send_activate = [impl, result, metadata, rp, ro, ctrl]() -> void
+    {
         try
         {
-            ro->finished(ListenerBase::Error, e.what());
-            throw;
+            auto real_ctrl = dynamic_pointer_cast<QueryCtrlImpl>(impl->fwd()->activate(result.p->activation_target(),
+                                                                                       metadata.serialize(),
+                                                                                       rp));
+            assert(real_ctrl);
+
+            auto new_proxy = dynamic_pointer_cast<MWQueryCtrl>(real_ctrl->proxy());
+            assert(new_proxy);
+            ctrl->set_proxy(new_proxy);
         }
-        catch (...)
+        catch (std::exception const& e)
         {
-            cerr << "activate(): unknown exception" << endl;
+            try
+            {
+                ro->finished(ListenerBase::Error, e.what());
+            }
+            catch (...)
+            {
+            }
         }
-        throw;
-    }
+    };
+
+    auto future = runtime_->async_pool()->submit(send_activate);
+    runtime_->future_queue()->push(move(future));
     return ctrl;
 }
 
@@ -170,35 +197,43 @@
         throw unity::InvalidArgumentException("Scope::perform_action(): invalid ActivationListenerBase (nullptr)");
     }
 
-    QueryCtrlProxy ctrl;
-    try
-    {
-        // Create a middleware server-side object that can receive incoming
-        // push() and finished() messages over the network.
-        ActivationReplyObject::SPtr ro(make_shared<ActivationReplyObject>(reply, runtime_, to_string()));
-        MWReplyProxy rp = fwd()->mw_base()->add_reply_object(ro);
-
-        // Forward the activate() method across the bus.
-        ctrl = fwd()->perform_action(result.p->activation_target(), metadata.serialize(), widget_id, action_id, rp);
-        assert(ctrl);
-    }
-    catch (std::exception const& e)
-    {
-        // TODO: log error
-        cerr << "perform_action(): " << e.what() << endl;
+    ActivationReplyObject::SPtr ro(make_shared<ActivationReplyObject>(reply, runtime_, to_string()));
+    MWReplyProxy rp = fwd()->mw_base()->add_reply_object(ro);
+
+    shared_ptr<QueryCtrlImpl> ctrl = make_shared<QueryCtrlImpl>(nullptr, rp);
+
+    auto impl = dynamic_pointer_cast<ScopeImpl>(shared_from_this());
+
+    auto send_perform_action = [impl, result, metadata, widget_id, action_id, rp, ro, ctrl]() -> void
+    {
         try
         {
-            // TODO: if things go wrong, we need to make sure that the reply object
-            // is disconnected from the middleware, so it gets deallocated.
-            reply->finished(ListenerBase::Error, e.what());
-            throw;
+            auto real_ctrl = dynamic_pointer_cast<QueryCtrlImpl>(impl->fwd()->perform_action(
+                                                                               result.p->activation_target(),
+                                                                               metadata.serialize(),
+                                                                               widget_id,
+                                                                               action_id,
+                                                                               rp));
+            assert(real_ctrl);
+
+            auto new_proxy = dynamic_pointer_cast<MWQueryCtrl>(real_ctrl->proxy());
+            assert(new_proxy);
+            ctrl->set_proxy(new_proxy);
         }
-        catch (...)
+        catch (std::exception const& e)
         {
-            cerr << "perform_action(): unknown exception" << endl;
+            try
+            {
+                ro->finished(ListenerBase::Error, e.what());
+            }
+            catch (...)
+            {
+            }
         }
-        throw;
-    }
+    };
+
+    auto future = runtime_->async_pool()->submit(send_perform_action);
+    runtime_->future_queue()->push(move(future));
     return ctrl;
 }
 
@@ -211,38 +246,40 @@
         throw unity::InvalidArgumentException("Scope::preview(): invalid PreviewListenerBase (nullptr)");
     }
 
-    QueryCtrlProxy ctrl;
     PreviewReplyObject::SPtr ro(make_shared<PreviewReplyObject>(reply, runtime_, to_string()));
-    try
-    {
-        // Create a middleware server-side object that can receive incoming
-        // push() and finished() messages over the network.
-        MWReplyProxy rp = fwd()->mw_base()->add_reply_object(ro);
-
-        // Forward the search() method across the bus. This is a
-        // synchronous twoway interaction with the scope, so it can return
-        // the QueryCtrlProxy. Because the Scope implementation has a separate
-        // thread for search() calls, this is guaranteed not to block for
-        // any length of time. (No application code other than the QueryBase constructor
-        // is called by search() on the server side.)
-        ctrl = fwd()->preview(result.p->activation_target(), hints.serialize(), rp);
-        assert(ctrl);
-    }
-    catch (std::exception const& e)
-    {
-        // TODO: log error
-        cerr << "preview(): " << e.what() << endl;
+    MWReplyProxy rp = fwd()->mw_base()->add_reply_object(ro);
+
+    shared_ptr<QueryCtrlImpl> ctrl = make_shared<QueryCtrlImpl>(nullptr, rp);
+
+    auto impl = dynamic_pointer_cast<ScopeImpl>(shared_from_this());
+
+    auto send_preview = [impl, result, hints, rp, ro, ctrl]() -> void
+    {
         try
         {
-            ro->finished(ListenerBase::Error, e.what());
-            throw;
+            auto real_ctrl = dynamic_pointer_cast<QueryCtrlImpl>(impl->fwd()->preview(result.p->activation_target(),
+                                                                                      hints.serialize(),
+                                                                                      rp));
+            assert(real_ctrl);
+
+            auto new_proxy = dynamic_pointer_cast<MWQueryCtrl>(real_ctrl->proxy());
+            assert(new_proxy);
+            ctrl->set_proxy(new_proxy);
         }
-        catch (...)
+        catch (std::exception const& e)
         {
-            cerr << "preview(): unknown exception" << endl;
+            try
+            {
+                ro->finished(ListenerBase::Error, e.what());
+            }
+            catch (...)
+            {
+            }
         }
-        throw;
-    }
+    };
+
+    auto future = runtime_->async_pool()->submit(send_preview);
+    runtime_->future_queue()->push(move(future));
     return ctrl;
 }
 
@@ -251,7 +288,7 @@
     return make_shared<ScopeImpl>(mw_proxy, runtime, scope_id);
 }
 
-MWScopeProxy ScopeImpl::fwd() const
+MWScopeProxy ScopeImpl::fwd()
 {
     return dynamic_pointer_cast<MWScope>(proxy());
 }

=== modified file 'src/scopes/internal/ScopeObject.cpp'
--- src/scopes/internal/ScopeObject.cpp	2014-03-10 04:17:09 +0000
+++ src/scopes/internal/ScopeObject.cpp	2014-04-28 05:57:56 +0000
@@ -68,7 +68,6 @@
         // to be safe, we don't assert, in case someone is running a broken client.
 
         // TODO: log error about incoming request containing an invalid reply proxy.
-
         throw LogicException("Scope \"" + runtime_->scope_id() + "\": query() called with null reply proxy");
     }
 

=== modified file 'src/scopes/internal/SearchReplyImpl.cpp'
--- src/scopes/internal/SearchReplyImpl.cpp	2014-03-10 10:34:43 +0000
+++ src/scopes/internal/SearchReplyImpl.cpp	2014-04-28 05:57:56 +0000
@@ -38,11 +38,13 @@
 namespace internal
 {
 
-SearchReplyImpl::SearchReplyImpl(MWReplyProxy const& mw_proxy, std::shared_ptr<QueryObjectBase> const& qo) :
+SearchReplyImpl::SearchReplyImpl(MWReplyProxy const& mw_proxy,
+                                 std::shared_ptr<QueryObjectBase> const& qo,
+                                 int cardinality) :
     ObjectImpl(mw_proxy),
     ReplyImpl(mw_proxy, qo),
     cat_registry_(new CategoryRegistry()),
-    cardinality_(qo->cardinality({ fwd()->identity(), fwd()->mw_base() })),
+    cardinality_(cardinality),
     num_pushes_(0)
 {
 }

=== modified file 'src/scopes/internal/ThreadPool.cpp'
--- src/scopes/internal/ThreadPool.cpp	2014-01-23 11:28:34 +0000
+++ src/scopes/internal/ThreadPool.cpp	2014-04-28 05:57:56 +0000
@@ -54,7 +54,6 @@
             threads_.push_back(std::thread(&ThreadPool::run, this));
         }
         auto future = threads_ready_.get_future();
-        future.wait();
         future.get();
     }
     catch (std::exception const&)   // LCOV_EXCL_LINE
@@ -66,11 +65,18 @@
 ThreadPool::~ThreadPool()
 {
     queue_->destroy();
+
+    vector<thread> threads;
+    {
+        lock_guard<mutex> lock(mutex_);
+        threads.swap(threads_);
+    }
+
     try
     {
-        for (size_t i = 0; i < threads_.size(); ++i)
+        for (size_t i = 0; i < threads.size(); ++i)
         {
-            threads_[i].join();
+            threads[i].join();
         }
     }
     catch (...) // LCOV_EXCL_LINE
@@ -81,18 +87,19 @@
 
 void ThreadPool::run()
 {
-    TaskQueue::value_type task;
+    {
+        lock_guard<mutex> lock(mutex_);
+        if (--num_threads_ == 0)
+        {
+            threads_ready_.set_value();
+        }
+    }
+
     for (;;)
     {
+        TaskQueue::value_type task;  // Task must go out of scope in each iteration, in case it stores shared_ptrs.
         try
         {
-            {
-                lock_guard<mutex> lock(mutex_);
-                if (--num_threads_ == 0)
-                {
-                    threads_ready_.set_value();
-                }
-            }
             task = queue_->wait_and_pop();
         }
         catch (runtime_error const&)

=== modified file 'src/scopes/internal/smartscopes/SSQueryObject.cpp'
--- src/scopes/internal/smartscopes/SSQueryObject.cpp	2014-03-07 03:49:03 +0000
+++ src/scopes/internal/smartscopes/SSQueryObject.cpp	2014-04-28 05:57:56 +0000
@@ -206,7 +206,7 @@
 
     // Create the reply proxy and keep a weak_ptr, which we will need
     // if cancel() is called later.
-    q_reply_proxy = make_shared<SearchReplyImpl>(reply, shared_from_this());
+    q_reply_proxy = make_shared<SearchReplyImpl>(reply, shared_from_this(), query->q_cardinality);
     assert(q_reply_proxy);
     query->q_reply_proxy = q_reply_proxy;
 

=== modified file 'src/scopes/internal/zmq_middleware/CMakeLists.txt'
--- src/scopes/internal/zmq_middleware/CMakeLists.txt	2014-04-02 09:24:02 +0000
+++ src/scopes/internal/zmq_middleware/CMakeLists.txt	2014-04-28 05:57:56 +0000
@@ -13,6 +13,7 @@
     ${CMAKE_CURRENT_SOURCE_DIR}/RethrowException.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/ScopeI.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/ServantBase.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/StopPublisher.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/StateReceiverI.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/Util.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/VariantConverter.cpp

=== modified file 'src/scopes/internal/zmq_middleware/ConnectionPool.cpp'
--- src/scopes/internal/zmq_middleware/ConnectionPool.cpp	2014-01-22 03:29:48 +0000
+++ src/scopes/internal/zmq_middleware/ConnectionPool.cpp	2014-04-28 05:57:56 +0000
@@ -36,8 +36,8 @@
 namespace zmq_middleware
 {
 
-ConnectionPool::ConnectionPool(zmqpp::context& context) :
-    context_(context)
+ConnectionPool::ConnectionPool(zmqpp::context& context)
+    : context_(context)
 {
 }
 
@@ -50,9 +50,7 @@
 {
     assert(!endpoint.empty());
 
-    lock_guard<mutex> lock(mutex_);
-
-    // Look for existing connection. If there is one, but with the wrong type, we throw.
+    // Look for existing connection.
     auto const& it = pool_.find(endpoint);
     if (it != pool_.end())
     {
@@ -66,19 +64,30 @@
     }
 
     // No existing connection yet, establish one.
+    auto entry = create_connection(endpoint, m, -1);
+    return pool_.emplace(move(entry)).first->second.socket;
+}
+
+ConnectionPool::CPool::value_type ConnectionPool::create_connection(std::string const& endpoint,
+                                                    RequestMode m,
+                                                    int64_t timeout)
+{
+    if (timeout == -1)
+    {
+        timeout = 0;    // Don't linger on close
+    }
     zmqpp::socket_type stype = m == RequestMode::Twoway ? zmqpp::socket_type::request : zmqpp::socket_type::push;
     zmqpp::socket s(context_, stype);
-    s.set(zmqpp::socket_option::linger, 0);
+    auto zmqpp_timeout = m == RequestMode::Twoway ? int32_t(timeout) : 0;
+    s.set(zmqpp::socket_option::linger, zmqpp_timeout);
     s.connect(endpoint);
-    return pool_.emplace(endpoint, Connection { move(s), m }).first->second.socket;
+    return CPool::value_type{ endpoint, SocketData{ move(s), m } };
 }
 
 void ConnectionPool::remove(std::string const& endpoint)
 {
     assert(!endpoint.empty());
 
-    lock_guard<mutex> lock(mutex_);
-
     auto const& it = pool_.find(endpoint);
     if (it != pool_.end())
     {
@@ -90,8 +99,7 @@
 {
     assert(!endpoint.empty());
 
-    lock_guard<mutex> lock(mutex_);
-    pool_.emplace(endpoint, Connection { move(socket), m });
+    pool_.emplace(endpoint, SocketData{ move(socket), m });
 }
 
 } // namespace zmq_middleware

=== modified file 'src/scopes/internal/zmq_middleware/ObjectAdapter.cpp'
--- src/scopes/internal/zmq_middleware/ObjectAdapter.cpp	2014-02-20 19:19:25 +0000
+++ src/scopes/internal/zmq_middleware/ObjectAdapter.cpp	2014-04-28 05:57:56 +0000
@@ -19,6 +19,7 @@
 #include <unity/scopes/internal/zmq_middleware/ObjectAdapter.h>
 
 #include <unity/scopes/internal/zmq_middleware/ServantBase.h>
+#include <unity/scopes/internal/zmq_middleware/StopPublisher.h>
 #include <unity/scopes/internal/zmq_middleware/Util.h>
 #include <unity/scopes/internal/zmq_middleware/ZmqException.h>
 #include <unity/scopes/internal/zmq_middleware/ZmqReceiver.h>
@@ -134,7 +135,7 @@
         lock_guard<mutex> state_lock(state_mutex_, adopt_lock);
         if (state_ == Destroyed || state_ == Failed)
         {
-            throw_bad_state(state_);
+            throw_bad_state("add()", state_);
         }
     }
 
@@ -158,7 +159,7 @@
             lock_guard<mutex> state_lock(state_mutex_, adopt_lock);
             if (state_ == Destroyed || state_ == Failed)
             {
-                throw_bad_state(state_);
+                throw_bad_state("remove()", state_);
             }
         }
 
@@ -185,7 +186,7 @@
         lock_guard<mutex> state_lock(state_mutex_, adopt_lock);
         if (state_ == Destroyed || state_ == Failed)
         {
-            throw_bad_state(state_);
+            throw_bad_state("find()", state_);
         }
     }
 
@@ -210,7 +211,7 @@
         lock_guard<mutex> state_lock(state_mutex_, adopt_lock);
         if (state_ == Destroyed || state_ == Failed)
         {
-            throw_bad_state(state_);
+            throw_bad_state("add_dflt_servant()", state_);
         }
     }
 
@@ -234,7 +235,7 @@
             lock_guard<mutex> state_lock(state_mutex_, adopt_lock);
             if (state_ == Destroyed || state_ == Failed)
             {
-                throw_bad_state(state_);
+                throw_bad_state("remove_dflt_servant()", state_);
             }
         }
 
@@ -262,7 +263,7 @@
         lock_guard<mutex> state_lock(state_mutex_, adopt_lock);
         if (state_ == Destroyed || state_ == Failed)
         {
-            throw_bad_state(state_);
+            throw_bad_state("find_dflt_servant()", state_);
         }
     }
 
@@ -308,7 +309,7 @@
         case Destroyed:
         case Failed:
         {
-            throw_bad_state(state_);
+            throw_bad_state("activate()", state_);
         }
         default:
         {
@@ -329,7 +330,7 @@
         }
         case Failed:
         {
-            throw_bad_state(state_);
+            throw_bad_state("shutdown() [state_ == Failed]", state_);
         }
         case Inactive:
         {
@@ -349,7 +350,7 @@
             }
             if (state_ == Failed)
             {
-                throw_bad_state(state_);
+                throw_bad_state("shutdown() [state == Activating]", state_);
             }
             // LCOV_EXCL_STOP
 
@@ -358,7 +359,7 @@
         case Active:
         {
             state_ = Deactivating;
-            stop_workers();
+            stopper_->stop();
             state_changed_.notify_all();
             break;
         }
@@ -388,11 +389,11 @@
 
     if (state == Failed)
     {
-        throw_bad_state(state);
+        throw_bad_state("wait_for_shutdown()", state);
     }
 }
 
-void ObjectAdapter::throw_bad_state(AdapterState state) const
+void ObjectAdapter::throw_bad_state(string const& label, AdapterState state) const
 {
     string bad_state;
     switch (state)
@@ -420,16 +421,31 @@
             break;
         }
     }
-    MiddlewareException e("Object adapter in " + bad_state + " state (adapter: " + name_ + ")");
+    MiddlewareException e(label + ": Object adapter in " + bad_state + " state (adapter: " + name_ + ")");
     e.remember(exception_);  // Remember any exception encountered by the broker thread or a worker thread.
     throw e;
 }
 
 void ObjectAdapter::run_workers()
 {
+    {
+        lock_guard<mutex> state_lock(state_mutex_);
+        try
+        {
+            // Start the publisher for stop messages.
+            stopper_.reset(new StopPublisher(mw_.context(), name_ + "-stopper"));
+        }
+        catch (...)
+        {
+            state_ = Failed;
+            state_changed_.notify_all();
+            throw MiddlewareException("ObjectAdapter::run_workers(): stop thread failure (adapter: " + name_ + ")"); // LCOV_EXCL_LINE
+        }
+    }
+
     // Start a broker thread to forward incoming messages to backend workers and
     // wait for the broker thread to finish its initialization. The broker
-    // signals after it has connected to the ctrl socket.
+    // signals after it has connected to the stop socket.
     {
         lock_guard<mutex> lock(ready_mutex_);
         ready_ = promise<void>();
@@ -469,56 +485,13 @@
         // LCOV_EXCL_START
         catch (...) //
         {
-            stop_workers();
+            stopper_->stop();
             throw MiddlewareException("ObjectAdapter::run_workers(): worker thread failure (adapter: " + name_ + ")");
         }
         // LCOV_EXCL_STOP
     }
 }
 
-void ObjectAdapter::init_ctrl_socket()
-{
-    lock_guard<mutex> lock(ctrl_mutex_);
-
-    // PUB socket to let the broker and workers know when it is time to shut down.
-    // Sending anything means the socket becomes ready for reading, which causes
-    // the broker and worker threads to finish.
-    ctrl_.reset(new zmqpp::socket(*mw_.context(), zmqpp::socket_type::publish));
-    ctrl_->set(zmqpp::socket_option::linger, 0);
-    ctrl_->bind("inproc://" + name_ + "_adapter_ctrl");
-}
-
-zmqpp::socket ObjectAdapter::subscribe_to_ctrl_socket()
-{
-    zmqpp::socket ctrl(*mw_.context(), zmqpp::socket_type::subscribe);
-    ctrl.set(zmqpp::socket_option::linger, 0);
-    ctrl.connect("inproc://" + name_ + "_adapter_ctrl"); // Once a thread can read from here, that's the command to stop.
-    ctrl.subscribe("");
-    return move(ctrl);
-}
-
-void ObjectAdapter::stop_workers() noexcept
-{
-    lock_guard<mutex> lock(ctrl_mutex_);
-
-    try
-    {
-        ctrl_->send("stop");
-    }
-    // LCOV_EXCL_START
-    catch (std::exception const& e)
-    {
-        // TODO: log this instead
-        cerr << "ObjectAdapter::stop_workers(): " << e.what() << endl;
-    }
-    catch (...)
-    {
-        // TODO: log this instead
-        cerr << "ObjectAdapter::stop_workers(): unknown exception" << endl;
-    }
-    // LCOV_EXCL_STOP
-}
-
 // For the ipc transport, zmq permits more than one server to bind to the same endpoint.
 // If a server binds to an endpoint while another server is using that endpoint, the
 // second server silently "steals" the endpoint from the previous server, so all
@@ -574,12 +547,8 @@
 {
     try
     {
-        // Create the writing end of the ctrl socket. We need to do this before subscribing because, for inproc
-        // pub-sub sockets, the bind must happen before the connect.
-        init_ctrl_socket();
-
-        // Subscribe to ctrl socket. Once this socket becomes readable, that's the command to finish.
-        auto ctrl = subscribe_to_ctrl_socket();
+        // Subscribe to stop socket. Once this socket becomes readable, that's the command to finish.
+        auto stop = stopper_->subscribe();
 
         // Set up message pump. Router-dealer for twoway adapter, pull-push for oneway adapter.
         auto socket_type = mode_ == RequestMode::Twoway ? zmqpp::socket_type::router : zmqpp::socket_type::pull;
@@ -592,7 +561,7 @@
 
         try
         {
-            poller.add(ctrl);
+            poller.add(stop);
 
             frontend.set(zmqpp::socket_option::linger, 0);
             // "Safe" bind: prevents two servers from binding to the same endpoint.
@@ -625,13 +594,13 @@
         {
             zmqpp::message message;
             poller.poll();
-            if (poller.has_input(ctrl))
+            if (poller.has_input(stop))
             {
-                // When the ctrl socket becomes ready, we need to get out of here.
+                // When the stop socket becomes ready, we need to get out of here.
                 // We stop reading more requests from the router, but continue processing
                 // while there are still replies outstanding.
-                ctrl.receive(message);
-                ctrl.close();
+                poller.remove(stop);
+                stop.close();
                 shutting_down = true;
             }
             if (!shutting_down && poller.has_input(frontend)) // Once shutting down, we no longer read incoming messages.
@@ -696,9 +665,9 @@
     {
         zmqpp::poller poller;
 
-        // Subscribe to ctrl socket. Once this socket becomes readable, that's the command to finish.
-        auto ctrl = subscribe_to_ctrl_socket();
-        poller.add(ctrl);
+        // Subscribe to stop socket. Once this socket becomes readable, that's the command to finish.
+        auto stop = stopper_->subscribe();
+        poller.add(stop);
 
         auto socket_type = mode_ == RequestMode::Twoway ? zmqpp::socket_type::reply : zmqpp::socket_type::pull;
         zmqpp::socket s(*mw_.context(), socket_type);
@@ -728,11 +697,10 @@
             }
 
             poller.poll();
-            if (poller.has_input(ctrl)) // Parent sent a stop message, so we are supposed to go away.
+            if (poller.has_input(stop)) // Parent sent a stop message, so we are supposed to go away.
             {
-                zmqpp::message message;
-                ctrl.receive(message);
-                ctrl.close();
+                poller.remove(stop);
+                stop.close();
                 finish = true;
             }
 
@@ -810,7 +778,16 @@
                 }
 
                 // Look for a servant with matching id.
-                auto servant = find_servant(current.id, current.category);
+                shared_ptr<ServantBase> servant;
+                try
+                {
+                    servant = find_servant(current.id, current.category);
+                }
+                catch (std::exception const&)
+                {
+                    // Ignore failure to find servant during destruction phase.
+                }
+
                 if (!servant)
                 {
                     if (mode_ == RequestMode::Twoway)
@@ -838,7 +815,7 @@
     // LCOV_EXCL_START
     catch (...)
     {
-        stop_workers();  // Fatal error, we need to stop all other workers and the broker.
+        stopper_->stop();  // Fatal error, we need to stop all other workers and the broker.
         MiddlewareException e("ObjectAdapter: worker thread failure (adapter: " + name_ + ")");
         store_exception(e);
         // We may not have signaled the parent yet, depending on where things went wrong.

=== added file 'src/scopes/internal/zmq_middleware/StopPublisher.cpp'
--- src/scopes/internal/zmq_middleware/StopPublisher.cpp	1970-01-01 00:00:00 +0000
+++ src/scopes/internal/zmq_middleware/StopPublisher.cpp	2014-04-28 05:57:56 +0000
@@ -0,0 +1,225 @@
+/*
+ * Copyright (C) 2014 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: Michi Henning <[e-mail]>
+ */
+
+#include <unity/scopes/internal/zmq_middleware/StopPublisher.h>
+
+#include <unity/scopes/ScopeExceptions.h>
+
+#include <cassert>
+#include <iostream>
+
+using namespace std;
+
+namespace unity
+{
+
+namespace scopes
+{
+
+namespace internal
+{
+
+namespace zmq_middleware
+{
+
+StopPublisher::StopPublisher(zmqpp::context* context, string const& inproc_name)
+    : context_(context)
+    , endpoint_("inproc://" + inproc_name)
+    , state_(Starting)
+{
+    assert(context);
+    assert(!inproc_name.empty());
+
+    thread_ = thread(&StopPublisher::stopper_thread, this);
+
+    unique_lock<mutex> lock(m_);
+    cond_.wait(lock, [this] { return state_ == Started || state_ == Failed; });
+
+    if (state_ == Failed)
+    {
+        if (thread_.joinable())
+        {
+            thread_.join();
+        }
+        try
+        {
+            rethrow_exception(ex_);
+        }
+        catch (...)
+        {
+            throw MiddlewareException("StopPublisher(): publisher thread failed (endpoint: " + endpoint_ + ")");
+        }
+    }
+}
+
+StopPublisher::~StopPublisher()
+{
+    try
+    {
+        stop();
+    }
+    // LCOV_EXCL_START
+    catch (std::exception const& e)
+    {
+        cerr << "~StopPublisher(): exception from stop(): " << e.what() << endl;
+        // TODO: log this
+    }
+    catch (...)
+    {
+        cerr << "~StopPublisher(): unknown exception from stop()" << endl;
+        // TODO: log this
+    }
+    // LCOV_EXCL_STOP
+    if (thread_.joinable())
+    {
+        thread_.join();
+    }
+}
+
+string StopPublisher::endpoint() const
+{
+    lock_guard<mutex> lock(m_);
+    return endpoint_;
+}
+
+// Return a socket that becomes ready for reading once stop() is called.
+// Once the socket becomes readable, it has exactly one zero-length message to be read.
+
+zmqpp::socket StopPublisher::subscribe()
+{
+    lock_guard<mutex> lock(m_);
+
+    switch (state_)
+    {
+        case Stopping:
+        case Stopped:
+        {
+            throw MiddlewareException(
+                "StopPublisher::subscribe(): cannot subscribe to stopped publisher "
+                "(endpoint: " + endpoint_ + ")");
+        }
+        // LCOV_EXCL_START
+        case Failed:
+        {
+            try
+            {
+                rethrow_exception(ex_);
+            }
+            catch (...)
+            {
+                throw MiddlewareException(
+                    "StopPublisher::subscribe(): cannot subscribe to failed publisher "
+                    "(endpoint: " + endpoint_ + ")");
+            }
+        }
+        // LCOV_EXCL_STOP
+        default:
+        {
+            assert(state_ == Started);
+            zmqpp::socket s(*context_, zmqpp::socket_type::subscribe);
+            s.set(zmqpp::socket_option::linger, 0);
+            s.connect(endpoint_);
+            s.subscribe("");
+            return move(s);
+        }
+    }
+}
+
+// Tell the stopper thread to publish the stop message and terminate itself.
+
+void StopPublisher::stop()
+{
+    unique_lock<mutex> lock(m_);
+    switch (state_)
+    {
+        case Stopping:
+        case Stopped:
+        {
+            return;  // no-op
+        }
+        // LCOV_EXCL_START
+        case Failed:
+        {
+            try
+            {
+                rethrow_exception(ex_);
+            }
+            catch (...)
+            {
+                throw MiddlewareException("StopPublisher::stop(): cannot stop failed publisher");
+            }
+        }
+        // LCOV_EXCL_STOP
+        default:
+        {
+            assert(state_ == Started);
+            state_ = Stopping;
+            cond_.notify_all();
+        }
+    }
+}
+
+void StopPublisher::stopper_thread() noexcept
+{
+    try
+    {
+        unique_lock<mutex> lock(m_);
+
+        assert(state_ == Starting);
+
+        // Create the publishing socket.
+        zmqpp::socket pub_socket(zmqpp::socket(*context_, zmqpp::socket_type::publish));
+        // Allow time for stop message to be buffered, so close() won't discard it.
+        pub_socket.set(zmqpp::socket_option::linger, 5000);
+        pub_socket.bind(endpoint_);
+
+        // Notify that we are ready. This ensures that subscribers
+        // cannot subscribe before the pub socket was bound.
+        state_ = Started;
+        cond_.notify_all();
+
+        lock.unlock();  // Allow parent to wake up.
+
+        // Wait until we are told to stop.
+        lock.lock();
+        cond_.wait(lock, [this] { return state_ == Stopping; });
+
+        // Write the stop message for the subscribers and close.
+        // Sending an empty string is OK; the receiver will get a zero-length message.
+        pub_socket.send("");
+        pub_socket.close();
+        state_ = Stopped;
+    }
+    catch (...)
+    {
+        lock_guard<mutex> lock(m_);
+        state_ = Failed;
+        ex_ = current_exception();
+        cond_.notify_all();
+    }
+
+    assert(state_ == Stopped || state_ == Failed);
+}
+
+}  // namespace zmq_middleware
+
+}  // namespace internal
+
+}  // namespace scopes
+
+}  // namespace unity

=== modified file 'src/scopes/internal/zmq_middleware/ZmqMiddleware.cpp'
--- src/scopes/internal/zmq_middleware/ZmqMiddleware.cpp	2014-04-02 09:24:02 +0000
+++ src/scopes/internal/zmq_middleware/ZmqMiddleware.cpp	2014-04-28 05:57:56 +0000
@@ -434,9 +434,9 @@
         shared_ptr<QueryCtrlI> qci(make_shared<QueryCtrlI>(ctrl));
         auto adapter = find_adapter(server_name_ + ctrl_suffix, config_.private_dir());
         function<void()> df;
-        auto proxy = safe_add(df, adapter, "", qci);
+        auto p = safe_add(df, adapter, "", qci);
         ctrl->set_disconnect_function(df);
-        return ZmqQueryCtrlProxy(new ZmqQueryCtrl(this, proxy->endpoint(), proxy->identity(), "QueryCtrl"));
+        proxy = ZmqQueryCtrlProxy(new ZmqQueryCtrl(this, p->endpoint(), p->identity(), "QueryCtrl"));
     }
     catch (std::exception const& e) // Should never happen unless our implementation is broken
     {
@@ -476,9 +476,9 @@
         shared_ptr<QueryI> qi(make_shared<QueryI>(query));
         auto adapter = find_adapter(server_name_ + query_suffix, config_.private_dir());
         function<void()> df;
-        auto proxy = safe_add(df, adapter, "", qi);
+        auto p = safe_add(df, adapter, "", qi);
         query->set_disconnect_function(df);
-        return ZmqQueryProxy(new ZmqQuery(this, proxy->endpoint(), proxy->identity(), "Query"));
+        proxy = ZmqQueryProxy(new ZmqQuery(this, p->endpoint(), p->identity(), "Query"));
     }
     catch (std::exception const& e) // Should never happen unless our implementation is broken
     {
@@ -519,9 +519,9 @@
         shared_ptr<RegistryI> ri(make_shared<RegistryI>(registry));
         auto adapter = find_adapter(server_name_, runtime()->registry_endpointdir());
         function<void()> df;
-        auto proxy = safe_add(df, adapter, identity, ri);
+        auto p = safe_add(df, adapter, identity, ri);
         registry->set_disconnect_function(df);
-        return ZmqRegistryProxy(new ZmqRegistry(this, proxy->endpoint(), proxy->identity(), "Registry", twoway_timeout_));
+        proxy = ZmqRegistryProxy(new ZmqRegistry(this, p->endpoint(), p->identity(), "Registry", twoway_timeout_));
     }
     catch (std::exception const& e) // Should never happen unless our implementation is broken
     {
@@ -542,9 +542,9 @@
         shared_ptr<ReplyI> ri(make_shared<ReplyI>(reply));
         auto adapter = find_adapter(server_name_ + reply_suffix, config_.public_dir());
         function<void()> df;
-        auto proxy = safe_add(df, adapter, "", ri);
+        auto p = safe_add(df, adapter, "", ri);
         reply->set_disconnect_function(df);
-        return ZmqReplyProxy(new ZmqReply(this, proxy->endpoint(), proxy->identity(), "Reply"));
+        proxy = ZmqReplyProxy(new ZmqReply(this, p->endpoint(), p->identity(), "Reply"));
     }
     catch (std::exception const& e) // Should never happen unless our implementation is broken
     {
@@ -566,9 +566,9 @@
         shared_ptr<ScopeI> si(make_shared<ScopeI>(scope));
         auto adapter = find_adapter(server_name_, config_.private_dir());
         function<void()> df;
-        auto proxy = safe_add(df, adapter, identity, si);
+        auto p = safe_add(df, adapter, identity, si);
         scope->set_disconnect_function(df);
-        return ZmqScopeProxy(new ZmqScope(this, proxy->endpoint(), proxy->identity(), "Scope", twoway_timeout_));
+        proxy = ZmqScopeProxy(new ZmqScope(this, p->endpoint(), p->identity(), "Scope", twoway_timeout_));
     }
     catch (std::exception const& e) // Should never happen unless our implementation is broken
     {

=== modified file 'src/scopes/internal/zmq_middleware/ZmqObject.cpp'
--- src/scopes/internal/zmq_middleware/ZmqObject.cpp	2014-02-03 07:28:36 +0000
+++ src/scopes/internal/zmq_middleware/ZmqObject.cpp	2014-04-28 05:57:56 +0000
@@ -18,7 +18,6 @@
 
 #include <unity/scopes/internal/zmq_middleware/ZmqObjectProxy.h>
 
-#include <unity/scopes/internal/zmq_middleware/ConnectionPool.h>
 #include <unity/scopes/internal/zmq_middleware/Util.h>
 #include <unity/scopes/internal/zmq_middleware/ZmqException.h>
 #include <unity/scopes/internal/zmq_middleware/ZmqSender.h>
@@ -127,7 +126,8 @@
     capnp::MallocMessageBuilder request_builder;
     make_request_(request_builder, "ping");
 
-    auto future = mw_base()->invoke_pool()->submit([&] { this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { this->invoke_twoway_(request_builder); });
+    // TODO: dubious, waiter thread in runtimeImpl should do this?
     future.wait();
 }
 
@@ -148,59 +148,82 @@
     return request;
 }
 
+#ifdef ENABLE_IPC_MONITOR
 void register_monitor_socket (ConnectionPool& pool, zmqpp::context_t const& context)
 {
     thread_local static bool monitor_initialized = false;
     if (!monitor_initialized) {
         monitor_initialized = true;
         zmqpp::socket monitor_socket(context, zmqpp::socket_type::publish);
+        monitor_socket.set(zmqpp::socket_option::linger, 0);
         monitor_socket.connect(MONITOR_ENDPOINT);
-        monitor_socket.set(zmqpp::socket_option::linger, 0);
         pool.register_socket(MONITOR_ENDPOINT, move(monitor_socket), RequestMode::Oneway);
     }
 }
-
-ZmqReceiver ZmqObjectProxy::invoke_(capnp::MessageBuilder& out_params)
-{
-    return invoke_(out_params, timeout_);
-}
-
-// Get a socket to the endpoint for this proxy and write the request on the wire.
-// For a twoway request, poll for the reply with the timeout set for this proxy.
-// Return a receiver for the response (whether this is a oneway or twoway request).
-
-ZmqReceiver ZmqObjectProxy::invoke_(capnp::MessageBuilder& out_params, int64_t timeout)
-{
-    // Each calling thread gets its own pool because zmq sockets are not thread-safe.
-    thread_local static ConnectionPool pool(*mw_base()->context());
-
-    zmqpp::socket& s = pool.find(endpoint_, mode_);
-    ZmqSender sender(s);
-    auto segments = out_params.getSegmentsForOutput();
-    sender.send(segments);
-
-#ifdef ENABLE_IPC_MONITOR
-    if (true) {
-        register_monitor_socket(pool, *mw_base()->context());
-        zmqpp::socket& monitor = pool.find(MONITOR_ENDPOINT, RequestMode::Oneway);
-        auto word_arr = capnp::messageToFlatArray(segments);
-        monitor.send_raw(reinterpret_cast<char*>(&word_arr[0]), word_arr.size() * sizeof(capnp::word));
-    }
-#endif
-
-    if (mode_ == RequestMode::Twoway)
+#endif
+
+// Get a socket to the endpoint for this proxy and write the request on the wire.
+
+void ZmqObjectProxy::invoke_oneway_(capnp::MessageBuilder& out_params)
+{
+    // Each calling thread gets its own pool because zmq sockets are not thread-safe.
+    thread_local static ConnectionPool pool(*mw_base()->context());
+
+    assert(mode_ == RequestMode::Oneway);
+    zmqpp::socket& s = pool.find(endpoint_, mode_);
+    ZmqSender sender(s);
+    auto segments = out_params.getSegmentsForOutput();
+    sender.send(segments);
+
+#ifdef ENABLE_IPC_MONITOR
+    if (true) {
+        register_monitor_socket(pool, *mw_base()->context());
+        zmqpp::socket& monitor = pool.find(MONITOR_ENDPOINT, RequestMode::Oneway);
+        auto word_arr = capnp::messageToFlatArray(segments);
+        monitor.send_raw(reinterpret_cast<char*>(&word_arr[0]), word_arr.size() * sizeof(capnp::word));
+    }
+#endif
+}
+
+ZmqReceiver ZmqObjectProxy::invoke_twoway_(capnp::MessageBuilder& out_params)
+{
+    return invoke_twoway_(out_params, timeout_);
+}
+
+// Get a socket to the endpoint for this proxy and write the request on the wire.
+// Poll for the reply with the given timeout.
+// Return a socket for the response or throw if the timeout expires.
+
+ZmqReceiver ZmqObjectProxy::invoke_twoway_(capnp::MessageBuilder& out_params, int64_t timeout)
+{
+    // Each calling thread gets its own pool because zmq sockets are not thread-safe.
+    thread_local static ConnectionPool pool(*mw_base()->context());
+
+    assert(mode_ == RequestMode::Twoway);
+    zmqpp::socket& s = pool.find(endpoint_, mode_);
+    ZmqSender sender(s);
+    auto segments = out_params.getSegmentsForOutput();
+    sender.send(segments);
+
+#ifdef ENABLE_IPC_MONITOR
+    if (true) {
+        register_monitor_socket(pool, *mw_base()->context());
+        zmqpp::socket& monitor = pool.find(MONITOR_ENDPOINT, RequestMode::Oneway);
+        auto word_arr = capnp::messageToFlatArray(segments);
+        monitor.send_raw(reinterpret_cast<char*>(&word_arr[0]), word_arr.size() * sizeof(capnp::word));
+    }
+#endif
+
+    zmqpp::poller p;
+    p.add(s);
+    p.poll(timeout);
+    if (!p.has_input(s))
     {
-        zmqpp::poller p;
-        p.add(s);
-        p.poll(timeout);
-        if (!p.has_input(s))
-        {
-            // If a request times out, we must close the corresponding socket, otherwise
-            // zmq gets confused: the reply will never be read, so the socket ends up
-            // in a bad state.
-            pool.remove(endpoint_);
-            throw TimeoutException("Request timed out after " + std::to_string(timeout) + " milliseconds");
-        }
+        // If a request times out, we must trash the corresponding socket, otherwise
+        // zmq gets confused: the reply will never be read, so the socket ends up
+        // in a bad state.
+        pool.remove(endpoint_);
+        throw TimeoutException("Request timed out after " + std::to_string(timeout) + " milliseconds");
     }
     return ZmqReceiver(s);
 }

=== modified file 'src/scopes/internal/zmq_middleware/ZmqQuery.cpp'
--- src/scopes/internal/zmq_middleware/ZmqQuery.cpp	2014-01-24 00:26:43 +0000
+++ src/scopes/internal/zmq_middleware/ZmqQuery.cpp	2014-04-28 05:57:56 +0000
@@ -66,7 +66,7 @@
     proxy.setIdentity(rp->identity().c_str());
     proxy.setCategory(rp->category().c_str());
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_oneway_(request_builder); });
     future.wait();
 }
 

=== modified file 'src/scopes/internal/zmq_middleware/ZmqQueryCtrl.cpp'
--- src/scopes/internal/zmq_middleware/ZmqQueryCtrl.cpp	2014-01-24 00:26:43 +0000
+++ src/scopes/internal/zmq_middleware/ZmqQueryCtrl.cpp	2014-04-28 05:57:56 +0000
@@ -58,7 +58,7 @@
     capnp::MallocMessageBuilder request_builder;
     make_request_(request_builder, "cancel");
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_oneway_(request_builder); });
     future.wait();
 }
 
@@ -67,7 +67,7 @@
     capnp::MallocMessageBuilder request_builder;
     make_request_(request_builder, "destroy");
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_oneway_(request_builder); });
     future.wait();
 }
 

=== modified file 'src/scopes/internal/zmq_middleware/ZmqRegistry.cpp'
--- src/scopes/internal/zmq_middleware/ZmqRegistry.cpp	2014-04-03 12:57:25 +0000
+++ src/scopes/internal/zmq_middleware/ZmqRegistry.cpp	2014-04-28 05:57:56 +0000
@@ -85,7 +85,7 @@
     auto in_params = request.initInParams().getAs<capnproto::Registry::GetMetadataRequest>();
     in_params.setIdentity(scope_id.c_str());
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_twoway_(request_builder); });
     auto receiver = future.get();
     auto segments = receiver.receive();
     capnp::SegmentArrayMessageReader reader(segments);
@@ -119,7 +119,7 @@
     capnp::MallocMessageBuilder request_builder;
     make_request_(request_builder, "list");
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_twoway_(request_builder); });
     auto receiver = future.get();
     auto segments = receiver.receive();
     capnp::SegmentArrayMessageReader reader(segments);
@@ -149,7 +149,7 @@
 
     // locate uses a custom timeout because it needs to potentially fork/exec a scope.
     int64_t timeout = 1000; // TODO: get timeout from config
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder, timeout); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_twoway_(request_builder, timeout); });
     auto receiver = future.get();
     auto segments = receiver.receive();
     capnp::SegmentArrayMessageReader reader(segments);

=== modified file 'src/scopes/internal/zmq_middleware/ZmqReply.cpp'
--- src/scopes/internal/zmq_middleware/ZmqReply.cpp	2014-01-24 00:26:43 +0000
+++ src/scopes/internal/zmq_middleware/ZmqReply.cpp	2014-04-28 05:57:56 +0000
@@ -64,7 +64,7 @@
     auto resultBuilder = in_params.getResult();
     to_value_dict(result, resultBuilder);
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_oneway_(request_builder); });
     future.wait();
 }
 
@@ -99,7 +99,7 @@
     }
     in_params.setReason(r);
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_oneway_(request_builder); });
     future.wait();
 }
 

=== modified file 'src/scopes/internal/zmq_middleware/ZmqScope.cpp'
--- src/scopes/internal/zmq_middleware/ZmqScope.cpp	2014-03-07 02:01:47 +0000
+++ src/scopes/internal/zmq_middleware/ZmqScope.cpp	2014-04-28 05:57:56 +0000
@@ -27,6 +27,8 @@
 #include <unity/scopes/Result.h>
 #include <unity/scopes/CannedQuery.h>
 
+#include <iostream> // TODO: remove this
+
 using namespace std;
 
 namespace unity
@@ -90,7 +92,7 @@
         p.setCategory(reply_proxy->category().c_str());
     }
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_twoway_(request_builder); });
 
     auto receiver = future.get();
     auto segments = receiver.receive();
@@ -122,7 +124,7 @@
         p.setIdentity(reply_proxy->identity().c_str());
     }
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_twoway_(request_builder); });
 
     auto receiver = future.get();
     auto segments = receiver.receive();
@@ -157,7 +159,7 @@
         p.setIdentity(reply_proxy->identity().c_str());
     }
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_twoway_(request_builder); });
     future.wait();
 
     auto receiver = future.get();
@@ -190,7 +192,7 @@
         p.setIdentity(reply_proxy->identity().c_str());
     }
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_twoway_(request_builder); });
 
     auto receiver = future.get();
     auto segments = receiver.receive();

=== modified file 'src/scopes/internal/zmq_middleware/ZmqStateReceiver.cpp'
--- src/scopes/internal/zmq_middleware/ZmqStateReceiver.cpp	2014-04-02 09:24:02 +0000
+++ src/scopes/internal/zmq_middleware/ZmqStateReceiver.cpp	2014-04-28 05:57:56 +0000
@@ -83,7 +83,7 @@
     in_params.setState(s);
     in_params.setSenderId(sender_id);
 
-    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
+    auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_oneway_(request_builder); });
     future.wait();
 }
 

=== modified file 'test/gtest/scopes/CMakeLists.txt'
--- test/gtest/scopes/CMakeLists.txt	2014-04-01 16:58:32 +0000
+++ test/gtest/scopes/CMakeLists.txt	2014-04-28 05:57:56 +0000
@@ -1,23 +1,24 @@
 add_subdirectory(internal)
 add_subdirectory(testing)
 
-add_subdirectory(Registry)
-add_subdirectory(Runtime)
 add_subdirectory(Activation)
 add_subdirectory(ActivationResponse)
 add_subdirectory(Annotation)
 add_subdirectory(CannedQuery)
-add_subdirectory(Department)
-add_subdirectory(Filters)
-add_subdirectory(ScopeBase)
-add_subdirectory(ScopeExceptions)
+add_subdirectory(CategorisedResult)
 add_subdirectory(Category)
-add_subdirectory(CategorisedResult)
 add_subdirectory(CategoryRenderer)
 add_subdirectory(ColumnLayout)
+add_subdirectory(Department)
+add_subdirectory(Filters)
+add_subdirectory(Invocation)
+add_subdirectory(OptionSelectorFilter)
 add_subdirectory(PreviewWidget)
 add_subdirectory(QueryMetadata)
-add_subdirectory(OptionSelectorFilter)
+add_subdirectory(Registry)
+add_subdirectory(Runtime)
+add_subdirectory(ScopeBase)
+add_subdirectory(ScopeExceptions)
 add_subdirectory(Variant)
 add_subdirectory(VariantBuilder)
 add_subdirectory(Version)

=== added directory 'test/gtest/scopes/Invocation'
=== added file 'test/gtest/scopes/Invocation/CMakeLists.txt'
--- test/gtest/scopes/Invocation/CMakeLists.txt	1970-01-01 00:00:00 +0000
+++ test/gtest/scopes/Invocation/CMakeLists.txt	2014-04-28 05:57:56 +0000
@@ -0,0 +1,8 @@
+configure_file(Registry.ini.in ${CMAKE_CURRENT_BINARY_DIR}/Registry.ini)
+configure_file(Runtime.ini.in ${CMAKE_CURRENT_BINARY_DIR}/Runtime.ini)
+configure_file(Zmq.ini.in ${CMAKE_CURRENT_BINARY_DIR}/Zmq.ini)
+
+add_executable(Invocation_test Invocation_test.cpp TestScope.cpp)
+target_link_libraries(Invocation_test ${TESTLIBS})
+
+add_test(Invocation Invocation_test)

=== added file 'test/gtest/scopes/Invocation/Invocation_test.cpp'
--- test/gtest/scopes/Invocation/Invocation_test.cpp	1970-01-01 00:00:00 +0000
+++ test/gtest/scopes/Invocation/Invocation_test.cpp	2014-04-28 05:57:56 +0000
@@ -0,0 +1,145 @@
+/*
+ * Copyright (C) 2014 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: Michi Henning <[e-mail]>
+ */
+
+#include <signal.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include <mutex>
+
+#include <unity/scopes/ActionMetadata.h>
+#include <unity/scopes/CategorisedResult.h>
+#include <unity/scopes/internal/MWScope.h>
+#include <unity/scopes/internal/RuntimeImpl.h>
+#include <unity/scopes/internal/ScopeImpl.h>
+#include <unity/scopes/ListenerBase.h>
+#include <unity/scopes/QueryCtrl.h>
+#include <unity/scopes/Runtime.h>
+#include <unity/scopes/SearchMetadata.h>
+#include <unity/UnityExceptions.h>
+
+#include <gtest/gtest.h>
+
+#include "TestScope.h"
+
+using namespace std;
+using namespace unity::scopes;
+
+class TestReceiver : public SearchListenerBase
+{
+public:
+    TestReceiver()
+        : query_complete_(false)
+    {
+    }
+
+    virtual void push(CategorisedResult /* result */) override
+    {
+    }
+
+    virtual void finished(ListenerBase::Reason reason, string const& error_message) override
+    {
+        lock_guard<mutex> lock(mutex_);
+        reason_ = reason;
+        error_message_ = error_message;
+        query_complete_ = true;
+        cond_.notify_one();
+    }
+
+    void wait_until_finished()
+    {
+        unique_lock<mutex> lock(mutex_);
+        cond_.wait(lock, [this] { return this->query_complete_; });
+        query_complete_ = false;
+    }
+
+    ListenerBase::Reason reason()
+    {
+        lock_guard<mutex> lock(mutex_);
+        return reason_;
+    }
+
+    string error_message()
+    {
+        lock_guard<mutex> lock(mutex_);
+        return error_message_;
+    }
+
+private:
+    bool query_complete_;
+    ListenerBase::Reason reason_;
+    string error_message_;
+    mutex mutex_;
+    condition_variable cond_;
+};
+
+// Check that invoking on a scope after a timeout exception from a previous
+// invocation works correctly. This tests that a failed socket is removed
+// from the connection pool in ZmqObject::invoke_twoway_().
+
+TEST(Invocation, timeout)
+{
+    auto rt = internal::RuntimeImpl::create("", "Runtime.ini");
+    auto mw = rt->factory()->create("TestScope", "Zmq", "Zmq.ini");
+    mw->start();
+    auto proxy = mw->create_scope_proxy("TestScope");
+    auto scope = internal::ScopeImpl::create(proxy, rt.get(), "no_such_scope");
+
+    auto receiver = make_shared<TestReceiver>();
+
+    // First call must time out.
+    scope->search("test", SearchMetadata("unused", "unused"), receiver);
+    receiver->wait_until_finished();
+
+    EXPECT_EQ(ListenerBase::Error, receiver->reason());
+    EXPECT_EQ("unity::scopes::TimeoutException: Request timed out after 300 milliseconds", receiver->error_message());
+
+   // this_thread::sleep_for(chrono::milliseconds(500));
+
+    // Second call must succeed
+    scope->search("test", SearchMetadata("unused", "unused"), receiver);
+    receiver->wait_until_finished();
+
+    EXPECT_EQ(ListenerBase::Finished, receiver->reason());
+    EXPECT_EQ("", receiver->error_message());
+}
+
+void scope_thread(Runtime::SPtr const& rt)
+{
+    TestScope scope;
+    rt->run_scope(&scope, "/foo");
+}
+
+int main(int argc, char **argv)
+{
+    ::testing::InitGoogleTest(&argc, argv);
+
+    Runtime::SPtr srt = move(Runtime::create_scope_runtime("TestScope", "Runtime.ini"));
+    std::thread scope_t(scope_thread, srt);
+
+    // Give thread some time to bind to its endpoint, to avoid getting ObjectNotExistException
+    // from a synchronous remote call.
+    this_thread::sleep_for(chrono::milliseconds(200));
+
+    auto rc = RUN_ALL_TESTS();
+
+    srt->destroy();
+    scope_t.join();
+
+    return rc;
+}

=== added file 'test/gtest/scopes/Invocation/Registry.ini.in'
--- test/gtest/scopes/Invocation/Registry.ini.in	1970-01-01 00:00:00 +0000
+++ test/gtest/scopes/Invocation/Registry.ini.in	2014-04-28 05:57:56 +0000
@@ -0,0 +1,8 @@
+[Registry]
+Middleware = Zmq
+Zmq.Endpoint = ipc:///tmp/socket_for_registry
+Zmq.EndpointDir = /tmp
+Zmq.ConfigFile = Zmq.ini
+Scope.InstallDir = /unused
+Click.InstallDir = /unused
+Scoperunner.Path = /unused

=== added file 'test/gtest/scopes/Invocation/Runtime.ini.in'
--- test/gtest/scopes/Invocation/Runtime.ini.in	1970-01-01 00:00:00 +0000
+++ test/gtest/scopes/Invocation/Runtime.ini.in	2014-04-28 05:57:56 +0000
@@ -0,0 +1,5 @@
+[Runtime]
+Registry.Identity = Registry
+Registry.ConfigFile = Registry.ini
+Default.Middleware = Zmq
+Zmq.ConfigFile = Zmq.ini

=== added file 'test/gtest/scopes/Invocation/TestScope.cpp'
--- test/gtest/scopes/Invocation/TestScope.cpp	1970-01-01 00:00:00 +0000
+++ test/gtest/scopes/Invocation/TestScope.cpp	2014-04-28 05:57:56 +0000
@@ -0,0 +1,94 @@
+/*
+ * Copyright (C) 2014 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: Michi Henning <[e-mail]>
+ */
+
+#include "TestScope.h"
+
+#include <unity/scopes/CategorisedResult.h>
+#include <unity/scopes/ScopeBase.h>
+#include <unity/scopes/SearchReply.h>
+
+#include <mutex>
+#include <thread>
+
+using namespace std;
+using namespace unity::scopes;
+
+namespace
+{
+
+class TestQuery : public SearchQueryBase
+{
+public:
+    TestQuery()
+    {
+    }
+
+    virtual void cancelled() override
+    {
+    }
+
+    virtual void run(SearchReplyProxy const& reply) override
+    {
+        auto cat = reply->register_category("cat1", "Category 1", "");
+        CategorisedResult res(cat);
+        res.set_uri("uri");
+        res.set_title("title");
+        res.set_art("art");
+        res.set_dnd_uri("dnd_uri");
+        reply->push(res);
+    }
+};
+
+}  // namespace
+
+int TestScope::start(string const&, RegistryProxy const &)
+{
+    return VERSION;
+}
+
+void TestScope::stop()
+{
+}
+
+void TestScope::run()
+{
+}
+
+namespace
+{
+
+mutex m;
+int count = 0;
+
+}  // namespace
+
+SearchQueryBase::UPtr TestScope::search(CannedQuery const&, SearchMetadata const &)
+{
+    lock_guard<mutex> lock(m);
+
+    if (count++ == 0)
+    {
+        this_thread::sleep_for(chrono::milliseconds(400));  // Force timeout on first call
+    }
+    return SearchQueryBase::UPtr(new TestQuery());
+}
+
+PreviewQueryBase::UPtr TestScope::preview(Result const&, ActionMetadata const &)
+{
+    return nullptr;  // unused
+}

=== added file 'test/gtest/scopes/Invocation/TestScope.h'
--- test/gtest/scopes/Invocation/TestScope.h	1970-01-01 00:00:00 +0000
+++ test/gtest/scopes/Invocation/TestScope.h	2014-04-28 05:57:56 +0000
@@ -0,0 +1,40 @@
+/*
+ * Copyright (C) 2014 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: Michi Henning <[e-mail]>
+ */
+
+#ifndef TEST_TESTSCOPE_H
+#define TEST_TESTSCOPE_H
+
+#include <unity/scopes/ScopeBase.h>
+
+using namespace std;
+using namespace unity::scopes;
+
+class TestScope : public ScopeBase
+{
+public:
+    virtual int start(string const&, RegistryProxy const &) override;
+
+    virtual void stop() override;
+
+    virtual void run() override;
+
+    virtual SearchQueryBase::UPtr search(CannedQuery const &, SearchMetadata const &) override;
+    virtual PreviewQueryBase::UPtr preview(Result const&, ActionMetadata const &) override;
+};
+
+#endif

=== added file 'test/gtest/scopes/Invocation/Zmq.ini.in'
--- test/gtest/scopes/Invocation/Zmq.ini.in	1970-01-01 00:00:00 +0000
+++ test/gtest/scopes/Invocation/Zmq.ini.in	2014-04-28 05:57:56 +0000
@@ -0,0 +1,3 @@
+[Zmq]
+EndpointDir.Public = /tmp
+EndpointDir.Private = /tmp

=== modified file 'test/gtest/scopes/Registry/Registry_test.cpp'
--- test/gtest/scopes/Registry/Registry_test.cpp	2014-04-08 10:34:19 +0000
+++ test/gtest/scopes/Registry/Registry_test.cpp	2014-04-28 05:57:56 +0000
@@ -24,8 +24,12 @@
 #include <boost/filesystem.hpp>
 #include <boost/system/error_code.hpp>
 #include <boost/filesystem/operations.hpp>
+#include <gtest/gtest.h>
+
+#include <condition_variable>
 #include <functional>
-#include <gtest/gtest.h>
+#include <mutex>
+
 #include <signal.h>
 #include <unistd.h>
 
@@ -34,13 +38,40 @@
 class Receiver : public SearchListenerBase
 {
 public:
+    Receiver()
+        : done_(false)
+        , finished_ok_(false)
+    {
+    }
+
     virtual void push(CategorisedResult /* result */) override
     {
     }
 
-    virtual void finished(ListenerBase::Reason /* reason */, std::string const& /* error_message */) override
-    {
-    }
+    virtual void finished(ListenerBase::Reason reason, std::string const& /* error_message */) override
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+
+        EXPECT_EQ(Finished, reason);
+        finished_ok_ = reason == Finished;
+        done_ = true;
+        cond_.notify_all();
+    }
+
+    bool wait_until_finished()
+    {
+        std::unique_lock<std::mutex> lock(mutex_);
+        auto now = std::chrono::steady_clock::now();
+        auto expiry_time = now + std::chrono::seconds(5);
+        EXPECT_TRUE(cond_.wait_until(lock, expiry_time, [this]{ return done_; })) << "finished message not delivered";
+        return finished_ok_;
+    }
+
+private:
+    bool done_;
+    bool finished_ok_;
+    std::mutex mutex_;
+    std::condition_variable cond_;
 };
 
 TEST(Registry, metadata)
@@ -75,18 +106,13 @@
 
     auto sp = meta.proxy();
 
-    SearchListenerBase::SPtr reply(new Receiver);
+    auto receiver = std::make_shared<Receiver>();
+    SearchListenerBase::SPtr reply(receiver);
     SearchMetadata metadata("C", "desktop");
 
     // search would fail if testscopeB can't be executed
-    try
-    {
-        auto ctrl = sp->search("foo", metadata, reply);
-    }
-    catch (...)
-    {
-        FAIL();
-    }
+    auto ctrl = sp->search("foo", metadata, reply);
+    EXPECT_TRUE(receiver->wait_until_finished());
 }
 
 bool wait_for_registry()

=== modified file 'test/gtest/scopes/Runtime/CMakeLists.txt'
--- test/gtest/scopes/Runtime/CMakeLists.txt	2014-02-19 04:18:26 +0000
+++ test/gtest/scopes/Runtime/CMakeLists.txt	2014-04-28 05:57:56 +0000
@@ -2,7 +2,7 @@
 configure_file(Runtime.ini.in ${CMAKE_CURRENT_BINARY_DIR}/Runtime.ini)
 configure_file(Zmq.ini.in ${CMAKE_CURRENT_BINARY_DIR}/Zmq.ini)
 
-add_executable(Runtime_test Runtime_test.cpp TestScope.cpp PusherScope.cpp)
+add_executable(Runtime_test Runtime_test.cpp TestScope.cpp PusherScope.cpp SlowCreateScope.cpp)
 target_link_libraries(Runtime_test ${TESTLIBS})
 
 add_test(Runtime Runtime_test)

=== modified file 'test/gtest/scopes/Runtime/PusherScope.cpp'
--- test/gtest/scopes/Runtime/PusherScope.cpp	2014-03-07 01:07:28 +0000
+++ test/gtest/scopes/Runtime/PusherScope.cpp	2014-04-28 05:57:56 +0000
@@ -23,11 +23,16 @@
 #include <unity/scopes/ScopeBase.h>
 #include <unity/scopes/SearchReply.h>
 
+#include <atomic>
+
 #include <gtest/gtest.h>
 
 using namespace std;
 using namespace unity::scopes;
 
+namespace
+{
+
 class PusherQuery : public SearchQueryBase
 {
 public:
@@ -59,9 +64,11 @@
     }
 
 private:
-    int cardinality_;
+    atomic_int cardinality_;
 };
 
+}  // namespace
+
 int PusherScope::start(string const&, RegistryProxy const &)
 {
     return VERSION;

=== modified file 'test/gtest/scopes/Runtime/PusherScope.h'
--- test/gtest/scopes/Runtime/PusherScope.h	2014-02-27 23:20:56 +0000
+++ test/gtest/scopes/Runtime/PusherScope.h	2014-04-28 05:57:56 +0000
@@ -16,6 +16,9 @@
  * Authored by: Michi Henning <[e-mail]>
  */
 
+#ifndef TEST_PUSHERSCOPE_H
+#define TEST_PUSHERSCOPE_H
+
 #include <unity/scopes/ScopeBase.h>
 
 using namespace std;
@@ -33,3 +36,5 @@
     virtual SearchQueryBase::UPtr search(CannedQuery const &, SearchMetadata const &) override;
     virtual PreviewQueryBase::UPtr preview(Result const& result, ActionMetadata const& metadata) override;
 };
+
+#endif

=== modified file 'test/gtest/scopes/Runtime/Runtime_test.cpp'
--- test/gtest/scopes/Runtime/Runtime_test.cpp	2014-04-03 14:22:02 +0000
+++ test/gtest/scopes/Runtime/Runtime_test.cpp	2014-04-28 05:57:56 +0000
@@ -1,5 +1,5 @@
 /*
-* Copyright (C) 2013 Canonical Ltd
+ * Copyright (C) 2013 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
@@ -22,20 +22,22 @@
 
 #include <mutex>
 
+#include <unity/scopes/ActionMetadata.h>
 #include <unity/scopes/CategorisedResult.h>
+#include <unity/scopes/internal/MWScope.h>
+#include <unity/scopes/internal/RuntimeImpl.h>
+#include <unity/scopes/internal/ScopeImpl.h>
 #include <unity/scopes/ListenerBase.h>
+#include <unity/scopes/QueryCtrl.h>
 #include <unity/scopes/Runtime.h>
-#include <unity/scopes/internal/RuntimeImpl.h>
-#include <unity/scopes/internal/MWScope.h>
-#include <unity/scopes/internal/ScopeImpl.h>
-#include <unity/scopes/ActionMetadata.h>
 #include <unity/scopes/SearchMetadata.h>
 #include <unity/UnityExceptions.h>
 
 #include <gtest/gtest.h>
 
+#include "PusherScope.h"
+#include "SlowCreateScope.h"
 #include "TestScope.h"
-#include "PusherScope.h"
 
 using namespace std;
 using namespace unity::scopes;
@@ -82,6 +84,7 @@
         count_++;
         last_result_ = std::make_shared<Result>(result);
     }
+
     virtual void push(Annotation annotation) override
     {
         EXPECT_EQ(1u, annotation.links().size());
@@ -92,6 +95,7 @@
         EXPECT_EQ("dep1", query.department_id());
         annotation_count_++;
     }
+
     virtual void finished(ListenerBase::Reason reason, string const& error_message) override
     {
         EXPECT_EQ(Finished, reason);
@@ -104,15 +108,18 @@
         query_complete_ = true;
         cond_.notify_one();
     }
+
     void wait_until_finished()
     {
         unique_lock<mutex> lock(mutex_);
         cond_.wait(lock, [this] { return this->query_complete_; });
     }
+
     std::shared_ptr<Result> last_result()
     {
         return last_result_;
     }
+
 private:
     bool query_complete_;
     mutex mutex_;
@@ -225,6 +232,47 @@
     receiver->wait_until_finished();
 }
 
+TEST(Runtime, consecutive_search)
+{
+    auto rt = internal::RuntimeImpl::create("", "Runtime.ini");
+    auto mw = rt->factory()->create("TestScope", "Zmq", "Zmq.ini");
+    mw->start();
+
+    auto proxy = mw->create_scope_proxy("TestScope");
+    auto scope = internal::ScopeImpl::create(proxy, rt.get(), "TestScope");
+
+    auto receiver = make_shared<Receiver>();
+    auto ctrl = scope->search("test", SearchMetadata("en", "phone"), receiver);
+    receiver->wait_until_finished();
+
+    std::vector<std::shared_ptr<Receiver>> replies;
+
+    const int num_searches = 100;
+    for (int i = 0; i < num_searches; ++i)
+    {
+        replies.push_back(std::make_shared<Receiver>());
+        scope->search("test", SearchMetadata("en", "phone"), replies.back());
+    }
+
+    for (int i = 0; i < num_searches; ++i)
+    {
+        replies[i]->wait_until_finished();
+    }
+
+    // Do it again, to test that re-use of previously-created connections works.
+    replies.clear();
+    for (int i = 0; i < num_searches; ++i)
+    {
+        replies.push_back(std::make_shared<Receiver>());
+        scope->search("test", SearchMetadata("en", "phone"), replies.back());
+    }
+
+    for (int i = 0; i < num_searches; ++i)
+    {
+        replies[i]->wait_until_finished();
+    }
+}
+
 TEST(Runtime, preview)
 {
     // connect to scope and run a query
@@ -272,6 +320,68 @@
     receiver->wait_until_finished();
 }
 
+class CancelReceiver : public SearchListenerBase
+{
+public:
+    CancelReceiver()
+        : query_complete_(false)
+    {
+    }
+
+    virtual void push(CategorisedResult /* result */) override
+    {
+        FAIL();
+    }
+
+    virtual void finished(ListenerBase::Reason reason, string const& error_message) override
+    {
+        EXPECT_EQ(Cancelled, reason);
+        EXPECT_EQ("", error_message);
+        // Signal that the query has completed.
+        unique_lock<mutex> lock(mutex_);
+        query_complete_ = true;
+        cond_.notify_one();
+    }
+
+    void wait_until_finished()
+    {
+        unique_lock<mutex> lock(mutex_);
+        cond_.wait(lock, [this] { return this->query_complete_; });
+    }
+
+private:
+    bool query_complete_;
+    mutex mutex_;
+    condition_variable cond_;
+};
+
+TEST(Runtime, early_cancel)
+{
+    auto rt = internal::RuntimeImpl::create("", "Runtime.ini");
+    auto mw = rt->factory()->create("SlowCreateScope", "Zmq", "Zmq.ini");
+    mw->start();
+    auto proxy = mw->create_scope_proxy("SlowCreateScope");
+    auto scope = internal::ScopeImpl::create(proxy, rt.get(), "SlowCreateScope");
+
+    // Check that, if a cancel is sent before search() returns on the server side, the
+    // cancel is correctly forwarded to the scope once the real reply proxy arrives
+    // over the wire.
+    auto receiver = make_shared<CancelReceiver>();
+    auto ctrl = scope->search("test", SearchMetadata("unused", "unused"), receiver);
+    // Allow some time for the search message to get there.
+    this_thread::sleep_for(chrono::milliseconds(100));
+    // search() in the scope doesn't return for some time, so the cancel() that follows
+    // is sent to the "fake" QueryCtrlProxy.
+    ctrl->cancel();
+    receiver->wait_until_finished();
+    // The receiver receives its cancel from the client-side run time instead of the
+    // scope because the run time short-cuts sending the cancel locally instead
+    // of waiting for the cancel message from the scope. Allow some time for the
+    // cancel to reach the scope before shutting down the run time, so the scope
+    // can test that it received the cancel.
+    this_thread::sleep_for(chrono::milliseconds(200));
+}
+
 void scope_thread(Runtime::SPtr const& rt)
 {
     TestScope scope;
@@ -284,6 +394,12 @@
     rt->run_scope(&scope, "/foo");
 }
 
+void slow_create_thread(Runtime::SPtr const& rt)
+{
+    SlowCreateScope scope;
+    rt->run_scope(&scope, "/foo");
+}
+
 int main(int argc, char **argv)
 {
     ::testing::InitGoogleTest(&argc, argv);
@@ -294,6 +410,13 @@
     Runtime::SPtr prt = move(Runtime::create_scope_runtime("PusherScope", "Runtime.ini"));
     std::thread pusher_t(pusher_thread, prt);
 
+    Runtime::SPtr scrt = move(Runtime::create_scope_runtime("SlowCreateScope", "Runtime.ini"));
+    std::thread slow_create_t(slow_create_thread, scrt);
+
+    // Give threads some time to bind to their endpoints, to avoid getting ObjectNotExistException
+    // from a synchronous remote call.
+    this_thread::sleep_for(chrono::milliseconds(200));
+
     auto rc = RUN_ALL_TESTS();
 
     srt->destroy();
@@ -302,5 +425,8 @@
     prt->destroy();
     pusher_t.join();
 
+    scrt->destroy();
+    slow_create_t.join();
+
     return rc;
 }

=== added file 'test/gtest/scopes/Runtime/SlowCreateScope.cpp'
--- test/gtest/scopes/Runtime/SlowCreateScope.cpp	1970-01-01 00:00:00 +0000
+++ test/gtest/scopes/Runtime/SlowCreateScope.cpp	2014-04-28 05:57:56 +0000
@@ -0,0 +1,101 @@
+/*
+ * Copyright (C) 2013 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: Michi Henning <[e-mail]>
+ */
+
+#include <unity/scopes/ScopeBase.h>
+
+#include "SlowCreateScope.h"
+
+#include <condition_variable>
+#include <mutex>
+#include <thread>
+
+#include <gtest/gtest.h>
+
+using namespace std;
+using namespace unity::scopes;
+
+namespace
+{
+
+class TestQuery : public SearchQueryBase
+{
+public:
+    TestQuery()
+    {
+        lock_guard<mutex> lock(mutex_);
+        cancelled_ = false;
+    }
+
+    virtual void cancelled() override
+    {
+        lock_guard<mutex> lock(mutex_);
+        cancelled_ = true;
+        cond_.notify_all();
+    }
+
+    virtual void run(SearchReplyProxy const&) override
+    {
+        if (!valid())
+        {
+            return; // Query was cancelled already
+        }
+        // We time out the wait because, otherwise, the scope's object adapter can't shut down
+        // and the test hangs forever.
+        // If this fails, the cancelled method wasn't called after five seconds.
+        auto stop_time = chrono::steady_clock::now() + chrono::seconds(5);
+        unique_lock<mutex> lock(mutex_);
+        EXPECT_TRUE(cond_.wait_until(lock, stop_time, [this]{ return cancelled_;}));
+    }
+
+private:
+    mutex mutex_;
+    condition_variable cond_;
+    bool cancelled_;
+};
+
+}  // namespace
+
+int SlowCreateScope::start(string const&, RegistryProxy const &)
+{
+    return VERSION;
+}
+
+void SlowCreateScope::stop()
+{
+}
+
+void SlowCreateScope::run()
+{
+}
+
+SearchQueryBase::UPtr SlowCreateScope::search(CannedQuery const&, SearchMetadata const &)
+{
+    // Sleep for a while. This allows the client to call cancel() before this function
+    // returns, while the test still holds a fake QueryCtrl returned by the async
+    // invocation. When this method returns, the client-side run time calls
+    // the real cancel, which triggers the cancelled() callback above, which, in turn,
+    // causes TestQuery::run() to complete.
+    this_thread::sleep_for(chrono::milliseconds(200));
+
+    return SearchQueryBase::UPtr(new TestQuery());
+}
+
+PreviewQueryBase::UPtr SlowCreateScope::preview(Result const&, ActionMetadata const &)
+{
+    return nullptr;  // Not called
+}

=== added file 'test/gtest/scopes/Runtime/SlowCreateScope.h'
--- test/gtest/scopes/Runtime/SlowCreateScope.h	1970-01-01 00:00:00 +0000
+++ test/gtest/scopes/Runtime/SlowCreateScope.h	2014-04-28 05:57:56 +0000
@@ -0,0 +1,40 @@
+/*
+ * Copyright (C) 2013 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: Michi Henning <[e-mail]>
+ */
+
+#ifndef TEST_SLOWCREATESCOPE_H
+#define TEST_SLOWCREATESCOPE_H
+
+#include <unity/scopes/ScopeBase.h>
+
+using namespace std;
+using namespace unity::scopes;
+
+class SlowCreateScope : public ScopeBase
+{
+public:
+    virtual int start(string const&, RegistryProxy const &) override;
+
+    virtual void stop() override;
+
+    virtual void run() override;
+
+    virtual SearchQueryBase::UPtr search(CannedQuery const &, SearchMetadata const &) override;
+    virtual PreviewQueryBase::UPtr preview(Result const&, ActionMetadata const &) override;
+};
+
+#endif

=== modified file 'test/gtest/scopes/Runtime/TestScope.cpp'
--- test/gtest/scopes/Runtime/TestScope.cpp	2014-03-07 01:07:28 +0000
+++ test/gtest/scopes/Runtime/TestScope.cpp	2014-04-28 05:57:56 +0000
@@ -30,6 +30,9 @@
 using namespace std;
 using namespace unity::scopes;
 
+namespace
+{
+
 class TestQuery : public SearchQueryBase
 {
 public:
@@ -41,6 +44,7 @@
     virtual void cancelled() override
     {
     }
+
     virtual void run(SearchReplyProxy const& reply) override
     {
         Department dep("news", query_, "News");
@@ -65,6 +69,8 @@
     CannedQuery query_;
 };
 
+}  // namespace
+
 class TestPreview : public PreviewQueryBase
 {
 public:

=== modified file 'test/gtest/scopes/Runtime/TestScope.h'
--- test/gtest/scopes/Runtime/TestScope.h	2014-02-27 23:20:56 +0000
+++ test/gtest/scopes/Runtime/TestScope.h	2014-04-28 05:57:56 +0000
@@ -16,6 +16,9 @@
  * Authored by: James Henstridge <[e-mail]>
  */
 
+#ifndef TEST_TESTSCOPE_H
+#define TEST_TESTSCOPE_H
+
 #include <unity/scopes/ScopeBase.h>
 
 using namespace std;
@@ -33,3 +36,5 @@
     virtual SearchQueryBase::UPtr search(CannedQuery const &, SearchMetadata const &) override;
     virtual PreviewQueryBase::UPtr preview(Result const&, ActionMetadata const &) override;
 };
+
+#endif

=== modified file 'test/gtest/scopes/ScopeExceptions/ScopeExceptions_test.cpp'
--- test/gtest/scopes/ScopeExceptions/ScopeExceptions_test.cpp	2014-01-09 09:47:56 +0000
+++ test/gtest/scopes/ScopeExceptions/ScopeExceptions_test.cpp	2014-04-28 05:57:56 +0000
@@ -60,3 +60,32 @@
         EXPECT_EQ(e.name(), e2.name());
     }
 }
+
+TEST(ObjectNotExistException, state)
+{
+    {
+        ObjectNotExistException e("some error", "some id");
+        EXPECT_STREQ("unity::scopes::ObjectNotExistException: some error (id = some id)", e.what());
+        EXPECT_EQ("unity::scopes::ObjectNotExistException", e.name());
+        EXPECT_EQ("some id", e.id());
+        EXPECT_THROW(rethrow_exception(e.self()), ObjectNotExistException);
+        ObjectNotExistException e2("blah", "some id");
+        e2 = e;
+        EXPECT_EQ(e.reason(), e2.reason());
+        EXPECT_EQ(e.name(), e2.name());
+    }
+}
+
+TEST(TimeoutException, state)
+{
+    {
+        TimeoutException e("some error");
+        EXPECT_STREQ("unity::scopes::TimeoutException: some error", e.what());
+        EXPECT_EQ("unity::scopes::TimeoutException", e.name());
+        EXPECT_THROW(rethrow_exception(e.self()), TimeoutException);
+        TimeoutException e2("blah");
+        e2 = e;
+        EXPECT_EQ(e.reason(), e2.reason());
+        EXPECT_EQ(e.name(), e2.name());
+    }
+}

=== modified file 'test/gtest/scopes/internal/ThreadPool/ThreadPool_test.cpp'
--- test/gtest/scopes/internal/ThreadPool/ThreadPool_test.cpp	2014-01-24 03:18:25 +0000
+++ test/gtest/scopes/internal/ThreadPool/ThreadPool_test.cpp	2014-04-28 05:57:56 +0000
@@ -104,3 +104,14 @@
         EXPECT_STREQ("unity::InvalidArgumentException: ThreadPool(): invalid pool size: -1", e.what());
     }
 }
+
+TEST(ThreadPool, throwing_task)
+{
+    ThreadPool p(1);
+    auto thrower = [](){ throw std::logic_error("some error"); };
+    auto thrower2 = [](){ throw 99; };
+    auto f = p.submit(thrower);
+    auto f2 = p.submit(thrower2);
+    EXPECT_THROW(f.get(), std::logic_error);
+    EXPECT_THROW(f2.get(), int);
+}

=== modified file 'test/gtest/scopes/internal/ThreadSafeQueue/ThreadSafeQueue_test.cpp'
--- test/gtest/scopes/internal/ThreadSafeQueue/ThreadSafeQueue_test.cpp	2014-02-25 12:08:41 +0000
+++ test/gtest/scopes/internal/ThreadSafeQueue/ThreadSafeQueue_test.cpp	2014-04-28 05:57:56 +0000
@@ -74,14 +74,26 @@
 
 TEST(ThreadSafeQueue, exception)
 {
-    unique_ptr<ThreadSafeQueue<string>> q(new ThreadSafeQueue<string>);
-    q->push("fred");
-    auto f = waiter_ready.get_future();
-    auto t = thread(waiter_thread, q.get());
-    f.wait();
-    this_thread::sleep_for(chrono::milliseconds(50));   // Make sure child thread has time to call wait_and_pop()
-    q.reset();
-    t.join();
+    {
+        unique_ptr<ThreadSafeQueue<string>> q(new ThreadSafeQueue<string>);
+        q->push("fred");
+        auto f = waiter_ready.get_future();
+        auto t = thread(waiter_thread, q.get());
+        f.wait();
+        this_thread::sleep_for(chrono::milliseconds(50));   // Make sure child thread has time to call wait_and_pop()
+        q->destroy();
+        t.join();
+
+        try
+        {
+            q->push("fred");
+            FAIL();
+        }
+        catch (std::runtime_error const& e)
+        {
+            EXPECT_STREQ("ThreadSafeQueue: cannot push onto destroyed queue", e.what());
+        }
+    }
 }
 
 atomic_int count;
@@ -112,6 +124,7 @@
 
     // Destroy the queue while multiple threads are sleeping in wait_and_pop().
     q.destroy();
+    q.wait_for_destroy();
 
     for (auto& t : threads)
     {
@@ -120,6 +133,45 @@
     EXPECT_EQ(20, count);
 }
 
+void destroy_thread(ThreadSafeQueue<int>* q)
+{
+    this_thread::sleep_for(chrono::milliseconds(100));
+    EXPECT_EQ(0u, q->size());
+    q->destroy();
+}
+
+TEST(ThreadSafeQueue, destroy_while_empty)
+{
+    ThreadSafeQueue<int> q;
+    thread t(destroy_thread, &q);
+    q.wait_for_destroy();
+    EXPECT_EQ(0u, q.size());
+    t.join();
+}
+
+void wait_for_destroy_thread(ThreadSafeQueue<int>* q)
+{
+    q->wait_for_destroy();
+}
+
+TEST(ThreadSafeQueue, destroy_while_waiting_for_destroy)
+{
+    ThreadSafeQueue<int> q;
+    q.push(42);
+    thread t(wait_for_destroy_thread, &q);
+    this_thread::sleep_for(chrono::milliseconds(100));
+    q.destroy();
+    t.join();
+
+    // Call destroy() and wait_for_destroy() again, to make sure they do nothing.
+    q.destroy();
+    q.wait_for_destroy();
+    q.destroy();
+    q.destroy();
+    q.wait_for_destroy();
+    q.wait_for_destroy();
+}
+
 class MoveOnly
 {
 public:
@@ -156,4 +208,15 @@
     EXPECT_EQ("world", m.val());
     m = q.wait_and_pop();
     EXPECT_EQ("again", m.val());
+
+    q.destroy();
+    try
+    {
+        q.push(move(MoveOnly("no_push")));
+        FAIL();
+    }
+    catch (std::runtime_error const& e)
+    {
+        EXPECT_STREQ("ThreadSafeQueue: cannot push onto destroyed queue", e.what());
+    }
 }

=== modified file 'test/gtest/scopes/internal/smartscopes/smartscopesproxy/smartscopesproxy_test.cpp'
--- test/gtest/scopes/internal/smartscopes/smartscopesproxy/smartscopesproxy_test.cpp	2014-04-01 15:26:50 +0000
+++ test/gtest/scopes/internal/smartscopes/smartscopesproxy/smartscopesproxy_test.cpp	2014-04-28 05:57:56 +0000
@@ -77,11 +77,8 @@
 
     ~smartscopesproxytest()
     {
-        scope_mw_->stop();
-        scope_mw_->wait_for_shutdown();
-
-        reg_mw_->stop();
-        reg_mw_->wait_for_shutdown();
+        scope_rt_->destroy();
+        reg_rt_->destroy();
     }
 
 protected:

=== modified file 'test/gtest/scopes/internal/zmq_middleware/CMakeLists.txt'
--- test/gtest/scopes/internal/zmq_middleware/CMakeLists.txt	2014-01-28 06:34:03 +0000
+++ test/gtest/scopes/internal/zmq_middleware/CMakeLists.txt	2014-04-28 05:57:56 +0000
@@ -1,6 +1,7 @@
 add_subdirectory(ObjectAdapter)
 add_subdirectory(RegistryI)
 add_subdirectory(ServantBase)
+add_subdirectory(StopPublisher)
 add_subdirectory(Util)
 add_subdirectory(VariantConverter)
 add_subdirectory(ZmqMiddleware)

=== modified file 'test/gtest/scopes/internal/zmq_middleware/ObjectAdapter/ObjectAdapter_test.cpp'
--- test/gtest/scopes/internal/zmq_middleware/ObjectAdapter/ObjectAdapter_test.cpp	2014-01-24 11:16:14 +0000
+++ test/gtest/scopes/internal/zmq_middleware/ObjectAdapter/ObjectAdapter_test.cpp	2014-04-28 05:57:56 +0000
@@ -44,7 +44,7 @@
 // an occastional "address in use" exception because the new socket
 // tries to bind while the old one is still in the process of destroying itself.
 
-void wait(int millisec = 20)
+void wait(int millisec = 200)
 {
     this_thread::sleep_for(chrono::milliseconds(millisec));
 }
@@ -176,7 +176,7 @@
         }
         catch (MiddlewareException const& e)
         {
-            EXPECT_STREQ("unity::scopes::MiddlewareException: Object adapter in Destroyed "
+            EXPECT_STREQ("unity::scopes::MiddlewareException: activate(): Object adapter in Destroyed "
                          "state (adapter: testscope)",
                          e.what());
         }
@@ -194,25 +194,22 @@
         }
         catch (MiddlewareException const& e)
         {
-            EXPECT_STREQ("unity::scopes::MiddlewareException: ObjectAdapter::run_workers(): broker thread "
+            EXPECT_STREQ("unity::scopes::MiddlewareException: ObjectAdapter::run_workers(): stop thread "
                          "failure (adapter: testscope):\n"
-                         "    unity::scopes::MiddlewareException: ObjectAdapter: broker thread failure "
-                         "(adapter: testscope):\n"
+                         "    unity::scopes::MiddlewareException: StopPublisher(): publisher thread "
+                         "failed (endpoint: inproc://testscope-stopper):\n"
                          "        Address already in use",
                          e.what());
         }
+
         try
         {
             b.shutdown();
         }
         catch (MiddlewareException const& e)
         {
-            EXPECT_STREQ("unity::scopes::MiddlewareException: Object adapter in Failed state (adapter: testscope)\n"
-                         "    Exception history:\n"
-                         "        Exception #1:\n"
-                         "            unity::scopes::MiddlewareException: ObjectAdapter: broker thread failure "
-                         "(adapter: testscope):\n"
-                         "                Address already in use",
+            EXPECT_STREQ("unity::scopes::MiddlewareException: shutdown() [state_ == Failed]: "
+                         "Object adapter in Failed state (adapter: testscope)",
                          e.what());
         }
         try
@@ -221,13 +218,9 @@
         }
         catch (MiddlewareException const& e)
         {
-            EXPECT_STREQ("unity::scopes::MiddlewareException: Object adapter in Failed state (adapter: testscope)\n"
-                          "    Exception history:\n"
-                          "        Exception #1:\n"
-                          "            unity::scopes::MiddlewareException: ObjectAdapter: broker thread failure "
-                          "(adapter: testscope):\n"
-                          "                Address already in use",
-                          e.what());
+            EXPECT_STREQ("unity::scopes::MiddlewareException: wait_for_shutdown(): "
+                         "Object adapter in Failed state (adapter: testscope)",
+                         e.what());
         }
     }
 }
@@ -907,7 +900,8 @@
             }
             catch (MiddlewareException const& e)
             {
-                EXPECT_STREQ("unity::scopes::MiddlewareException: Object adapter in Destroyed state (adapter: testscope)",
+                EXPECT_STREQ("unity::scopes::MiddlewareException: remove(): "
+                             "Object adapter in Destroyed state (adapter: testscope)",
                              e.what());
             }
         };
@@ -1005,7 +999,7 @@
         }
         catch (MiddlewareException const& e)
         {
-            EXPECT_STREQ("unity::scopes::MiddlewareException: Object adapter in Destroyed "
+            EXPECT_STREQ("unity::scopes::MiddlewareException: add(): Object adapter in Destroyed "
                          "state (adapter: testscope)",
                          e.what());
         }
@@ -1016,7 +1010,7 @@
         }
         catch (MiddlewareException const& e)
         {
-            EXPECT_STREQ("unity::scopes::MiddlewareException: Object adapter in Destroyed "
+            EXPECT_STREQ("unity::scopes::MiddlewareException: find(): Object adapter in Destroyed "
                          "state (adapter: testscope)",
                          e.what());
         }
@@ -1212,7 +1206,8 @@
     }
     catch (MiddlewareException const& e)
     {
-        EXPECT_STREQ("unity::scopes::MiddlewareException: Object adapter in Failed state (adapter: testscope2)",
+        EXPECT_STREQ("unity::scopes::MiddlewareException: add_dflt_servant(): "
+                     "Object adapter in Failed state (adapter: testscope2)",
                      e.what());
     }
 
@@ -1224,7 +1219,8 @@
     }
     catch (MiddlewareException const& e)
     {
-        EXPECT_STREQ("unity::scopes::MiddlewareException: Object adapter in Failed state (adapter: testscope2)",
+        EXPECT_STREQ("unity::scopes::MiddlewareException: remove_dflt_servant(): "
+                     "Object adapter in Failed state (adapter: testscope2)",
                      e.what());
     }
 
@@ -1236,7 +1232,8 @@
     }
     catch (MiddlewareException const& e)
     {
-        EXPECT_STREQ("unity::scopes::MiddlewareException: Object adapter in Failed state (adapter: testscope2)",
+        EXPECT_STREQ("unity::scopes::MiddlewareException: find_dflt_servant(): "
+                     "Object adapter in Failed state (adapter: testscope2)",
                      e.what());
     }
 }

=== modified file 'test/gtest/scopes/internal/zmq_middleware/RegistryI/RegistryI_test.cpp'
--- test/gtest/scopes/internal/zmq_middleware/RegistryI/RegistryI_test.cpp	2014-04-14 09:31:25 +0000
+++ test/gtest/scopes/internal/zmq_middleware/RegistryI/RegistryI_test.cpp	2014-04-28 05:57:56 +0000
@@ -404,7 +404,7 @@
         ///! TODO: HACK:
         /// we have to start scope-C and scope-D before starting scope-B here, as B aggregates C and D.
         /// (When re-binding logic is introduced, this will be unnecessary)
-        scope_ids = {"scope-A", "scope-C", "scope-D", "scope-B", "scope-N", "scope-S"};
+        scope_ids = { {"scope-A", "scope-C", "scope-D", "scope-B", "scope-N", "scope-S"} };
         for (auto& scope_id : scope_ids)
         {
             proxies[scope_id] = ScopeImpl::create(mw->create_scope_proxy(scope_id), mw->runtime(), scope_id);
@@ -509,7 +509,7 @@
 // test locating all scopes
 TEST_F(RegistryTest, locate_all)
 {
-    // locate all scopes (hense starting all scope processes)
+    // locate all scopes (hence starting all scope processes)
     for (auto const& scope_id : scope_ids)
     {
         EXPECT_EQ(proxies[scope_id], reg->locate(scope_id));

=== added directory 'test/gtest/scopes/internal/zmq_middleware/StopPublisher'
=== added file 'test/gtest/scopes/internal/zmq_middleware/StopPublisher/CMakeLists.txt'
--- test/gtest/scopes/internal/zmq_middleware/StopPublisher/CMakeLists.txt	1970-01-01 00:00:00 +0000
+++ test/gtest/scopes/internal/zmq_middleware/StopPublisher/CMakeLists.txt	2014-04-28 05:57:56 +0000
@@ -0,0 +1,4 @@
+add_executable(StopPublisher_test StopPublisher_test.cpp)
+target_link_libraries(StopPublisher_test ${LIBS} ${TESTLIBS})
+
+add_test(StopPublisher StopPublisher_test)

=== added file 'test/gtest/scopes/internal/zmq_middleware/StopPublisher/StopPublisher_test.cpp'
--- test/gtest/scopes/internal/zmq_middleware/StopPublisher/StopPublisher_test.cpp	1970-01-01 00:00:00 +0000
+++ test/gtest/scopes/internal/zmq_middleware/StopPublisher/StopPublisher_test.cpp	2014-04-28 05:57:56 +0000
@@ -0,0 +1,136 @@
+/*
+ * Copyright (C) 2014 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/lzmqnses/>.
+ *
+ * Authored by: Michi Henning <[e-mail]>
+ */
+
+#include <unity/scopes/internal/zmq_middleware/StopPublisher.h>
+#include <unity/scopes/ScopeExceptions.h>
+#include <zmqpp/poller.hpp>
+
+#include <gtest/gtest.h>
+
+using namespace std;
+using namespace unity::scopes;
+using namespace unity::scopes::internal::zmq_middleware;
+
+TEST(StopPublisher, basic)
+{
+    {
+        zmqpp::context c;
+        StopPublisher p(&c, "x");
+        EXPECT_EQ("inproc://x", p.endpoint());
+    }
+
+    {
+        zmqpp::context c;
+        StopPublisher p(&c, "x");
+        p.stop();                       // no-op if stop() is called several times.
+        p.stop();
+        EXPECT_EQ("inproc://x", p.endpoint());
+    }
+}
+
+TEST(StopPublisher, exceptions)
+{
+    {
+        zmqpp::context c;
+        StopPublisher p(&c, "x");
+        p.stop();
+        try
+        {
+            p.subscribe();
+            FAIL();
+        }
+        catch (MiddlewareException const& e)
+        {
+            EXPECT_STREQ("unity::scopes::MiddlewareException: StopPublisher::subscribe(): "
+                         "cannot subscribe to stopped publisher (endpoint: inproc://x)",
+                         e.what());
+        }
+    }
+
+    {
+        zmqpp::context c;
+        StopPublisher p(&c, "x");
+        try
+        {
+            StopPublisher q(&c, "x");   // Second publisher at the same endpoint must fail
+            FAIL();
+        }
+        catch (MiddlewareException const& e)
+        {
+            EXPECT_STREQ("unity::scopes::MiddlewareException: StopPublisher(): publisher thread "
+                         "failed (endpoint: inproc://x):\n"
+                         "    Address already in use",
+                         e.what());
+        }
+    }
+}
+
+TEST(StopPublisher, subscribe)
+{
+    {
+        zmqpp::context c;
+        StopPublisher p(&c, "x");
+        auto socket = p.subscribe();
+        zmqpp::poller poller;
+        poller.add(socket);
+        poller.poll(100);
+        EXPECT_FALSE(poller.has_input(socket));
+        p.stop();
+        poller.poll(100);
+        EXPECT_TRUE(poller.has_input(socket));
+    }
+}
+
+void subscriber_thread(StopPublisher* p)
+{
+    auto socket = p->subscribe();
+    zmqpp::poller poller;
+    poller.add(socket);
+    poller.poll();          // Blocks indefinitely until socket is ready
+}
+
+void publisher_thread(unique_ptr<StopPublisher>)
+{
+    // Sleep for a while, so all the subscribers get a chance to block in poll().
+    this_thread::sleep_for(chrono::milliseconds(500));
+    // Destructor of unique_ptr in-param calls stop()
+}
+
+// Test that subscriber threads terminate correctly even if the destructor of the
+// publisher is called from a different thread.
+
+TEST(StopPublisher, threading)
+{
+    zmqpp::context c;
+    unique_ptr<StopPublisher> p(new StopPublisher(&c, "x"));
+    const int num_subscribers = 5;
+    vector<thread> subscribers;
+    for (int i = 0; i < num_subscribers; ++i)
+    {
+        subscribers.emplace_back(thread(subscriber_thread, p.get()));
+    }
+
+    thread publisher(publisher_thread, move(p));  // Moves publisher to publisher thread
+
+    for (auto& s : subscribers)
+    {
+        s.join();
+    }
+
+    publisher.join();
+}