Supercedes #432

There are two ways in which this can deadlock.

1. All cores in the cluster are waiting on these tasks
2. All threads in an individual worker are waiting on these tasks while additional tasks are already in the thread-pool-executor's work queue.

The first case is unfortunate, but somewhat controllable and unlikely.

The second case can happen more easily and is harder to control for. Because we're not using coroutines/generator functions we don't have an easy way to jump out of the task and free up the thread. I think our only option here is to add another thread to the ThreadPoolExecutor. This can be done dynamically by changing private variables however the standard concurrent.futures.ThreadPoolExecutor can't scale back down, so adding more space like this is a somewhat irreversible operation. That is unless of course we want to reimplement the ThreadPoolExecutor, which is fairly doable.

I've resolved the more serious of the deadlocks by allowing tasks that call get_executor() to leave the common thread pool, making space for additional work. This required forking the concurrent.futures.ThreadPoolExecutor.

OK, now scatter dumps data directly to the local worker's data dict, bypassing the scheduler.

Additionally I've removed the first deadlock by temporarily increasing the number of slots on any worker that is running a client.

I'm now pretty happy with this. Merging on passed tests