Repository: dask/dask
Pull request: Update setup/python docs for async/await API
Fixes #2895

cc @jcrist for review (I think that this is in line with what you asked for a while ago)

Problem statement:
Slow performance adding to a large Dask-graph
I was replacing a delayed object (`A` in the example), with a Dask array of 1000 chunks (`B` in the example). This increased the bottom of the Dask-graph with a factor of 1000.
The result was that the time for constructing the graph became unpleasantly long.
Testing some more, the speed of adding elements to the Dask-graph seems to be heavily dependent on the prior size of the graph (See the result below). (Line-)Profiling the code, didn't gave me one particular function to blame.
I would have expected it to be more depended on the amount of nodes added, and less on the prior size of the Dask-graph.

Is there something I can change to speed this up?
Is it possible to construct parts of a Dask-graph separately, and connecting them afterwards? Thereby applying the checks and optimizations on the Dask-graph only once, and not 1000 times.

Note: Computing the graph with a large N_prior in this example, leads to a recursion error.
```python
# -*- coding: utf-8 -*-
from dask import delayed, array
import numpy as np
import sys
import dask
print('Dask:',dask.__version__)
print('Numpy:',np.__version__)
print('Python:', sys.version)

from time import clock as time

N_Signal = 10000


d0 = None
for N_Prior in [1000,2000,10000]:
  print('N_Prior',N_Prior)


  ### Existing graph1
  R1 = np.zeros((1000,1))
  for i in range(N_Prior):
    R1 = delayed(np.sum)(R1)

  ### Existing graph2
  R2 = np.zeros((1000,1))
  for i in range(N_Prior):
    R2 = delayed(np.sum)(R2)
    
  @delayed
  def A(a):
    Result = np.zeros((1000,N_Signal))
    for i in range(1000):
      Result[i,:] = np.zeros((N_Signal,)) + i
    return Result
  
  def B(a):
    @delayed
    def Bsub(a,i):
      return np.zeros((N_Signal,)) + i
    Results = []
    for i in range(1000):
      Results.append(array.from_delayed(Bsub(a,i),shape=(N_Signal,),dtype=float))
    return array.stack(Results)
  
  t = time()
  Ra = A(R1)
  matrixA = array.from_delayed(Ra, shape=(1000,N_Signal), dtype=float)
  d1 = time()-t;print('A:',d1);t=time()
  
  Rb = B(R2)
  matrixB = Rb
  d2 = time()-t;print('B:',d2);t=time()
  if not d0:
    d0 = d2
  
  print('Ratio between A & B', round(d2/d1))
  print('Ratio between N_Prior 1000', round(d2/d0,2))
  print()
```

```
Dask: 0.15.3
Numpy: 1.13.3
Python: 3.6.3 |Anaconda custom (64-bit)| (default, Oct 15 2017, 03:27:45) [MSC v.1900 64 bit (AMD64)]
N_Prior 1000
A: 0.0003334181934633307
B: 0.37659461767773905
Ratio between A & B 1129
Ratio between N_Prior 1000 1.0

N_Prior 2000
A: 0.0007378725291573573
B: 0.5679343675319615
Ratio between A & B 770
Ratio between N_Prior 1000 1.51

N_Prior 10000
A: 0.0020814000279187894
B: 2.278880913861485
Ratio between A & B 1095
Ratio between N_Prior 1000 6.05
```

Hints from issue discussion:
Have you tried with `cytoolz` installed?
`cytoolz` version `0.8.2` was already installed.
Could you please include all the dependencies from the environment used? If using `conda`, can just run `conda env export`. Alternatively `pip freeze` would work if not using `conda`.
I'm not sure how this helps, but here is the output of `conda env export`

