Serialization should be done on a separate thread

Consider the following sequence of events:

T+1: Worker 1 finished a task (Task 1), and notifies the scheduler about it
T+2: The scheduler replies with a new task for Worker 1 (which, let's say, only depends on local data to keep this example simple)
T+3: Worker 1 starts working on the new task (Task 2)
T+4: Worker 2 which was at work all this while finishes and notified the scheduler
T+5: The scheduler replies with a new task (Task 3) for Worker 2. This task depends on the output of Task 1 which is still with Worker 1.
T+6: Worker 2 tries to fetch the data required to start working on Task 3, but blocks because Worker 2 is still busy with Task 2 (typically in a compute intensive for loop) and cannot serve Worker 2 the data immediately.
....worker 2 waits....
....and waits...
T+7: Worker 1 finishes Task 2 (and notifies the scheduler).
...worker 2 is still waiting...
T+8: Worker 1 reads Worker 2's request and responds to it with the requested data.
T+9: Finally the wait is over! Worker 2 can start on Task 3.
T+10: Worker 1 reads Task 4 from the scheduler, and Task 4 better not depend on Task 2, because then Worker 1 would have to wait for Worker 2 to finish Task 3........... God help these vengeful quibbling workers.

It works this way because:
1. We don't want the scheduler process doing serialize/deserialize while it should really be scheduling things as fast as possible.
2. It would double the number of times we do serialize/deserialize

The workers really shouldn't be waiting on each other. This happens all the time in matrix multiplication, and sometimes when it doesn't, it's much faster! There are a few solutions to this:

1. Have half the workers (or maybe just 2 workers) just reading and relaying intermediate data. So every worker sends its output to these processes after its done and reads from these processes when it needs some other worker's data before starting a new task.
2. Create a special kind of task that says "Wait for _someone_ to take the data that you have computed for Task 1 and then start working on this Task 2"
3. Do a hybrid of 1 and 2. i.e. have a deadline for the wait in 2 and fall back to 1..

This problem arises because we just use the shared memory scheduler from Dask (which doesn't have this problem, because well you don't need to communicate in a shared-memory set up). I believe the multiprocessing scheduler in Dask was just sending results to the scheduler at the time. So I figured this might not be great for embarassingly parallel workloads.

A great way to solve this issue is to hook into [distributed scheduler](https://distributed.readthedocs.io/en/latest/) and implement its client interface. This, I beleive will not only solve this issue, but make sure other things (like data locality in HDFS) work great in a cluster set-up. (@mrocklin had got me bootstrapped with this process. here are the simple `send_msg` and `recv_msg` functions https://gist.github.com/shashi/e8f37c5f61bab4219555cd3c4fef1dc4) see https://github.com/dask/distributed/issues/586 for a discussion on this.