title: streamz exception on dask gather
body: <p>i am trying to use streamz to manage an image processing pipeline.
  I am simulating a camera using the class Camera and have set up the pipeline below. </p>
  
  <pre><code>import time
  from skimage.io import imread
  from skimage.color import rgb2gray
  
  
  class Camera(object):
      imgs = [imread(f + '.jpg') for f in ['1', '2', '3']]
  
      @staticmethod
      def grab_one():
          return Camera.imgs[int(time.time()) % 3]
  
  def mc_streamz():
      from streamz import Stream
      def np2jpeg(np_im):
          from six import BytesIO
          from PIL import Image
          pim = Image.fromarray(np_im)
          io = BytesIO()
          pim.save(io,format='jpeg')
          io.seek(0)
          return io.read()
  
      def focusmetric(np_im):
          from skimage.filters import laplace
          return laplace(np_im).var()
  
      def asjpegframe(b_im):
          return b'--frame\r\n'b'Content-Type: image/jpeg\r\n\r\n' + b_im + b'\r\n'
  
      from dask.distributed import Client
      client = Client()
  
      source = Stream()
      source.map(np2jpeg).map(asjpegframe).sink(print)
      source.scatter().map(rgb2gray).map(focusmetric).gather().sink(print)
  
      cam = Camera()
  
      for i in range(10):
          source.emit(cam.grab_one())
  
  if __name__ == '__main__':
      mc_streamz()
  </code></pre>
  
  <p>However, when i run this i get the exception below when the gather() function is invoked and I cannot understand why. I hope someone can guide me on the right path, thanks</p>
  
  <pre><code>Traceback (most recent call last):
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/distributed/client.py", line 1438, in _gather
      exception = st.exception
  AttributeError: exception
  
  During handling of the above exception, another exception occurred:
  
  Traceback (most recent call last):
    File "/Users/johndoe/projects/streamz/streamz/core.py", line 1255, in f
      result[0] = yield make_coro()
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1133, in run
      value = future.result()
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1141, in run
      yielded = self.gen.throw(*exc_info)
    File "/Users/johndoe/projects/streamz/streamz/core.py", line 322, in _
      result = yield self._emit(x)
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1133, in run
      value = future.result()
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 883, in callback
      result_list.append(f.result())
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1141, in run
      yielded = self.gen.throw(*exc_info)
    File "/Users/johndoe/projects/streamz/streamz/dask.py", line 98, in update
      f = yield self._emit(future)
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1133, in run
      value = future.result()
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 883, in callback
      result_list.append(f.result())
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1141, in run
      yielded = self.gen.throw(*exc_info)
    File "/Users/johndoe/projects/streamz/streamz/dask.py", line 123, in update
      result = yield client.gather(x, asynchronous=True)
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1133, in run
      value = future.result()
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1141, in run
      yielded = self.gen.throw(*exc_info)
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/distributed/client.py", line 1443, in _gather
      None)
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/six.py", line 693, in reraise
      raise value
  concurrent.futures._base.CancelledError: focusmetric-42e56666b87e375f3ece9246419ad199
  Traceback (most recent call last):
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/distributed/client.py", line 1438, in _gather
      exception = st.exception
  AttributeError: exception
  
  During handling of the above exception, another exception occurred:
  
  Traceback (most recent call last):
    File "streamprocessing.py", line 83, in &lt;module&gt;
      mc_streamz()
    File "streamprocessing.py", line 51, in mc_streamz
      source.emit(cam.grab_one())
    File "/Users/johndoe/projects/streamz/streamz/core.py", line 327, in emit
      sync(self.loop, _)
    File "/Users/johndoe/projects/streamz/streamz/core.py", line 1267, in sync
      six.reraise(*error[0])
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/six.py", line 693, in reraise
      raise value
    File "/Users/johndoe/projects/streamz/streamz/core.py", line 1255, in f
      result[0] = yield make_coro()
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1133, in run
      value = future.result()
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1141, in run
      yielded = self.gen.throw(*exc_info)
    File "/Users/johndoe/projects/streamz/streamz/core.py", line 322, in _
      result = yield self._emit(x)
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1133, in run
      value = future.result()
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 883, in callback
      result_list.append(f.result())
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1141, in run
      yielded = self.gen.throw(*exc_info)
    File "/Users/johndoe/projects/streamz/streamz/dask.py", line 98, in update
      f = yield self._emit(future)
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1133, in run
      value = future.result()
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 883, in callback
      result_list.append(f.result())
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1141, in run
      yielded = self.gen.throw(*exc_info)
    File "/Users/johndoe/projects/streamz/streamz/dask.py", line 123, in update
      result = yield client.gather(x, asynchronous=True)
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1133, in run
      value = future.result()
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/tornado/gen.py", line 1141, in run
      yielded = self.gen.throw(*exc_info)
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/distributed/client.py", line 1443, in _gather
      None)
    File "/Users/johndoe/anaconda3431/lib/python3.6/site-packages/six.py", line 693, in reraise
      raise value
  concurrent.futures._base.CancelledError: focusmetric-42e56666b87e375f3ece9246419ad199
  </code></pre>