name: c:\Anaconda3
channels:
- defaults
- conda-forge
dependencies:
- argcomplete=1.8.2=py36_0
- args=0.1.0=py36_0
- arrow-cpp=0.7.1=py36_vc14_2
- boost=1.65.1=py36_vc14_0
- boost-cpp=1.65.1=vc14_1
- bzip2=1.0.6=vc14_1
- clint=0.5.1=py36_0
- curl=7.55.1=vc14_0
- fasteners=0.14.1=py36_2
- feather-format=0.4.0=py36_vc14_2
- freetype=2.7=vc14_2
- hdf5=1.10.1=vc14_1
- hyperspy=1.3=py36_0
- icu=58.2=vc14_0
- jpeg=9b=vc14_2
- jupyter_contrib_core=0.3.3=py36_0
- jupyter_contrib_nbextensions=0.3.3=py36_0
- jupyter_dashboards=0.7.0=py36_0
- jupyter_highlight_selected_word=0.0.10=py36_0
- jupyter_latex_envs=1.3.8.2=py36_1
- jupyter_nbextensions_configurator=0.2.8=py36_0
- krb5=1.14.2=vc14_0
- libiconv=1.14=vc14_4
- libpng=1.6.28=vc14_2
- libssh2=1.8.0=vc14_2
- libtiff=4.0.8=vc14_0
- libwebp=0.5.2=vc14_7
- libxml2=2.9.5=vc14_1
- libxslt=1.1.29=vc14_5
- lzo=2.10=vc14_0
- monotonic=1.3=py36_0
- opencv=3.3.0=py36_200
- openssl=1.0.2l=vc14_0
- parquet-cpp=1.3.0.post=vc14_2
- pillow=4.2.1=py36_0
- pint=0.8.1=py36_0
- pyarrow=0.7.1=py36_vc14_1
- qt=5.6.2=vc14_1
- requests-toolbelt=0.8.0=py36_0
- snappy=1.1.7=vc14_1
- spyder-line-profiler=0.1.1=py36_0
- sqlite=3.19.3=vc14_1
- tk=8.5.19=vc14_1
- twine=1.9.1=py36_2
- websocket-client=0.44.0=py36_0
- wxpython=4.0.0rc1.dev3440+0f9b36e=py36_0
- yaml=0.1.6=vc14_0
- zarr=2.1.4=py36_0
- zlib=1.2.11=vc14_0
- _license=1.1=py36_1
- alabaster=0.7.10=py36hcd07829_0
- anaconda-clean=1.1.0=py36_0
- anaconda-client=1.6.5=py36hd36550c_0
- anaconda=custom=py36h363777c_0
- anaconda-navigator=1.6.9=py36hc720852_0
- anaconda-project=0.8.0=py36h8b3bf89_0
- asn1crypto=0.22.0=py36h8e79faa_1
- astroid=1.5.3=py36h9d85297_0
- astropy=2.0.2=py36h06391c4_4
- babel=2.5.0=py36h35444c1_0
- backports=1.0=py36h81696a8_1
- backports.shutil_get_terminal_size=1.0.0=py36h79ab834_2
- bcolz=1.1.2=py36h783174e_0
- beautifulsoup4=4.6.0=py36hd4cc5e8_1
- bitarray=0.8.1=py36h6af124b_0
- bkcharts=0.2=py36h7e685f7_0
- blaze=0.11.3=py36h8a29ca5_0
- bleach=2.0.0=py36h0a7e3d6_0
- blosc=1.7.0=vc14_2
- bokeh=0.12.4=py36_0
- boto=2.48.0=py36h1a776d2_1
- boto3=1.4.7=py36h56c9f47_0
- botocore=1.7.20=py36hafc5adb_0
- bottleneck=1.2.1=py36hd119dfa_0
- ca-certificates=2017.08.26=h94faf87_0
- cachecontrol=0.12.3=py36hfe50d7b_0
- certifi=2017.7.27.1=py36h043bc9e_0
- cffi=1.10.0=py36hae3d1b5_1
- chardet=3.0.4=py36h420ce6e_1
- chest=0.2.3=py36_0
- click=6.7=py36hec8c647_0
- cloudpickle=0.4.0=py36h639d8dc_0
- clyent=1.2.2=py36hb10d595_1
- colorama=0.3.9=py36h029ae33_0
- colorcet=0.9.1=py36_0
- comtypes=1.1.2=py36heb9b3d1_0
- conda=4.3.30=py36h7e176b0_0
- conda-build=3.0.27=py36h309a530_0
- conda-env=2.6.0=h36134e3_1
- conda-verify=2.0.0=py36h065de53_0
- configobj=5.0.6=py36_0
- console_shortcut=0.1.1=h6bb2dd7_3
- contextlib2=0.5.5=py36he5d52c0_0
- cryptography=2.0.3=py36h123decb_1
- cycler=0.10.0=py36h009560c_0
- cython=0.26.1=py36h18049ac_0
- cytoolz=0.8.2=py36h547e66e_0
- dask=0.15.3=py36h396fcb9_0
- dask-core=0.15.3=py36hd651449_0
- datashader=0.4.0=py36_0
- datashape=0.5.4=py36h5770b85_0
- decorator=4.1.2=py36he63a57b_0
- dill=0.2.7.1=py36hf552773_0
- distlib=0.2.5=py36h51371be_0
- distributed=1.19.1=py36h8504682_0
- docutils=0.14=py36h6012d8f_0
- dynd-python=removed=0
- entrypoints=0.2.3=py36hfd66bb0_2
- et_xmlfile=1.0.1=py36h3d2d736_0
- fastcache=1.0.2=py36hffdae1b_0
- filelock=2.0.12=py36hd7ddd41_0
- flask=0.12.2=py36h98b5e8f_0
- flask-cors=3.0.3=py36h8a3855d_0
- future=0.16.0=py36_1
- get_terminal_size=1.0.0=h38e98db_0
- gevent=1.2.2=py36h342a76c_0
- glob2=0.5=py36h11cc1bd_1
- graphviz=2.38.0=4
- greenlet=0.4.12=py36ha00ad21_0
- h5py=2.7.0=py36hfbe0a52_1
- heapdict=1.0.0=py36h21fa5f4_0
- html5lib=0.999999999=py36ha09b1f3_0
- icc_rt=2017.0.4=h97af966_0
- idna=2.6=py36h148d497_1
- imageio=2.2.0=py36had6c2d2_0
- imagesize=0.7.1=py36he29f638_0
- intel-openmp=2018.0.0=hcd89f80_7
- ipykernel=4.6.1=py36hbb77b34_0
- ipyparallel=6.0.2=py36_0
- ipython=6.1.0=py36h236ecc8_1
- ipython_genutils=0.2.0=py36h3c5d0ee_0
- ipywidgets=7.0.0=py36h2e74ada_0
- isort=4.2.15=py36h6198cc5_0
- itsdangerous=0.24=py36hb6c5a24_1
- jdcal=1.3=py36h64a5255_0
- jedi=0.10.2=py36hed927a0_0
- jinja2=2.9.6=py36h10aa3a0_1
- jmespath=0.9.3=py36h0745840_0
- jsonschema=2.6.0=py36h7636477_0
- jupyter=1.0.0=py36h422fd7e_2
- jupyter_client=5.1.0=py36h9902a9a_0
- jupyter_console=5.2.0=py36h6d89b47_1
- jupyter_core=4.3.0=py36h511e818_0
- jupyterlab=0.27.0=py36h34cc53b_2
- jupyterlab_launcher=0.4.0=py36h22c3ccf_0
- lazy-object-proxy=1.3.1=py36hd1c21d2_0
- libboost=1.65.1=vc14h7b94727_3
- libdynd=0.7.2=0
- line_profiler=2.0=py36_0
- llvmlite=0.20.0=py36_0
- locket=0.2.0=py36hfed976d_1
- lockfile=0.12.2=py36h0468280_0
- lxml=4.1.0=py36h0dcd83c_0
- markupsafe=1.0=py36h0e26971_1
- matplotlib=2.0.2=np113py36_0
- mccabe=0.6.1=py36hb41005a_1
- menuinst=1.4.10=py36h42196fb_0
- mistune=0.7.4=py36h4874169_0
- mkl=2018.0.0=h36b65af_4
- mkl-service=1.1.2=py36h57e144c_4
- mpmath=0.19=py36he326802_2
- msgpack-python=0.4.8=py36h58b1e9d_0
- multipledispatch=0.4.9=py36he44c36e_0
- natsort=5.1.0=py36_0
- navigator-updater=0.1.0=py36h8a7b86b_0
- nbconvert=5.3.1=py36h8dc0fde_0
- nbformat=4.4.0=py36h3a5bc1b_0
- networkx=2.0=py36hff991e3_0
- nltk=3.2.4=py36hd0e0a39_0
- nose=1.3.7=py36h1c3779e_2
- notebook=5.0.0=py36hd9fbf6f_2
- numba=0.35.0=np113py36_10
- numexpr=2.6.2=py36h7ca04dc_1
- numpy=1.13.3=py36ha320f96_0
- numpydoc=0.7.0=py36ha25429e_0
- odo=0.5.1=py36h7560279_0
- olefile=0.44=py36h0a7bdd2_0
- openpyxl=2.4.8=py36hf3b77f6_1
- packaging=16.8=py36ha0986f6_1
- pandas=0.20.3=py36hce827b7_2
- pandoc=1.19.2.1=hb2460c7_1
- pandocfilters=1.4.2=py36h3ef6317_1
- param=1.5.1=py36_0
- paramiko=2.1.2=py36_0
- partd=0.3.8=py36hc8e763b_0
- patch=2.5.9=1
- path.py=10.3.1=py36h3dd8b46_0
- pathlib2=2.3.0=py36h7bfb78b_0
- patsy=0.4.1=py36h42cefec_0
- pep8=1.7.0=py36h0f3d67a_0
- pickleshare=0.7.4=py36h9de030f_0
- pip=9.0.1=py36hadba87b_3
- pkginfo=1.4.1=py36hb0f9cfa_1
- plotly=2.1.0=py36h9404cfe_0
- ply=3.10=py36h1211beb_0
- progress=1.3=py36hbeca8d3_0
- prompt_toolkit=1.0.15=py36h60b8f86_0
- psutil=5.4.0=py36h4e662fb_0
- py=1.4.34=py36ha4aca3a_1
- py-boost=1.65.1=py36_vc14ha18ea20_3
- pyasn1=0.3.7=py36h6f5899e_0
- pycodestyle=2.3.1=py36h7cc55cd_0
- pycosat=0.6.2=py36hf17546d_1
- pycparser=2.18=py36hd053e01_1
- pycrypto=2.6.1=py36he68e6e2_1
- pycurl=7.43.0=py36h086bf4c_3
- pyflakes=1.6.0=py36h0b975d6_0
- pygments=2.2.0=py36hb010967_0
- pylint=1.7.4=py36ha4e6ded_0
- pyodbc=4.0.17=py36h0006bc2_0
- pyopenssl=17.2.0=py36h15ca2fc_0
- pypandoc=1.4=py36_0
- pyparsing=2.2.0=py36h785a196_1
- pyqt=5.6.0=py36hb5ed885_5
- pyreadline=2.1=py36h094d952_1
- pyserial=3.4=py36hcb15667_0
- pysocks=1.6.7=py36h698d350_1
- pytables=3.4.2=py36h71138e3_2
- pytest=3.2.1=py36h753b05e_1
- python=3.6.3=h9e2ca53_1
- python-dateutil=2.6.1=py36h509ddcb_1
- pytz=2017.2=py36h05d413f_1
- pywavelets=0.5.2=py36hc649158_0
- pywin32=221=py36h9c10281_0
- pyyaml=3.12=py36h1d1928f_1
- pyzmq=16.0.2=py36h38c27d9_2
- qtawesome=0.4.4=py36h5aa48f6_0
- qtconsole=4.3.1=py36h99a29a9_0
- qtpy=1.3.1=py36hb8717c5_0
- requests=2.18.4=py36h4371aae_1
- rope=0.10.5=py36hcaf5641_0
- ruamel_yaml=0.11.14=py36h9b16331_2
- s3fs=0.1.2=py36_0
- s3transfer=0.1.10=py36h70b205f_1
- scikit-image=0.13.0=py36h6dffa3f_1
- scikit-learn=0.19.1=py36h53aea1b_0
- scipy=0.19.1=py36h7565378_3
- seaborn=0.8.0=py36h62cb67c_0
- setuptools=36.5.0=py36h65f9e6e_0
- simplegeneric=0.8.1=py36heab741f_0
- singledispatch=3.4.0.3=py36h17d0c80_0
- sip=4.18.1=py36h9c25514_2
- six=1.11.0=py36h4db2310_1
- snowballstemmer=1.2.1=py36h763602f_0
- sockjs-tornado=1.0.3=py36_0
- sortedcollections=0.5.3=py36hbefa0ab_0
- sortedcontainers=1.5.7=py36ha90ac20_0
- sphinx=1.6.3=py36h9bb690b_0
- sphinx_rtd_theme=0.2.4=py36_0
- sphinxcontrib=1.0=py36hbbac3d2_1
- sphinxcontrib-websupport=1.0.1=py36hb5e5916_1
- spyder=3.2.4=py36h8845eaa_0
- sqlalchemy=1.1.13=py36h5948d12_0
- statsmodels=0.8.0=py36h6189b4c_0
- sympy=1.1.1=py36h96708e0_0
- tblib=1.3.2=py36h30f5020_0
- testpath=0.3.1=py36h2698cfe_0
- toolz=0.8.2=py36he152a52_0
- tornado=4.5.2=py36h57f6048_0
- tqdm=4.19.4=py36h02a35f0_0
- traitlets=4.3.2=py36h096827d_0
- traits=4.6.0=py36_0
- typing=3.6.2=py36hb035bda_0
- unicodecsv=0.14.1=py36h6450c06_0
- urllib3=1.22=py36h276f60a_0
- vc=14=h2379b0c_2
- vs2015_runtime=14.0.25123=hd4c4e62_2
- wcwidth=0.1.7=py36h3d5aa90_0
- webencodings=0.5.1=py36h67c50ae_1
- werkzeug=0.12.2=py36h866a736_0
- wheel=0.29.0=py36h6ce6cde_1
- widgetsnbextension=3.0.2=py36h364476f_1
- win_inet_pton=1.0.1=py36he67d7fd_1
- win_unicode_console=0.5=py36hcdbd4b5_0
- wincertstore=0.2=py36h7fe50ca_0
- wrapt=1.10.11=py36he5f5981_0
- xarray=0.9.6=py36_0
- xlrd=1.1.0=py36h1cb58dc_1
- xlsxwriter=1.0.2=py36hf723b7d_0
- xlwings=0.11.4=py36hd3cf94d_0
- xlwt=1.3.0=py36h1a4751e_0
- zict=0.1.3=py36h2d8e73e_0
- pip:
  - backports.shutil-get-terminal-size==1.0.0
  - crccheck==0.6
  - docopt==0.6.2
  - ephem==3.7.6.0
  - et-xmlfile==1.0.1
  - holoviews==1.7.dev5
  - ipython-genutils==0.2.0
  - jupyter-client==5.1.0
  - jupyter-console==5.2.0
  - jupyter-contrib-core==0.3.3
  - jupyter-contrib-nbextensions==0.3.3
  - jupyter-core==4.3.0
  - jupyter-dashboards==0.7.0
  - jupyter-highlight-selected-word==0.0.10
  - jupyter-latex-envs==1.3.8.2
  - jupyter-nbextensions-configurator==0.2.8
  - jupyterlab-launcher==0.4.0
  - kivy==1.10.0
  - kivy-examples==1.10.0
  - kivy-garden==0.1.4
  - kivy.deps.angle==0.1.6
  - kivy.deps.glew==0.1.9
  - kivy.deps.gstreamer==0.1.12
  - kivy.deps.sdl2==0.1.17
  - line-profiler==2.0
  - matlab-kernel==0.10.2
  - metakernel==0.13.6
  - mlabwrap-purepy==0.1
  - mpfshell==0.7.6
  - nbopen==0.4.2
  - paho-mqtt==1.2
  - prompt-toolkit==1.0.15
  - pymatbridge==0.5.2
  - pypiwin32==220
  - pyusb==1.0.0
  - pywake==0.1.3
  - ruamel-yaml==0.11.14
  - selenium==3.4.1
  - sphinx-rtd-theme==0.2.4
  - splinter==0.7.5
  - subprocess-maximize==0.9.0
  - tables==3.4.2
  - thingspeak==0.3.6
  - vanity==2.2.2
  - vispy==0.4.0
  - win-inet-pton==1.0.1
  - win-unicode-console==0.5
