title: dask, execute non-serializable object on every worker
body: <p>I am trying to execute the following graph:</p>
  
  <p><a href="https://i.stack.imgur.com/0hzL8.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/0hzL8.png" alt="graph"></a></p>
  
  <p>which is generated by the following code:</p>
  
  <pre><code>energies = [10, 20]
  system = delayed(make_non_serializable_oject)(x=1)
  trans = [delayed(some_function_that_uses_system)(system, energy) for energy in energies]
  result = delayed(list)(trans)
  result.visualize()
  </code></pre>
  
  <p>When I call <code>result.compute()</code> the calculation never finishes.</p>
  
  <p>Calling <code>result.compute(get=dask.async.get_sync)</code>, and <code>result.compute(dask.threaded.get)</code> both do work. However <code>result.compute(dask.multiprocessing.get)</code> does not and generates the following error:</p>
  
  <pre><code>---------------------------------------------------------------------------
  RemoteError                               Traceback (most recent call last)
  &lt;ipython-input-70-b5c8f2a1c6f6&gt; in &lt;module&gt;()
  ----&gt; 1 result.compute(get=dask.multiprocessing.get)
  
  /home/bnijholt/anaconda3/lib/python3.5/site-packages/dask/base.py in compute(self, **kwargs)
       76             Extra keywords to forward to the scheduler ``get`` function.
       77         """
  ---&gt; 78         return compute(self, **kwargs)[0]
       79 
       80     @classmethod
  
  /home/bnijholt/anaconda3/lib/python3.5/site-packages/dask/base.py in compute(*args, **kwargs)
      169         dsk = merge(var.dask for var in variables)
      170     keys = [var._keys() for var in variables]
  --&gt; 171     results = get(dsk, keys, **kwargs)
      172 
      173     results_iter = iter(results)
  
  /home/bnijholt/anaconda3/lib/python3.5/site-packages/dask/multiprocessing.py in get(dsk, keys, num_workers, func_loads, func_dumps, optimize_graph, **kwargs)
       81         # Run
       82         result = get_async(apply_async, len(pool._pool), dsk3, keys,
  ---&gt; 83                            queue=queue, get_id=_process_get_id, **kwargs)
       84     finally:
       85         if cleanup:
  
  /home/bnijholt/anaconda3/lib/python3.5/site-packages/dask/async.py in get_async(apply_async, num_workers, dsk, result, cache, queue, get_id, raise_on_exception, rerun_exceptions_locally, callbacks, **kwargs)
      479                     _execute_task(task, data)  # Re-execute locally
      480                 else:
  --&gt; 481                     raise(remote_exception(res, tb))
      482             state['cache'][key] = res
      483             finish_task(dsk, key, state, results, keyorder.get)
  
  RemoteError: 
  ---------------------------------------------------------------------------
  Traceback (most recent call last):
    File "/home/bnijholt/anaconda3/lib/python3.5/multiprocessing/managers.py", line 228, in serve_client
      request = recv()
    File "/home/bnijholt/anaconda3/lib/python3.5/multiprocessing/connection.py", line 251, in recv
      return ForkingPickler.loads(buf.getbuffer())
    File "kwant/graph/core.pyx", line 664, in kwant.graph.core.CGraph_malloc.__cinit__ (kwant/graph/core.c:8330)
  TypeError: __cinit__() takes exactly 6 positional arguments (0 given)
  ---------------------------------------------------------------------------
  
  Traceback
  ---------
    File "/home/bnijholt/anaconda3/lib/python3.5/site-packages/dask/async.py", line 273, in execute_task
      queue.put(result)
    File "&lt;string&gt;", line 2, in put
    File "/home/bnijholt/anaconda3/lib/python3.5/multiprocessing/managers.py", line 732, in _callmethod
      raise convert_to_error(kind, result)
  </code></pre>
  
  <p>With <code>ipyparallel</code> I would execute the <code>make_non_serializable_oject</code> on each engine, which solves the problem for that case. </p>
  
  <p>I would like to use <code>dask</code> for my parallel computations, how can I solve this?</p>