prefix: c:\Anaconda3


Mainly was trying to get enough info so someone else could reproduce it if interested.
I see that the example code makes a lot of use of `dask.delayed`. While it can be ok to use `dask.delayed` for some simple cases or at the beginning of a graph, there is a lot of overhead from building an entire graph with `dask.delayed` as it triggers optimize calls, which can become expensive (particularly for large graphs). There are a lot of operations included in Dask (particularly for Arrays), which will be much better suited. Pretty much all (if not all ufuncs) are included (e.g. `sum`) as well as equivalent array creation operations. Would encourage you to browse the [API docs]( http://dask.pydata.org/en/latest/array-api.html ) and try replacing the `delayed` operations with Dask Array operations and see how it goes.
Are you able to reproduce a similar effect using the example code?

I just used delayed and a simple sum function to construct a large graph to start with. My original code can be less trivially converted to a graph without delayed. I probably can use more Dask Array operations, but there will always be delayed functions left. 
Actually I stumbled upon this problem while converting more and more to Dask-Array operations (conversion from function A to function B). The more delayed graph was much faster to construct.
I should remark that adding more Dask-Array operations actually made the code cleaner and easier to read.

You say the delay is possible due to the optimizations of the delayed functions.
So, might it possible to postpone these optimizations calls for all the small tiny additions to the graph, and replace them by one big addition and optimization call?  
Or is there a way to convert a delay function into a proper Dask-Array like function, with less expensive optimizations?
I have not tried running the code yet. This is based on giving it an initial cursory look.

Saw a mention that chunked Dask Arrays were explored, but am not seeing where that happened. 😕 Am only seeing the `delayed` operations. Could you please point out the lines in particular if I'm just overlooking them or include them if they are missing?

Also it sounds like from the OP that the array is being chunked into very small pieces (1 value per chunk). If I'm reading this correctly, that will definitely result in poor performance. It is best to design chunks so that they will hold sizeable NumPy arrays in memory. It would also make sense of why `delayed` performs well here as it forces everything into 1 big chunk (something to keep in mind as you scale), which seems appropriate for this data. Chunks end up making up the nodes in the Dask Graph. So the size of the graph and its performance are directly linked to the chunking.

To provide some context, for my use cases its pretty typical to have ~100M values per chunk. This not only to make graph construction performant, but to make up for things like scheduler overhead. Also to get the largest benefit from using NumPy in parallel. Depending on the size of your data and computation resources, you may want to tweak that up or down. Also [FAQ question 3]( http://dask.pydata.org/en/latest/faq.html ) is pretty relevant here. 

Cannot comment on how best to solve things in your existing codebase without knowing more. That said, would take a look at functions like [`map_blocks`]( http://dask.pydata.org/en/latest/array-api.html#dask.array.Array.map_blocks ), [`map_overlap`]( http://dask.pydata.org/en/latest/array-api.html#dask.array.Array.map_overlap ), [`apply_along_axis`]( http://dask.pydata.org/en/latest/array-api.html#dask.array.apply_along_axis ), and [`apply_over_axes`]( http://dask.pydata.org/en/latest/array-api.html#dask.array.apply_over_axes ) to wrap up difficult to express operations in Dask. If the operation is really involved, [`atop`]( http://dask.pydata.org/en/latest/array-api.html#dask.array.core.atop ) and [`top`]( http://dask.pydata.org/en/latest/array-api.html#dask.array.core.top ) may help as well. These will help you avoid using `delayed` for more complex computations and get the most out of Dask.

I cannot really comment on the details of how `delayed` works as I'm not very intimate with details. Am only passing on recommendations given to me. Maybe someone else ( @mrocklin perhaps? ) can comment. Also they may catch something I have missed.
In function `B` I tried converting the function `A` into a function which uses less delayed, by constructing a Dask-Array earlier in the function (not over whole function `B`). See line `Results.append(array.from_delayed(Bsub(a,i),shape=(N_Signal,),dtype=float))`
Although I could go deeper converting more and more of the inside of function `subB`, at a certain point the overhead will become more than the computation.

The prior Dask Graph is just arbitrary, and in my complex code, it contains both Dask Array and delayed function.  
I changed the example code to a prior Dask graph containing only Dask-Array operations, surprisingly things got worse. Times increased, and in the final stage, memory usage peaked at 2.5GB. The timing of N_Prior 10000 shouldn't be considers valid, due to swapping.

```
# -*- coding: utf-8 -*-
from dask import delayed, array
import numpy as np
import sys
import dask
print('Dask:',dask.__version__)
print('Numpy:',np.__version__)
print('Python:', sys.version)

from time import clock as time
import dask.array as da

N_Signal = 10000


d0 = None
for N_Prior in [1000,2000,10000]:
  print('N_Prior',N_Prior)


  ### Existing graph1
  R1 = da.zeros((1000,1), chunks=(-1,-1))
  for i in range(N_Prior):
    R1 = R1.sum()

  ### Existing graph2
  R2 = da.zeros((1000,1), chunks=(-1,-1))
  for i in range(N_Prior):
    R2 = R2.sum()
    
  @delayed
  def A(a):
    Result = np.zeros((1000,N_Signal))
    for i in range(1000):
      Result[i,:] = np.zeros((N_Signal,)) + i
    return Result
  
  def B(a):
    @delayed
    def Bsub(a,i):
      return np.zeros((N_Signal,)) + i
    Results = []
    for i in range(1000):
      Results.append(array.from_delayed(Bsub(a,i),shape=(N_Signal,),dtype=float))
    return array.stack(Results)
  
  t = time()
  Ra = A(R1)
  matrixA = array.from_delayed(Ra, shape=(1000,N_Signal), dtype=float)
  d1 = time()-t;print('A:',d1);t=time()
  
  Rb = B(R2)
  matrixB = Rb
  d2 = time()-t;print('B:',d2);t=time()
  if not d0:
    d0 = d2
  
  print('Ratio between A & B', round(d2/d1))
  print('Ratio between N_Prior 1000', round(d2/d0,2))
  print('Normalized Ratio', round(d2/d0 * 1000 / N_Prior,2) )
  
  print()
```


```
Dask: 0.16.0
Numpy: 1.13.3
Python: 3.6.3 |Anaconda custom (64-bit)| (default, Nov  8 2017, 15:10:56) [MSC v.1900 64 bit (AMD64)]
N_Prior 1000
A: 0.058604623279165935
B: 72.49667175966209
Ratio between A & B 1237
Ratio between N_Prior 1000 1.0
Normalized Ratio 1.0

N_Prior 2000
A: 0.1338954105965513
B: 157.84215259194596
Ratio between A & B 1179
Ratio between N_Prior 1000 2.18
Normalized Ratio 1.09

N_Prior 10000
A: 0.724686300362464
B: 928.6408255662075
Ratio between A & B 1281
Ratio between N_Prior 1000 12.81
Normalized Ratio 1.28
```
Conversation seems to have died here so I'm going to close.  Happy to reopen if things start up again.

Generally we are working on a variety of ways to improve graph construction time, but they're handled in other issues.  This issue seems to be mostly about a particular application.  If there is a particular action that is being particularly slow then please re-raise.

Commit message: Update setup/python docs for async/await API

Fixes #2895

Patch to review:
diff --git a/docs/source/futures.rst b/docs/source/futures.rst
index e12118c07c9..aca0e1cf079 100644
--- a/docs/source/futures.rst
+++ b/docs/source/futures.rst
@@ -257,7 +257,7 @@ Or collect all futures in batches that had arrived since the last iteration:
       for future, result in batch:
           ...
 
-Additionally, for iterative algorithms, you can add more futures into the 
+Additionally, for iterative algorithms, you can add more futures into the
 ``as_completed`` iterator *during* iteration:
 
 .. code-block:: python
@@ -380,7 +380,7 @@ thread that does not take up a slot within the Dask worker:
 If you intend to do more work in the same thread after waiting on client work,
 you may want to explicitly block until the thread is able to *rejoin* the
 thread pool.  This allows some control over the number of threads that are
-created and stops too many threads from being active at once, over-saturating 
+created and stops too many threads from being active at once, over-saturating
 your hardware:
 
 .. code-block:: python
@@ -575,7 +575,7 @@ Using a consistent name is convenient when you want to lock some known named res
 
    futures = client.map(load, filenames)
 
-Passing around a lock works as well and is easier when you want to create short-term 
+Passing around a lock works as well and is easier when you want to create short-term
 locks for a particular situation:
 
 .. code-block:: python
@@ -793,7 +793,6 @@ API
 .. autoclass:: Future
    :members:
 
-
 .. autoclass:: Queue
    :members:
 
diff --git a/docs/source/setup/python-advanced.rst b/docs/source/setup/python-advanced.rst
index 9a7c34232a4..de0392205ed 100644
--- a/docs/source/setup/python-advanced.rst
+++ b/docs/source/setup/python-advanced.rst
@@ -1,62 +1,207 @@
 Python API (advanced)
 =====================
 
-In some rare cases, experts may want to create ``Scheduler`` and ``Worker``
-objects explicitly in Python manually.  This is often necessary when making
+.. currentmodule:: distributed
+
+In some rare cases, experts may want to create ``Scheduler``, ``Worker``, and
+``Nanny``  objects explicitly in Python.  This is often necessary when making
 tools to automatically deploy Dask in custom settings.
 
-However, often it is sufficient to rely on the :doc:`Dask command line interface
-<cli>`.
+It is more common to create a :doc:`Local cluster with Client() on a single
+machine <single-distributed>` or use the :doc:`Command Line Interface (CLI) <cli>`.
+New readers are recommended to start there.
+
+If you do want to start Scheduler and Worker objects yourself you should be a
+little familiar with ``async``/``await`` style Python syntax.  These objects
+are awaitable and are commonly used within ``async with`` context managers.
+Here are a few examples to show a few ways to start and finish things.
+
+Full Example
+------------
+
+.. autosummary::
+   Scheduler
+   Worker
+   Client
+
+We first start with a comprehensive example of setting up a Scheduler, two Workers,
+and one Client in the same event loop, running a simple computation, and then
+cleaning everything up.
+
+.. code-block:: python
+
+   import asyncio
+   from dask.distributed import Scheduler, Worker, Client
+
+   async def f():
+       async with Scheduler() as s:
+           async with Worker(s.address) as w1, Worker(s.address) as w2:
+               async with Client(s.address, asynchronous=True) as client:
+                   future = client.submit(lambda x: x + 1, 10)
+                   result = await future
+                   print(result)
+
+   asyncio.get_event_loop().run_until_complete(f())
+
+Now we look at simpler examples that build up to this case.
 
 Scheduler
 ---------
 
-To start the Scheduler, provide the listening port (defaults to 8786) and Tornado
-IOLoop (defaults to ``IOLoop.current()``)
+.. autosummary::
+   Scheduler
+
+We create scheduler by creating a ``Scheduler()`` object, and then ``await``
+that object to wait for it to start up.  We can then wait on the ``.finished``
+method to wait until it closes.  In the meantime the scheduler will be active
+managing the cluster..
 
 .. code-block:: python
 
-   from distributed import Scheduler
-   from tornado.ioloop import IOLoop
-   from threading import Thread
+   import asyncio
+   from dask.distributed import Scheduler, Worker
 
-   s = Scheduler()
-   s.start('tcp://:8786')   # Listen on TCP port 8786
+   async def f():
+       s = Scheduler()        # scheduler created, but not yet running
+       s = await Scheduler()  # the scheduler is running
+       await s.finished()     # wait until the scheduler closes
 
-   loop = IOLoop.current()
-   loop.start()
+   asyncio.get_event_loop().run_until_complete(f())
 
-Alternatively, you may want the IOLoop and scheduler to run in a separate
-thread.  In this case, you would replace the ``loop.start()`` call with the
-following:
+This program will run forever, or until some external process connects to the
+scheduler and tells it to stop.  If you want to close things yourself you can
+close any ``Scheduler``, ``Worker``, ``Nanny``, or ``Client`` class by awaiting
+the ``.close`` method:
 
 .. code-block:: python
 
-   t = Thread(target=loop.start, daemon=True)
-   t.start()
+   await s.close()
 
 
 Worker
 ------
 
-On other nodes, start worker processes that point to the URL of the scheduler.
+.. autosummary::
+   Worker
+
+The worker follows the same API.
+The only difference is that the worker needs to know the address of the
+scheduler.
 
 .. code-block:: python
 
-   from distributed import Worker
-   from tornado.ioloop import IOLoop
-   from threading import Thread
+   import asyncio
+   from dask.distributed import Scheduler, Worker
 
-   w = Worker('tcp://127.0.0.1:8786')
-   w.start()  # choose randomly assigned port
+   async def f(scheduler_address):
+       w = await Worker(scheduler_address)
+       await w.finished()
 
-   loop = IOLoop.current()
-   loop.start()
+   asyncio.get_event_loop().run_until_complete(f("tcp://127.0.0.1:8786"))
 
-Alternatively, replace ``Worker`` with ``Nanny`` if you want your workers to be
-managed in a separate process by a local nanny process.  This allows workers to
-restart themselves in case of failure. Also, it provides some additional monitoring, 
-and is useful when coordinating many workers that should live in different
-processes in order to avoid the GIL_.
+
+Start many in one event loop
+----------------------------
+
+.. autosummary::
+   Scheduler
+   Worker
+
+We can run as many of these objects as we like in the same event loop.
+
+.. code-block:: python
+
+   import asyncio
+   from dask.distributed import Scheduler, Worker
+
+   async def f():
+       s = await Scheduler()
+       w = await Worker(s.address)
+       await w.finished()
+       await s.finished()
+
+   asyncio.get_event_loop().run_until_complete(f())
+
+
+Use Context Managers
+--------------------
+
+We can also use ``async with`` context managers to make sure that we clean up
+properly.  Here is the same example as from above:
+
+.. code-block:: python
+
+   import asyncio
+   from dask.distributed import Scheduler, Worker
+
+   async def f():
+       async with Scheduler() as s:
+           async with Worker(s.address) as w:
+               await w.finished()
+               await s.finished()
+
+   asyncio.get_event_loop().run_until_complete(f())
+
+Alternatively, in the example below we also include a ``Client``, run a small
+computation, and then allow things to clean up after that computation..
+
+.. code-block:: python
+
+   import asyncio
+   from dask.distributed import Scheduler, Worker, Client
+
+   async def f():
+       async with Scheduler() as s:
+           async with Worker(s.address) as w1, Worker(s.address) as w2:
+               async with Client(s.address, asynchronous=True) as client:
+                   future = client.submit(lambda x: x + 1, 10)
+                   result = await future
+                   print(result)
+
+   asyncio.get_event_loop().run_until_complete(f())
+
+This is equivalent to creating and ``awaiting`` each server, and then calling
+``.close`` on each as we leave the context.
+In this example we don't wait on ``s.finished()``, so this will terminate
+relatively quickly.  You could have called ``await s.finished()`` though if you
+wanted this to run forever.
+
+Nanny
+-----
+
+.. autosummary::
+   Nanny
+
+Alternatively, we can replace ``Worker`` with ``Nanny`` if we want your workers
+to be managed in a separate process.  The ``Nanny`` constructor follows the
+same API. This allows workers to restart themselves in case of failure. Also,
+it provides some additional monitoring, and is useful when coordinating many
+workers that should live in different processes in order to avoid the GIL_.
+
+.. code-block:: python
+
+   # w = await Worker(s.address)
+   w = await Nanny(s.address)
 
 .. _GIL: https://docs.python.org/3/glossary.html#term-gil
+
+
+API
+---
+
+Scheduler
+~~~~~~~~~
+.. autoclass:: Scheduler
+   :members:
+
+Worker
+~~~~~~
+
+.. autoclass:: Worker
+   :members:
+
+Nanny
+~~~~~
+
+.. autoclass:: Nanny
+   :members:


Code review comments:
diff_hunk: @@ -1,62 +1,207 @@
 Python API (advanced)
 =====================
 
-In some rare cases, experts may want to create ``Scheduler`` and ``Worker``
-objects explicitly in Python manually.  This is often necessary when making
+.. currentmodule:: distributed
+
+In some rare cases, experts may want to create ``Scheduler``, ``Worker``, and
+``Nanny``  objects explicitly in Python.  This is often necessary when making
 tools to automatically deploy Dask in custom settings.
 
-However, often it is sufficient to rely on the :doc:`Dask command line interface
-<cli>`.
+It is more common to create a :doc:`Local cluster with Client() on a single
+machine <single-distributed>` or use the :doc:`Command Line Interface (CLI) <cli>`.
+New readers are recommended to start there.
+
+If you do want to start Scheduler and Worker objects yourself you should be a
+little familiar with ``async``/``await`` style Python syntax.  These objects
+are awaitable and are commonly used within ``async with`` context managers.
+Here are a few examples to show a few ways to start and finish things.
+
+Full Example
+------------
+
+.. autosummary::
+   Scheduler
+   Worker
+   Client
+
+We first start with a comprehensive example of setting up a Scheduler, two Workers,
+and one Client in the same event loop, running a simple computation, and then
+cleaning everything up.
+
+.. code-block:: python
+
+   import asyncio
+   from dask.distributed import Scheduler, Worker, Client
+
+   async def f():
+       async with Scheduler() as s:
+           async with Worker(s.address) as w1, Worker(s.address) as w2:
+               async with Client(s.address, asynchronous=True) as client:
+                   future = client.submit(lambda x: x + 1, 10)
+                   result = await future
+                   print(result)
+
+   asyncio.get_event_loop().run_until_complete(f())
+
+Now we look at simpler examples that build up to this case.
 
 Scheduler
 ---------
 
-To start the Scheduler, provide the listening port (defaults to 8786) and Tornado
-IOLoop (defaults to ``IOLoop.current()``)
+.. autosummary::
+   Scheduler
+
+We create scheduler by creating a ``Scheduler()`` object, and then ``await``
+that object to wait for it to start up.  We can then wait on the ``.finished``
+method to wait until it closes.  In the meantime the scheduler will be active
+managing the cluster..
 
 .. code-block:: python
 
-   from distributed import Scheduler
-   from tornado.ioloop import IOLoop
-   from threading import Thread
+   import asyncio
+   from dask.distributed import Scheduler, Worker
 
-   s = Scheduler()
-   s.start('tcp://:8786')   # Listen on TCP port 8786
+   async def f():
+       s = Scheduler()        # scheduler created, but not yet running
+       s = await Scheduler()  # the scheduler is running
path: docs/source/setup/python-advanced.rst
text: @user1:
What happens if the s = Scheduler(*args) and await s = Scheduler(*args) have different args?