Repository: dask/dask

Issue:
request: HTTP file system
It would be convenient to be able to do `open_files` on a list of HTTP URLs. This cannot be a complete file-system like the other backends, because there is no possibility of `glob()`, but I still think it would be useful for accessing remote data. 

In general, the size of a given URL's data can be found through a HEAD command, and chunks read using the `Range` header keyword; some servers do not support that, however, and then downloads could only be parallelised across multiple files.

Discussion:
martindurant: Note that the simple http server used for testing here does not actually support Range, so the code is not very complete; should I write a little tornado thing, actually hosting some sorts of files that we might want to work with (csv...) ?
mrocklin: Testing against tornado would be fine.  

Alternatively I think that some of the dask.bag test suite still tests against the open internet.  They mark with `@pytest.mark.network` for people who want to avoid network-based tests.
martindurant: A couple of things to consider here:
- some servers do not provide the content-length with a HEAD call, so we cannot know the size of those objects before downloading them
- some servers do not respect the Range header, and send the whole object every time
- some servers still do not provide the length even when getting the data. In other cases they might, and streamed download mode could be used to bail in the case that the download looks too big

If the length is not known, or Range is ignored, clearly we cannot use intra-file partitioning. Is it then an error to do anything other than `read()` from position 0? For a smallish file (less than the block-size), we would just download that block and be able to seek within it. So perhaps the fail condition is: we are doing something other than `tell()==0, read()`, we tried to download a block (say 5MB), and while streaming, the header says the arriving data is bigger than we asked for OR we are streaming and have already seen more data than we asked for.

Thoughts on this?
mrocklin: My hope would be that the servers that would serve large datasets *would* support these features.  Is that likely to be true?

Is it possible to check if they support these and, if not, err?
martindurant: We can know up front whether the content-length is missing in the HEAD call, we could error early on that, but it should not prevent the simple file-access pattern of one full file per call.

We cannot know whether Range is supported without trying with a Range header. [This SO answer](https://stackoverflow.com/questions/720419/how-can-i-find-out-whether-a-server-supports-the-range-header) says that you can combine HEAD and Range to see if the server supports it without getting data

That could mean three calls to get any data: one HEAD to get the size, one HEAD to check Range, and then a GET to actually fetch data.
mrocklin: That sounds unpleasant.  What do you think is best?
martindurant: Here is my stab at putting checks and expressive error messages. I don't know how to go about testing these bits, though.
martindurant: I think this is a useful addition at this point. There may be more polish needed down the road, but I would defer them to future PRs. Any thoughts?

martindurant: (appveyor failure appears unrelated, possibly the numpy version problem that has been noticed elsewhere)
mrocklin: Appveyor issues resolved 
mrocklin: +1 from me on the code.  Should this capability be mentioned in documentation somewhere?
martindurant: Yes, it should be in the changelog, and the limitations mentioned on http://dask.pydata.org/en/latest/remote-data-services.html

Patch:
diff --git a/dask/bytes/core.py b/dask/bytes/core.py
index 3e99a7d7a..5585ae068 100644
--- a/dask/bytes/core.py
+++ b/dask/bytes/core.py
@@ -449,6 +449,15 @@ def get_fs(protocol, storage_options=None):
     elif protocol == 'hdfs':
         cls = get_hdfs_driver(_globals.get("hdfs_driver", "auto"))
 
+    elif protocol in ['http', 'https']:
+        import_required('requests',
+                        "Need to install `requests` for HTTP support\n"
+                        "   conda install requests\n"
+                        "    or\n"
+                        "   pip install requests")
+        import dask.bytes.http  # noqa, registers HTTP backend
+        cls = _filesystems[protocol]
+
     else:
         raise ValueError("Unknown protocol %s" % protocol)
 
diff --git a/dask/bytes/http.py b/dask/bytes/http.py
new file mode 100644
index 000000000..b0563a2e0
--- /dev/null
+++ b/dask/bytes/http.py
@@ -0,0 +1,283 @@
+from __future__ import print_function, division, absolute_import
+
+import requests
+import uuid
+
+from . import core
+
+DEFAULT_BLOCK_SIZE = 5 * 2 ** 20
+
+
+class HTTPFileSystem(core.FileSystem):
+    """
+    Simple File-System for fetching data via HTTP(S)
+
+    Unlike other file-systems, HTTP is limited in that it does not provide glob
+    or write capability.
+    """
+    sep = '/'
+
+    def __init__(self, **storage_options):
+        """
+        Parameters
+        ----------
+        storage_options: key-value
+            May be credentials, e.g., `{'auth': ('username', 'pword')}` or any
+            other parameters for requests
+        """
+        self.block_size = storage_options.pop('block_size', DEFAULT_BLOCK_SIZE)
+        self.kwargs = storage_options
+        self.session = requests.Session()
+
+    def glob(self, url):
+        """For a template path, return matching files"""
+        raise NotImplementedError
+
+    def mkdirs(self, url):
+        """Make any intermediate directories to make path writable"""
+        raise NotImplementedError
+
+    def open(self, url, mode='rb', block_size=None, **kwargs):
+        """Make a file-like object
+
+        Parameters
+        ----------
+        url: str
+            Full URL with protocol
+        mode: string
+            must be "rb"
+        kwargs: key-value
+            Any other parameters, passed to requests calls
+        """
+        if mode != 'rb':
+            raise NotImplementedError
+        block_size = block_size if block_size is not None else self.block_size
+        return HTTPFile(url, self.session, block_size, **self.kwargs)
+
+    def ukey(self, url):
+        """Unique identifier, implied file might have changed every time"""
+        return uuid.uuid1().hex
+
+    def size(self, url):
+        """Size in bytes of the file at path"""
+        return file_size(url, session=self.session, **self.kwargs)
+
+
+core._filesystems['http'] = HTTPFileSystem
+core._filesystems['https'] = HTTPFileSystem
+
+
+class HTTPFile(object):
+    """
+    A file-like object pointing to a remove HTTP(S) resource
+
+    Supports only reading, with read-ahead of a predermined block-size.
+
+    In the case that the server does not supply the filesize, only reading of
+    the complete file in one go is supported.
+
+    Parameters
+    ----------
+    url: str
+        Full URL of the remote resource, including the protocol
+    session: requests.Session or None
+        All calls will be made within this session, to avoid restarting
+        connections where the server allows this
+    block_size: int or None
+        The amount of read-ahead to do, in bytes. Default is 5MB, or the value
+        configured for the FileSystem creating this file
+    kwargs: all other key-values are passed to reqeuests calls.
+    """
+
+    def __init__(self, url, session=None, block_size=None, **kwargs):
+        self.url = url
+        self.kwargs = kwargs
+        self.loc = 0
+        self.session = session if session is not None else requests.Session()
+        self.blocksize = (block_size if block_size is not None
+                          else DEFAULT_BLOCK_SIZE)
+        try:
+            self.size = file_size(url, self.session, allow_redirects=True,
+                                  **self.kwargs)
+        except ValueError:
+            # No size information - only allow read() and no seek()
+            self.size = None
+        self.cache = None
+        self.closed = False
+        self.start = None
+        self.end = None
+
+    def seek(self, where, whence=0):
+        """Set file position
+
+        Parameters
+        ----------
+        where: int
+            Location to set
+        whence: int (default 0)
+            If zero, set from start of file (value should be positive); if 1,
+            set relative to current position; if 2, set relative to end of file
+            (value shoulf be negative)
+
+        Returns the position.
+        """
+        if self.size is None:
+            raise ValueError('Cannot seek since size of file is not known')
+        if whence == 0:
+            nloc = where
+        elif whence == 1:
+            nloc += where
+        elif whence == 2:
+            nloc = self.size + where
+        else:
+            raise ValueError('Whence must be in [1, 2, 3], but got %s' % whence)
+        if nloc < 0:
+            raise ValueError('Seek before start of file')
+        self.loc = nloc
+        return nloc
+
+    def tell(self):
+        """Get current file byte position"""
+        return self.loc
+
+    def read(self, length=-1):
+        """Read bytes from file
+
+        Parameters
+        ----------
+        length: int
+            Read up to this many bytes. If negative, read all content to end of
+            file. If the server has not supplied the filesize, attempting to
+            read only part of the data will raise a ValueError.
+        """
+        if self.size is None:
+            if length >= 0:
+                raise ValueError('File size is unknown, must read all data')
+            else:
+                return self._fetch_all()
+        if length < 0 or self.loc + length > self.size:
+            end = self.size
+        else:
+            end = self.loc + length
+        if self.loc >= self.size:
+            return b''
+        self. _fetch(self.loc, end)
+        data = self.cache[self.loc - self.start:end - self.start]
+        self.loc = end
+        return data
+
+    def _fetch(self, start, end):
+        """Set new bounds for data cache and fetch data, if required"""
+        if self.start is None and self.end is None:
+            # First read
+            self.start = start
+            self.end = end + self.blocksize
+            self.cache = self._fetch_range(start, self.end)
+        elif start < self.start:
+            if self.end - end > self.blocksize:
+                self.start = start
+                self.end = end + self.blocksize
+                self.cache = self._fetch_range(self.start, self.end)
+            else:
+                new = self._fetch_range(start, self.start)
+                self.start = start
+                self.cache = new + self.cache
+        elif end > self.end:
+            if self.end > self.size:
+                return
+            if end - self.end > self.blocksize:
+                self.start = start
+                self.end = end + self.blocksize
+                self.cache = self._fetch_range(self.start, self.end)
+            else:
+                new = self._fetch_range(self.end, end + self.blocksize)
+                self.end = end + self.blocksize
+                self.cache = self.cache + new
+
+    def _fetch_all(self):
+        """Read whole file in one shot, without caching
+
+        This is only called when size is None and read() is called without a
+        byte-count.
+        """
+        r = self.session.get(self.url, **self.kwargs)
+        r.raise_for_status()
+        return r.content
+
+    def _fetch_range(self, start, end):
+        """Download a block of data
+
+        The expectation is that the server returns only the requested bytes,
+        with HTTP code 206. If this is not the case, we first check the headers,
+        and then stream the output - if the data size is bigger than we
+        requested, an exception is raised.
+        """
+        kwargs = self.kwargs.copy()
+        headers = self.kwargs.pop('headers', {})
+        headers['Range'] = 'bytes=%i-%i' % (start, end - 1)
+        r = self.session.get(self.url, headers=headers, stream=True, **kwargs)
+        r.raise_for_status()
+        if r.status_code == 206:
+            # partial content, as expected
+            return r.content
+        if 'Content-Length' in r.headers:
+            cl = int(r.headers['Content-Length'])
+            if cl <= end - start:
+                # data size OK
+                return r.content
+            else:
+                raise ValueError('Got more bytes (%i) than requested (%i)' % (
+                    cl, end - start))
+        cl = 0
+        out = []
+        for chunk in r.iter_content(chunk_size=2 ** 20):
+            # data size unknown, let's see if it goes too big
+            if chunk:
+                out.append(chunk)
+                cl += len(chunk)
+                if cl > end - start:
+                    raise ValueError(
+                        'Got more bytes so far (>%i) than requested (%i)' % (
+                            cl, end - start))
+            else:
+                break
+        return b''.join(out)
+
+    def __enter__(self):
+        self.loc = 0
+        return self
+
+    def __exit__(self, *args):
+        self.close()
+
+    def __iter__(self):
+        # no text lines here, use TextIOWrapper
+        raise NotImplementedError
+
+    def write(self):
+        raise NotImplementedError
+
+    def flush(self):
+        pass
+
+    def close(self):
+        self.closed = True
+
+    def seekable(self):
+        return True
+
+    def writable(self):
+        return False
+
+    def readable(self):
+        return True
+
+
+def file_size(url, session, **kwargs):
+    """Call HEAD on the server to get file size"""
+    r = session.head(url, **kwargs)
+    r.raise_for_status()
+    if 'Content-Length' in r.headers:
+        return int(r.headers['Content-Length'])
+    else:
+        raise ValueError("Server did not supply size of %s" % url)
diff --git a/dask/bytes/utils.py b/dask/bytes/utils.py
index ad0fb3968..619eaf385 100644
--- a/dask/bytes/utils.py
+++ b/dask/bytes/utils.py
@@ -51,6 +51,10 @@ def infer_storage_options(urlpath, inherit_storage_options=None):
         if windows_path:
             path = '%s:%s' % windows_path.groups()
 
+    if protocol in ['http', 'https']:
+        # for HTTP, we don't want to parse, as requests will anyway
+        return {'protocol': protocol, 'path': urlpath}
+
     options = {
         'protocol': protocol,
         'path': path,
diff --git a/docs/source/array-creation.rst b/docs/source/array-creation.rst
index d7bf27c0e..26be15e01 100644
--- a/docs/source/array-creation.rst
+++ b/docs/source/array-creation.rst
@@ -1,22 +1,28 @@
 Create Dask Arrays
 ==================
 
-We store and manipulate large arrays in a wide variety of ways.  There are some
-standards like HDF5 and NetCDF but just as often people use custom storage
-solutions.  This page talks about how to build dask graphs to interact with
-your array.
+You can load or store dask arrays from a variety of common sources like HDF5,
+and NetCDF, `Zarr <http://zarr.readthedocs.io/en/stable/>`_, or any format that
+supports Numpy-style slicing.
 
-In principle we need functions that return NumPy arrays.  These functions and
-their arrangement can be as simple or as complex as the situation dictates.
+.. currentmodule:: dask.array
 
+.. autosummary::
+   from_array
+   from_delayed
+   from_npy_stack
+   stack
+   concatenate
 
-Simple case - Format Supports NumPy Slicing
--------------------------------------------
+NumPy Slicing
+-------------
+
+.. autosummary::
+   from_array
 
 Many storage formats have Python projects that expose storage using NumPy
 slicing syntax.  These include HDF5, NetCDF, BColz, Zarr, GRIB, etc..  For
-example the ``HDF5`` file format has the ``h5py`` Python project, which
-provides a ``Dataset`` object into which we can slice in NumPy fashion.
+example we can load a Dask array from an HDF5 file using `h5py <http://www.h5py.org/>`_:
 
 .. code-block:: Python
 
@@ -28,45 +34,79 @@ provides a ``Dataset`` object into which we can slice in NumPy fashion.
 
    >>> x = d[:5, :5]                # We slice to get numpy arrays
 
-It is common for Python wrappers of on-disk array formats to present a NumPy
-slicing syntax.  The full dataset looks like a NumPy array with ``.shape`` and
-``.dtype`` attributes even though the data hasn't yet been loaded in and still
-lives on disk.  Slicing in to this array-like object fetches the appropriate
-data from disk and returns that region as an in-memory NumPy array.
-
-For this common case ``dask.array`` presents the convenience function
-``da.from_array``
+Given an object like ``d`` above that has ``dtype`` and ``shape`` properties
+and that supports Numpy style slicing we can construct a lazy Dask array.
 
 .. code-block:: Python
 
    >>> import dask.array as da
    >>> x = da.from_array(d, chunks=(1000, 1000))
 
+This process is entirely lazy.  Neither creating the h5py object nor wrapping
+it with ``da.from_array`` have loaded any data.
+
 
 Concatenation and Stacking
 --------------------------
 
-Often we store data in several different locations and want to stitch them
-together.
+.. autosummary::
+   stack
+   concatenate
+
+Often we store data in several different locations and want to stitch them together.
 
 .. code-block:: Python
 
-   >>> filenames = sorted(glob('2015-*-*.hdf5')
-   >>> dsets = [h5py.File(fn)['/data'] for fn in filenames]
-   >>> arrays = [da.from_array(dset, chunks=(1000, 1000)) for dset in dsets]
-   >>> x = da.concatenate(arrays, axis=0)  # Concatenate arrays along first axis
+    dask_arrays = []
+    for fn in filenames:
+        f = h5py.File(fn)
+        d = f['/data']
+        x = da.from_array(d, chunks=(1000, 1000))
+        dask_arrays.append(x)
+
+    x = da.concatenate(arrays, axis=0)  # concatenate arrays along first axis
 
 For more information see :doc:`concatenation and stacking <array-stack>` docs.
 
+
 Using ``dask.delayed``
 ----------------------
 
-You can create a plan to arrange many numpy arrays into a grid with normal for
-loops using :doc:`dask.delayed<delayed-overview>` and then convert each of these
-Dask.delayed objects into a single-chunk Dask array with ``da.from_delayed``.
-You can then arrange these single-chunk Dask arrays into a larger
-multiple-chunk Dask array using :doc:`concatenation and stacking <array-stack>`,
-as described above.
+.. autosummary::
+   from_delayed
+   stack
+   concatenate
+
+Sometimes Numpy-style data resides in formats that do not support numpy-style
+slicing.  We can still construct Dask arrays around this data if we have a
+Python function that can generate pieces of the full array if we use
+:doc:`dask.delayed <delayed>`.  Dask delayed lets us delay a single function
+call that would create a numpy array.  We can then wrap this delayed object
+with ``da.from_delayed``, providing a dtype and shape to produce a
+single-chunked Dask array.  We can then use ``stack`` or ``concatenate`` from
+before to construct a larger lazy array.
+
+
+As an example, consider loading a stack of images using ``skimage.io.imread``:
+
+.. code-block:: python
+
+    import skimage.io
+    import dask.array as da
+    import dask
+
+    imread = dask.delayed(skimage.io.imread, pure=True)  # Lazy version of imread
+
+    filenames = sorted(glob.glob('*.jpg'))
+
+    lazy_images = [imread(url) for url in urls]     # Lazily evaluate imread on each url
+
+    arrays = [da.from_delayed(lazy_image,           # Construct a small Dask array
+                              dtype=sample.dtype,   # for every lazy value
+                              shape=sample.shape)
+              for lazy_value in lazy_values]
+
+    stack = da.stack(arrays, axis=0)                # Stack all small Dask arrays into one
 
 See :doc:`documentation on using dask.delayed with collections<delayed-collections>`.
 
@@ -260,9 +300,18 @@ For example, if you plan to take out thin slices along the first dimension then
 Store Dask Arrays
 =================
 
+.. autosummary::
+   store
+   to_hdf5
+   to_npy_stack
+   compute
+
 In Memory
 ---------
 
+.. autosummary::
+   compute
+
 If you have a small amount of data, you can call ``np.array`` or ``.compute()``
 on your Dask array to turn in to a normal NumPy array:
 
@@ -277,41 +326,48 @@ on your Dask array to turn in to a normal NumPy array:
    array([0, 1, 4, 9, 16, 25])
 
 
-HDF5
-----
+Numpy style slicing
+-------------------
+
+.. autosummary::
+   store
 
-Use the ``to_hdf5`` function to store data into HDF5 using ``h5py``:
+You can store dask arrays in any object that supports numpy-style slice
+assignment like ``h5py.Dataset``:
 
 .. code-block:: Python
 
-   >>> da.to_hdf5('myfile.hdf5', '/y', y)  # doctest: +SKIP
+   >>> import h5py
+   >>> f = h5py.File('myfile.hdf5')
+   >>> d = f.require_dataset('/data', shape=x.shape, dtype=x.dtype)
+   >>> da.store(x, d)
 
-Store several arrays in one computation with the function
-``da.to_hdf5`` by passing in a dict:
+You can store several arrays in one computation by passing lists of sources and
+destinations:
 
 .. code-block:: Python
 
-   >>> da.to_hdf5('myfile.hdf5', {'/x': x, '/y': y})  # doctest: +SKIP
+   >>> da.store([array1, array2], [output1, output2])  # doctest: +SKIP
 
+HDF5
+----
 
-Other On-Disk Storage
----------------------
+.. autosummary::
+   to_hdf5
 
-Alternatively, you can store dask arrays in any object that supports numpy-style
-slice assignment like ``h5py.Dataset``, or ``bcolz.carray``:
+HDF5 is sufficiently common that there is a special function, ``to_hdf5`` to
+store data into HDF5 files using ``h5py``:
 
 .. code-block:: Python
 
-   >>> import bcolz  # doctest: +SKIP
-   >>> out = bcolz.zeros(shape=y.shape, rootdir='myfile.bcolz')  # doctest: +SKIP
-   >>> da.store(y, out)  # doctest: +SKIP
+   >>> da.to_hdf5('myfile.hdf5', '/y', y)  # doctest: +SKIP
 
-You can store several arrays in one computation by passing lists of sources and
-destinations:
+Store several arrays in one computation with the function
+``da.to_hdf5`` by passing in a dict:
 
 .. code-block:: Python
 
-   >>> da.store([array1, array2], [output1, output2])  # doctest: +SKIP
+   >>> da.to_hdf5('myfile.hdf5', {'/x': x, '/y': y})  # doctest: +SKIP
 
 
 Plugins
diff --git a/docs/source/array-ghost.rst b/docs/source/array-ghost.rst
index 33bad3a0b..b92d39c35 100644
--- a/docs/source/array-ghost.rst
+++ b/docs/source/array-ghost.rst
@@ -1,5 +1,5 @@
-Overlapping Blocks with Ghost Cells
-===================================
+Overlapping Computations
+========================
 
 Some array operations require communication of borders between neighboring
 blocks.  Example operations include the following:
diff --git a/docs/source/array-overview.rst b/docs/source/array-overview.rst
deleted file mode 100644
index 6977e0adb..000000000
--- a/docs/source/array-overview.rst
+++ /dev/null
@@ -1,73 +0,0 @@
-Overview
-========
-
-Dask Array implements a subset of the NumPy ndarray interface using blocked
-algorithms, cutting up the large array into many small arrays. This lets us
-compute on arrays larger than memory using all of our cores.  We coordinate
-these blocked algorithms using dask graphs.
-
-Design
-------
-
-.. image:: images/dask-array-black-text.svg
-   :alt: Dask arrays coordinate many numpy arrays
-   :align: right
-
-Dask arrays coordinate many NumPy arrays arranged into a grid.  These
-NumPy arrays may live on disk or on other machines.
-
-Common Uses
------------
-
-Today Dask array is commonly used in the sort of gridded data analysis that
-arises in weather, climate modeling, or oceanography, especially when data
-sizes become inconveniently large.  Dask array complements large on-disk array
-stores like HDF5, NetCDF, and BColz.  Additionally Dask array is commonly used
-to speed up expensive in-memory computations using multiple cores, such as you
-might find in image analysis or statistical and machine learning applications.
-
-Scope
------
-
-The ``dask.array`` library supports the following interface from ``numpy``:
-
-*  Arithmetic and scalar mathematics, ``+, *, exp, log, ...``
-*  Reductions along axes, ``sum(), mean(), std(), sum(axis=0), ...``
-*  Tensor contractions / dot products / matrix multiply, ``tensordot``
-*  Axis reordering / transpose, ``transpose``
-*  Slicing, ``x[:100, 500:100:-2]``
-*  Fancy indexing along single axes with lists or numpy arrays, ``x[:, [10, 1, 5]]``
-*  The array protocol ``__array__``
-*  Some linear algebra ``svd, qr, solve, solve_triangular, lstsq``
-
-See :doc:`the dask.array API<array-api>` for a more extensive list of
-functionality.
-
-Execution
----------
-
-By default Dask array uses the threaded scheduler in order to avoid data
-transfer costs and because NumPy releases the GIL well.  It is also quite
-effective on a cluster using the `dask.distributed`_ scheduler.
-
-.. _`dask.distributed`: https://distributed.readthedocs.io/en/latest/
-
-Limitations
------------
-
-Dask array does not implement the entire numpy interface.  Users expecting this
-will be disappointed.  Notably, Dask array has the following limitations:
-
-1.  Dask array does not implement all of ``np.linalg``.  This has been done by a
-    number of excellent BLAS/LAPACK implementations, and is the focus of
-    numerous ongoing academic research projects.
-2.  Dask array with unknown shapes do not support all operations
-3.  Dask array does not attempt operations like ``sort`` which are notoriously
-    difficult to do in parallel, and are of somewhat diminished value on very
-    large data (you rarely actually need a full sort).
-    Often we include parallel-friendly alternatives like ``topk``.
-4.  Dask array doesn't implement operations like ``tolist`` that would be very
-    inefficient for larger datasets. Likewise it is very inefficient to iterate
-    over a Dask array with for loops.
-5.  Dask development is driven by immediate need, and so many lesser used
-    functions have not been implemented. Community contributions are encouraged.
diff --git a/docs/source/array.rst b/docs/source/array.rst
index 16b52df8d..e8208b29c 100644
--- a/docs/source/array.rst
+++ b/docs/source/array.rst
@@ -1,25 +1,83 @@
 Array
 =====
 
-Dask arrays implement a subset of the NumPy interface on large arrays using
-blocked algorithms and task scheduling.
-
 .. toctree::
    :maxdepth: 1
+   :hidden:
 
-   array-overview.rst
-   array-creation.rst
    array-api.rst
-
-Other topics
-
-.. toctree::
-   :maxdepth: 1
-
-   array-slicing.rst
-   array-stack.rst
+   array-creation.rst
    array-ghost.rst
    array-design.rst
-   array-linear-operator.rst
    array-sparse.rst
    array-stats.rst
+   array-linear-operator.rst
+   array-slicing.rst
+   array-stack.rst
+
+Dask Array implements a subset of the NumPy ndarray interface using blocked
+algorithms, cutting up the large array into many small arrays. This lets us
+compute on arrays larger than memory using all of our cores.  We coordinate
+these blocked algorithms using dask graphs.
+
+Design
+------
+
+.. image:: images/dask-array-black-text.svg
+   :alt: Dask arrays coordinate many numpy arrays
+   :align: right
+
+Dask arrays coordinate many NumPy arrays arranged into a grid.  These
+NumPy arrays may live on disk or on other machines.
+
+Common Uses
+-----------
+
+Dask array used in fields like atmospheric and oceanographic science, large
+scale imaging, genomics, numerical algorithms for optimization or statistics ,
+and more.
+
+Scope
+-----
+
+Dask arrays supports most of the Numpy interface like the following:
+
+-  Arithmetic and scalar mathematics, ``+, *, exp, log, ...``
+-  Reductions along axes, ``sum(), mean(), std(), sum(axis=0), ...``
+-  Tensor contractions / dot products / matrix multiply, ``tensordot``
+-  Axis reordering / transpose, ``transpose``
+-  Slicing, ``x[:100, 500:100:-2]``
+-  Fancy indexing along single axes with lists or numpy arrays, ``x[:, [10, 1, 5]]``
+-  Array protocols like ``__array__``, and ``__array_ufunc__``
+-  Some linear algebra ``svd, qr, solve, solve_triangular, lstsq``
+-  ...
+
+
+However Dask array does not implement the entire numpy interface.  Users expecting this
+will be disappointed.  Notably, Dask array lacks the following features:
+
+-   Much of ``np.linalg`` has not been implemented.
+    This has been done by a number of excellent BLAS/LAPACK implementations,
+    and is the focus of numerous ongoing academic research projects.
+-   Arrays with unknown shapes do not support all operations
+-   Operations like ``sort`` which are notoriously
+    difficult to do in parallel, and are of somewhat diminished value on very
+    large data (you rarely actually need a full sort).
+    Often we include parallel-friendly alternatives like ``topk``.
+-   Dask array doesn't implement operations like ``tolist`` that would be very
+    inefficient for larger datasets. Likewise it is very inefficient to iterate
+    over a Dask array with for loops.
+-   Dask development is driven by immediate need, and so many lesser used
+    functions have not been implemented. Community contributions are encouraged.
+
+See :doc:`the dask.array API<array-api>` for a more extensive list of
+functionality.
+
+Execution
+---------
+
+By default Dask array uses the threaded scheduler in order to avoid data
+transfer costs and because NumPy releases the GIL well.  It is also quite
+effective on a cluster using the `dask.distributed`_ scheduler.
+
+.. _`dask.distributed`: https://distributed.readthedocs.io/en/latest/
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index 0d6de29b8..5a6ee716b 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -24,6 +24,7 @@ Bag
 Core
 ++++
 
+- New file-system for HTTP(S), allowing direct loading from specific URLs (:pr:`3160`) `Martin Durant`_
 
 0.17.0 / 2018-02-09
 -------------------
diff --git a/docs/source/conf.py b/docs/source/conf.py
index 72713486a..326fd2337 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -12,6 +12,7 @@
 # serve to show the default.
 
 import sys, os
+from shutil import copyfile
 
 # If extensions (or modules to document with autodoc) are in another directory,
 # add these directories to sys.path here. If the directory is relative to the
@@ -312,3 +313,37 @@ intersphinx_mapping = {'pandas': ('http://pandas.pydata.org/pandas-docs/stable/'
                                   'http://pandas.pydata.org/pandas-docs/stable/objects.inv'),
                        'numpy': ('https://docs.scipy.org/doc/numpy/',
                                  'https://docs.scipy.org/doc/numpy/objects.inv')}
+
+# Redirects
+# https://tech.signavio.com/2017/managing-sphinx-redirects
+redirect_files = [
+    # old html, new html
+    ('array-overview.html', 'array.html'),
+    ('dataframe-overview.html', 'dataframe.html'),
+    ('delayed-overview.html', 'delayed.html'),
+]
+
+
+redirect_template = """\
+<html>
+  <head>
+    <meta http-equiv="refresh" content="1; url={new}" />
+    <script>
+      window.location.href = "{new}"
+    </script>
+  </head>
+</html>
+"""
+
+
+def copy_legacy_redirects(app, docname):
+    if app.builder.name == 'html':
+        for html_src_path, new in redirect_files:
+            page = redirect_template.format(new=new)
+            target_path = app.outdir + '/' + html_src_path
+            with open(target_path, 'w') as f:
+                f.write(page)
+
+
+def setup(app):
+    app.connect('build-finished', copy_legacy_redirects)
diff --git a/docs/source/dataframe-create.rst b/docs/source/dataframe-create.rst
index 759e4b0e6..f969b6674 100644
--- a/docs/source/dataframe-create.rst
+++ b/docs/source/dataframe-create.rst
@@ -22,7 +22,6 @@ file formats, and other Dask or Python collections.
 File Formats:
 
 .. autosummary::
-
     read_csv
     read_parquet
     read_hdf
diff --git a/docs/source/dataframe-overview.rst b/docs/source/dataframe-overview.rst
deleted file mode 100644
index f821a06f8..000000000
--- a/docs/source/dataframe-overview.rst
+++ /dev/null
@@ -1,162 +0,0 @@
-Overview
-========
-
-Dask Dataframe implements a subset of the Pandas Dataframe interface using
-blocked algorithms, cutting up the large DataFrame into many small Pandas
-DataFrames.  This lets us compute on dataframes that are larger than memory
-using all of our cores or on many dataframes spread across a cluster.  One
-operation on a dask.dataframe triggers many operations on the constituent
-Pandas dataframes.
-
-Design
-------
-
-.. image:: images/dask-dataframe.svg
-   :alt: Dask DataFrames coordinate many Pandas DataFrames
-   :align: right
-   :width: 40%
-
-Dask dataframes coordinate many Pandas DataFrames/Series arranged along the
-index.  Dask.dataframe is partitioned *row-wise*, grouping rows by index value
-for efficiency.  These Pandas objects may live on disk or on other machines.
-
-
-Common Uses and Anti-Uses
--------------------------
-
-Dask.dataframe is particularly useful in the following situations:
-
-*  Manipulating large datasets on a single machine, even when those datasets
-   don't fit comfortably into memory.
-*  Fast computation on large workstation machines by parallelizing many Pandas
-   calls across many cores.
-*  Distributed computing of very large tables stored in the Hadoop File System
-   (HDFS), S3, or other parallel file systems.
-*  Parallel groupby, join, or time series computations
-
-However in the following situations Dask.dataframe may not be the best choice:
-
-*  If your dataset fits comfortably into RAM on your laptop then you may be
-   better off just using Pandas_.  There may be simpler ways to improve
-   performance than through parallelism.
-*  If your dataset doesn't fit neatly into the Pandas tabular model then you
-   might find more use in :doc:`dask.bag <bag>` or :doc:`dask.array <array>`
-*  If you need functions that are not implemented in dask.dataframe then you
-   might want to look at :doc:`dask.delayed <delayed>` which offers more
-   flexibility.
-*  If you need a proper database with all that databases offer you might prefer
-   something like Postgres_
-
-.. _Pandas: https://pandas.pydata.org/
-.. _Postgres: https://www.postgresql.org/
-
-
-Dask.dataframe copies the pandas API
-------------------------------------
-
-Because the ``dask.dataframe`` application programming interface (API) is a
-subset of the pandas API it should be familiar to pandas users.  There are some
-slight alterations due to the parallel nature of dask:
-
-.. code-block:: python
-
-   >>> import dask.dataframe as dd
-   >>> df = dd.read_csv('2014-*.csv')
-   >>> df.head()
-      x  y
-   0  1  a
-   1  2  b
-   2  3  c
-   3  4  a
-   4  5  b
-   5  6  c
-
-   >>> df2 = df[df.y == 'a'].x + 1
-
-As with all dask collections (for example Array, Bag, DataFrame) one triggers
-computation by calling the ``.compute()`` method:
-
-.. code-block:: python
-
-   >>> df2.compute()
-   0    2
-   3    5
-   Name: x, dtype: int64
-
-
-Scope
------
-
-Dask.dataframe covers a small but well-used portion of the pandas API.
-This limitation is for two reasons:
-
-1.  The pandas API is *huge*
-2.  Some operations are genuinely hard to do in parallel (for example sort).
-
-Additionally, some important operations like ``set_index`` work, but are slower
-than in pandas because they may write out to disk.
-
-The following class of computations works well:
-
-* Trivially parallelizable operations (fast):
-    *  Elementwise operations:  ``df.x + df.y``, ``df * df``
-    *  Row-wise selections:  ``df[df.x > 0]``
-    *  Loc:  ``df.loc[4.0:10.5]``
-    *  Common aggregations:  ``df.x.max()``, ``df.max()``
-    *  Is in:  ``df[df.x.isin([1, 2, 3])]``
-    *  Datetime/string accessors:  ``df.timestamp.month``
-* Cleverly parallelizable operations (fast):
-    *  groupby-aggregate (with common aggregations): ``df.groupby(df.x).y.max()``,
-       ``df.groupby('x').max()``
-    *  groupby-apply on index: ``df.groupby(['idx', 'x']).apply(myfunc)``, where
-       ``idx`` is the index level name
-    *  value_counts:  ``df.x.value_counts()``
-    *  Drop duplicates:  ``df.x.drop_duplicates()``
-    *  Join on index:  ``dd.merge(df1, df2, left_index=True, right_index=True)``
-       or ``dd.merge(df1, df2, on=['idx', 'x'])`` where ``idx`` is the index
-       name for both ``df1`` and ``df2``
-    *  Join with Pandas DataFrames: ``dd.merge(df1, df2, on='id')``
-    *  Elementwise operations with different partitions / divisions: ``df1.x + df2.y``
-    *  Datetime resampling: ``df.resample(...)``
-    *  Rolling averages:  ``df.rolling(...)``
-    *  Pearson Correlations: ``df[['col1', 'col2']].corr()``
-* Operations requiring a shuffle (slow-ish, unless on index)
-    *  Set index:  ``df.set_index(df.x)``
-    *  groupby-apply not on index (with anything):  ``df.groupby(df.x).apply(myfunc)``
-    *  Join not on the index:  ``dd.merge(df1, df2, on='name')``
-
-See :doc:`DataFrame API documentation<dataframe-api>` for a more extensive
-list.
-
-
-Execution
----------
-
-By default ``dask.dataframe`` uses the multi-threaded scheduler.
-This exposes some parallelism when pandas or the underlying numpy operations
-release the global interpreter lock (GIL).  Generally pandas is more GIL
-bound than NumPy, so multi-core speed-ups are not as pronounced for
-``dask.dataframe`` as they are for ``dask.array``.  This is changing, and
-the pandas development team is actively working on releasing the GIL.
-
-In some cases you may experience speedups by switching to the multiprocessing
-or distributed scheduler.
-
-.. code-block:: python
-
-   >>> dask.set_options(get=dask.multiprocessing.get)
-
-See :doc:`scheduler docs<scheduler-overview>` for more information.
-
-
-Limitations
------------
-
-Dask.DataFrame does not implement the entire Pandas interface.  Users expecting this
-will be disappointed.  Notably, dask.dataframe has the following limitations:
-
-1.  Setting a new index from an unsorted column is expensive
-2.  Many operations, like groupby-apply and join on unsorted columns require
-    setting the index, which as mentioned above, is expensive
-3.  The Pandas API is very large.  Dask.dataframe does not attempt to implement
-    many pandas features or any of the more exotic data structures like NDFrames
diff --git a/docs/source/dataframe.rst b/docs/source/dataframe.rst
index 623ce8e9b..45d31d787 100644
--- a/docs/source/dataframe.rst
+++ b/docs/source/dataframe.rst
@@ -1,32 +1,155 @@
-DataFrame
+Dataframe
 =========
 
+.. toctree::
+   :maxdepth: 1
+   :hidden:
+
+   dataframe-api.rst
+   dataframe-create.rst
+   dataframe-performance.rst
+   dataframe-design.rst
+   dataframe-groupby.rst
+
 A Dask DataFrame is a large parallel dataframe composed of many smaller Pandas
 dataframes, split along the index.  These pandas dataframes may live on disk
 for larger-than-memory computing on a single machine, or on many different
-machines in a cluster.
+machines in a cluster.  One Dask dataframe operation triggers many operations
+on the constituent Pandas dataframes.
 
-Dask.dataframe implements a commonly used subset of the Pandas_ interface
-including elementwise operations, reductions, grouping operations, joins,
-timeseries algorithms, and more.  It copies the Pandas interface for these
-operations exactly and so should be very familiar to Pandas users.  Because
-Dask.dataframe operations merely coordinate Pandas operations they usually
-exhibit similar performance characteristics as are found in Pandas.
+Design
+------
 
-.. _Pandas: http://pandas.pydata.org/
+Dask dataframes coordinate many Pandas DataFrames/Series arranged along the
+index.  Dask.dataframe is partitioned *row-wise*, grouping rows by index value
+for efficiency.  These Pandas objects may live on disk or on other machines.
 
-.. toctree::
-   :maxdepth: 1
+.. image:: images/dask-dataframe.svg
+   :alt: Dask DataFrames coordinate many Pandas DataFrames
+   :width: 40%
 
-   dataframe-overview.rst
-   dataframe-create.rst
-   dataframe-api.rst
-   dataframe-performance.rst
 
-Other topics
+Dask.dataframe copies the Pandas API
+------------------------------------
 
-.. toctree::
-   :maxdepth: 1
+Because the ``dask.dataframe`` application programming interface (API) is a
+subset of the Pandas API it should be familiar to Pandas users.  There are some
+slight alterations due to the parallel nature of dask:
 
-   dataframe-design.rst
-   dataframe-groupby.rst
+.. code-block:: python
+
+   >>> import dask.dataframe as dd
+   >>> df = dd.read_csv('2014-*.csv')
+   >>> df.head()
+      x  y
+   0  1  a
+   1  2  b
+   2  3  c
+   3  4  a
+   4  5  b
+   5  6  c
+
+   >>> df2 = df[df.y == 'a'].x + 1
+
+As with all dask collections one triggers computation by calling the
+``.compute()`` method:
+
+.. code-block:: python
+
+   >>> df2.compute()
+   0    2
+   3    5
+   Name: x, dtype: int64
+
+
+Common Uses and Anti-Uses
+-------------------------
+
+Dask.dataframe is used in situations where Pandas is commonly needed, but when
+Pandas fails due to data size or computation speed.
+
+-  Manipulating large datasets, even when those datasets don't fit in memory
+-  Accelerating long computations by using many cores
+-  Distributed computing on large datasets with standard Pandas operations like
+   groupby, join, and time series computations
+
+Dask dataframe may not be the best choice in the following situations:
+
+*  If your dataset fits comfortably into RAM on your laptop then you may be
+   better off just using Pandas .  There may be simpler ways to improve
+   performance than through parallelism.
+*  If your dataset doesn't fit neatly into the Pandas tabular model then you
+   might find more use in :doc:`dask.bag <bag>` or :doc:`dask.array <array>`
+*  If you need functions that are not implemented in Dask dataframe then you
+   might want to look at :doc:`dask.delayed <delayed>` which offers more
+   flexibility.
+*  If you need a proper database with all that databases offer you might prefer
+   something like Postgres_
+
+.. _Pandas: https://pandas.pydata.org/
+.. _Postgres: https://www.postgresql.org/
+
+
+Scope
+-----
+
+Dask.dataframe covers a well-used portion of the Pandas API.
+The following class of computations works well:
+
+* Trivially parallelizable operations (fast):
+    *  Elementwise operations:  ``df.x + df.y``, ``df * df``
+    *  Row-wise selections:  ``df[df.x > 0]``
+    *  Loc:  ``df.loc[4.0:10.5]``
+    *  Common aggregations:  ``df.x.max()``, ``df.max()``
+    *  Is in:  ``df[df.x.isin([1, 2, 3])]``
+    *  Datetime/string accessors:  ``df.timestamp.month``
+* Cleverly parallelizable operations (fast):
+    *  groupby-aggregate (with common aggregations): ``df.groupby(df.x).y.max()``,
+       ``df.groupby('x').max()``
+    *  groupby-apply on index: ``df.groupby(['idx', 'x']).apply(myfunc)``, where
+       ``idx`` is the index level name
+    *  value_counts:  ``df.x.value_counts()``
+    *  Drop duplicates:  ``df.x.drop_duplicates()``
+    *  Join on index:  ``dd.merge(df1, df2, left_index=True, right_index=True)``
+       or ``dd.merge(df1, df2, on=['idx', 'x'])`` where ``idx`` is the index
+       name for both ``df1`` and ``df2``
+    *  Join with Pandas DataFrames: ``dd.merge(df1, df2, on='id')``
+    *  Elementwise operations with different partitions / divisions: ``df1.x + df2.y``
+    *  Datetime resampling: ``df.resample(...)``
+    *  Rolling averages:  ``df.rolling(...)``
+    *  Pearson Correlations: ``df[['col1', 'col2']].corr()``
+* Operations requiring a shuffle (slow-ish, unless on index)
+    *  Set index:  ``df.set_index(df.x)``
+    *  groupby-apply not on index (with anything):  ``df.groupby(df.x).apply(myfunc)``
+    *  Join not on the index:  ``dd.merge(df1, df2, on='name')``
+
+However Dask dataframe does not implement the entire Pandas interface.  Users
+expecting this will be disappointed.  Notably, Dask dataframe has the following
+limitations:
+
+1.  Setting a new index from an unsorted column is expensive
+2.  Many operations, like groupby-apply and join on unsorted columns require
+    setting the index, which as mentioned above, is expensive
+3.  The Pandas API is very large.  Dask dataframe does not attempt to implement
+    many Pandas features or any of the more exotic data structures like NDFrames
+4.  Operations that were slow on Pandas, like iterating through row-by-row,
+    remain slow on Dask dataframe
+
+See :doc:`DataFrame API documentation<dataframe-api>` for a more extensive list.
+
+
+Execution
+---------
+
+By default ``dask.dataframe`` uses the multi-threaded scheduler.
+This exposes some parallelism when Pandas or the underlying numpy operations
+release the global interpreter lock (GIL).  Generally Pandas is more GIL
+bound than NumPy, so multi-core speed-ups are not as pronounced for
+``dask.dataframe`` as they are for ``dask.array``.  This is changing, and
+the Pandas development team is actively working on releasing the GIL.
+
+When dealing with text data you may see speedups by switching to the newer
+:doc:`distributed scheduler <setup/single-distributed>` either on a cluster or
+single machine.
+
+.. _Pandas: http://pandas.pydata.org/
diff --git a/docs/source/delayed-api.rst b/docs/source/delayed-api.rst
index f322b420c..0b88bd71b 100644
--- a/docs/source/delayed-api.rst
+++ b/docs/source/delayed-api.rst
@@ -1,6 +1,43 @@
 API
 ===
 
+The ``dask.delayed`` interface consists of one function, ``delayed``:
+
+- ``delayed`` wraps functions
+
+   Wraps functions. Can be used as a decorator, or around function calls
+   directly (i.e. ``delayed(foo)(a, b, c)``). Outputs from functions wrapped in
+   ``delayed`` are proxy objects of type ``Delayed`` that contain a graph of
+   all operations done to get to this result.
+
+- ``delayed`` wraps objects
+
+   Wraps objects. Used to create ``Delayed`` proxies directly.
+
+``Delayed`` objects can be thought of as representing a key in the dask. A
+``Delayed`` supports *most* python operations, each of which creates another
+``Delayed`` representing the result:
+
+- Most operators (``*``, ``-``, and so on)
+- Item access and slicing (``a[0]``)
+- Attribute access (``a.size``)
+- Method calls (``a.index(0)``)
+
+Operations that aren't supported include:
+
+- Mutating operators (``a += 1``)
+- Mutating magics such as ``__setitem__``/``__setattr__`` (``a[0] = 1``, ``a.foo = 1``)
+- Iteration. (``for i in a: ...``)
+- Use as a predicate (``if a: ...``)
+
+The last two points in particular mean that ``Delayed`` objects cannot be used for
+control flow, meaning that no ``Delayed`` can appear in a loop or if statement.
+In other words you can't iterate over a ``Delayed`` object, or use it as part of
+a condition in an if statement, but ``Delayed`` object can be used in a body of a loop
+or if statement (i.e. the example above is fine, but if ``data`` was a ``Delayed``
+object it wouldn't be).
+Even with this limitation, many workflows can easily be parallelized.
+
 .. currentmodule:: dask.delayed
 
 .. autosummary::
diff --git a/docs/source/delayed-overview.rst b/docs/source/delayed-overview.rst
deleted file mode 100644
index 12bccf61a..000000000
--- a/docs/source/delayed-overview.rst
+++ /dev/null
@@ -1,119 +0,0 @@
-Overview
-========
-
-Motivation and Example
-----------------------
-
-Dask.delayed lets you parallelize custom code.  It is useful whenever your
-problem doesn't quite fit a high-level parallel object like dask.array or
-dask.dataframe but could still benefit from parallelism.  Dask.delayed works by
-delaying your function evaluations and putting them into a dask graph.
-Dask.delayed is useful when wrapping existing code or when handling
-non-standard problems.
-
-Consider the following example:
-
-.. code-block:: python
-
-    def inc(x):
-        return x + 1
-
-    def double(x):
-        return x + 2
-
-    def add(x, y):
-        return x + y
-
-    data = [1, 2, 3, 4, 5]
-
-    output = []
-    for x in data:
-        a = inc(x)
-        b = double(x)
-        c = add(a, b)
-        output.append(c)
-
-    total = sum(output)
-
-As written this code runs sequentially in a single thread.  However we see that
-a lot of this could be executed in parallel.  We use the ``delayed`` function
-to parallelize this code by turning it into a dask graph.  We slightly modify
-our code by wrapping functions in ``delayed``.  This delays the execution of
-the function and generates a dask graph instead.
-
-.. code-block:: python
-
-    from dask import delayed
-
-    output = []
-    for x in data:
-        a = delayed(inc)(x)
-        b = delayed(double)(x)
-        c = delayed(add)(a, b)
-        output.append(c)
-
-    total = delayed(sum)(output)
-
-We used the ``delayed`` function to wrap the function calls that we want
-to turn into tasks.  None of the ``inc``, ``double``, ``add`` or ``sum`` calls
-have happened yet, instead the object ``total`` is a ``Delayed`` result that
-contains a task graph of the entire computation.  Looking at the graph we see
-clear opportunities for parallel execution.  The dask schedulers will exploit
-this parallelism, generally improving performance.  (although not in this
-example, because these functions are already very small and fast.)
-
-.. code-block:: python
-
-   total.visualize()  # see image to the right
-
-.. image:: images/delayed-inc-double-add.svg
-   :align: right
-   :alt: simple task graph created with dask.delayed
-
-We can now compute this lazy result to execute the graph in parallel:
-
-.. code-block:: python
-
-   >>> total.compute()
-   45
-
-
-Delayed Function
-----------------
-
-The ``dask.delayed`` interface consists of one function, ``delayed``:
-
-- ``delayed`` wraps functions
-
-   Wraps functions. Can be used as a decorator, or around function calls
-   directly (i.e. ``delayed(foo)(a, b, c)``). Outputs from functions wrapped in
-   ``delayed`` are proxy objects of type ``Delayed`` that contain a graph of
-   all operations done to get to this result.
-
-- ``delayed`` wraps objects
-
-   Wraps objects. Used to create ``Delayed`` proxies directly.
-
-``Delayed`` objects can be thought of as representing a key in the dask. A
-``Delayed`` supports *most* python operations, each of which creates another
-``Delayed`` representing the result:
-
-- Most operators (``*``, ``-``, and so on)
-- Item access and slicing (``a[0]``)
-- Attribute access (``a.size``)
-- Method calls (``a.index(0)``)
-
-Operations that aren't supported include:
-
-- Mutating operators (``a += 1``)
-- Mutating magics such as ``__setitem__``/``__setattr__`` (``a[0] = 1``, ``a.foo = 1``)
-- Iteration. (``for i in a: ...``)
-- Use as a predicate (``if a: ...``)
-
-The last two points in particular mean that ``Delayed`` objects cannot be used for 
-control flow, meaning that no ``Delayed`` can appear in a loop or if statement. 
-In other words you can't iterate over a ``Delayed`` object, or use it as part of
-a condition in an if statement, but ``Delayed`` object can be used in a body of a loop 
-or if statement (i.e. the example above is fine, but if ``data`` was a ``Delayed`` 
-object it wouldn't be).
-Even with this limitation, many workflows can easily be parallelized.
diff --git a/docs/source/delayed.rst b/docs/source/delayed.rst
index c7190893b..ec098cf8a 100644
--- a/docs/source/delayed.rst
+++ b/docs/source/delayed.rst
@@ -1,6 +1,13 @@
 Delayed
 =======
 
+.. toctree::
+   :maxdepth: 1
+   :hidden:
+
+   delayed-api.rst
+   delayed-collections.rst
+
 Sometimes problems don't fit into one of the collections like ``dask.array`` or
 ``dask.dataframe``. In these cases, users can parallelize custom algorithms
 using the simpler ``dask.delayed`` interface. This allows one to create graphs
@@ -18,9 +25,125 @@ directly with a light annotation of normal python code.
 .. image:: images/inc-add.svg
    :alt: simple task graph created with dask.delayed
 
-.. toctree::
-   :maxdepth: 1
+Example
+-------
 
-   delayed-overview.rst
-   delayed-api.rst
-   delayed-collections.rst
+Sometimes we face problems that are parallelizable, but don't fit high-level
+abstractions Dask array or Dask dataframe.  Consider the following example:
+
+.. code-block:: python
+
+    def inc(x):
+        return x + 1
+
+    def double(x):
+        return x + 2
+
+    def add(x, y):
+        return x + y
+
+    data = [1, 2, 3, 4, 5]
+
+    output = []
+    for x in data:
+        a = inc(x)
+        b = double(x)
+        c = add(a, b)
+        output.append(c)
+
+    total = sum(output)
+
+There is clearly parallelism in this problem (many of the ``inc`` and
+``double`` and ``add`` functions can evaluate independently), but it's not
+clear how to convert this to a big array or big dataframe computation.
+
+As written this code runs sequentially in a single thread.  However we see that
+a lot of this could be executed in parallel.
+
+The Dask ``delayed`` function decorates your functions so that they operate
+*lazily*.  Rather than executing your function immediately it will defer
+execution, placing the function and its arguments into a task graph.
+
+.. currentmodule:: dask.delayed
+
+.. autosummary::
+    delayed
+
+We slightly modify our code our code by wrapping functions in ``delayed``.
+This delays the execution of the function and generates a dask graph instead.
+
+.. code-block:: python
+
+    import dask
+
+    output = []
+    for x in data:
+        a = dask.delayed(inc)(x)
+        b = dask.delayed(double)(x)
+        c = dask.delayed(add)(a, b)
+        output.append(c)
+
+    total = dask.delayed(sum)(output)
+
+We used the ``dask.delayed`` function to wrap the function calls that we want
+to turn into tasks.  None of the ``inc``, ``double``, ``add`` or ``sum`` calls
+have happened yet, instead the object ``total`` is a ``Delayed`` result that
+contains a task graph of the entire computation.  Looking at the graph we see
+clear opportunities for parallel execution.  The dask schedulers will exploit
+this parallelism, generally improving performance.  (although not in this
+example, because these functions are already very small and fast.)
+
+.. code-block:: python
+
+   total.visualize()  # see image to the right
+
+.. image:: images/delayed-inc-double-add.svg
+   :align: right
+   :alt: simple task graph created with dask.delayed
+
+We can now compute this lazy result to execute the graph in parallel:
+
+.. code-block:: python
+
+   >>> total.compute()
+   45
+
+Decorator
+---------
+
+It is also common to see the delayed function used as a decorator.  Here is a
+reproduction of our original problem as a parallel code.
+
+.. code-block:: python
+
+    import dask
+
+    @dask.delayed
+    def inc(x):
+        return x + 1
+
+    @dask.delayed
+    def double(x):
+        return x + 2
+
+    @dask.delayed
+    def add(x, y):
+        return x + y
+
+    data = [1, 2, 3, 4, 5]
+
+    output = []
+    for x in data:
+        a = inc(x)
+        b = double(x)
+        c = add(a, b)
+        output.append(c)
+
+    total = dask.delayed(sum)(output)
+
+
+Real time
+---------
+
+Sometimes you want to create and destroy work during execution, launch tasks
+from other tasks, etc..  For this, see the :doc:`Futures <futures>` interface.
diff --git a/docs/source/futures.rst b/docs/source/futures.rst
index 718a3b131..8a5ecfb95 100644
--- a/docs/source/futures.rst
+++ b/docs/source/futures.rst
@@ -316,7 +316,9 @@ Submit Tasks from Tasks
 -----------------------
 
 .. autosummary::
+   compute
    get_client
+   rejoin
    secede
 
 Tasks can launch other tasks by getting their own client.  This enables complex
@@ -372,6 +374,41 @@ thread that does not take up a slot within the Dask worker:
           future = client.submit(process, data)
           fire_and_forget(future)
 
+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.
+
+.. code-block:: python
+
+   def f(n):
+      client = get_client()
+
+      secede()  # secede while we wait for results to come back
+      futures = client.map(func, range(n))
+      results = client.gather(futures)
+
+      rejoin()  # block until a slot is open in the thread pool
+      result = analyze(results)
+      return result
+
+
+Alternatively, you can just use the normal ``dask.compute`` function *within* a
+task.  This will automatically call ``secede`` and ``rejoin`` appropriately.
+
+.. code-block:: python
+
+   def f(name, fn):
+       df = dd.read_csv(fn)  # note that this is a dask collection
+       result = df[df.name == name].count()
+
+       # This calls secede
+       # Then runs the computation on the cluster (including this worker)
+       # Then blocks on rejoin, and finally delivers the answer
+       result = result.compute()
+
+       return result
+
 
 Coordinate Data Between Clients
 -------------------------------
@@ -449,6 +486,33 @@ If you want to share large pieces of information then scatter the data first
    >>> future = client.scatter(parameters)
    >>> var.set(future)
 
+
+Locks
+-----
+
+.. autosummary::
+   Lock
+
+You can also hold onto cluster-wide locks using the ``Lock`` object.
+This lock can either be given a consistent name, or you can pass the lock
+object around itself.
+
+.. code-block:: python
+
+   from dask.distributed import Lock
+   lock = Lock()
+
+   def load(fn, lock=None):
+       with lock:
+           # read data from filename using some sensitive source
+           return ...
+
+   futures = client.map(load, filenames, lock=lock)
+
+This can be useful if you want to control concurrent access to some external
+resource like a database or un-thread-safe library.
+
+
 API
 ---
 
@@ -523,3 +587,6 @@ API
 
 .. autoclass:: Variable
    :members:
+
+.. autoclass:: Lock
+   :members:
diff --git a/docs/source/remote-data-services.rst b/docs/source/remote-data-services.rst
index f5f9a9caa..8fe0f913a 100644
--- a/docs/source/remote-data-services.rst
+++ b/docs/source/remote-data-services.rst
@@ -224,6 +224,23 @@ Possible additional storage options:
       a JSON file created by gcloud.
 
 
+HTTP
+----
+
+Direct file-like access to arbitrary URLs is available over HTTP and HTTPS. However,
+there is no such thing as ``glob`` functionality over HTTP, so only explicit lists
+of files can be used.
+
+Server implementations differ in the information they provide - they may or may
+not specify the size of a file via a HEAD request or at the start of a download -
+and some servers may not respect bytes range requests. The HTTPFileSystem therefore
+offers best-effort behaviour: the download is streamed, but if more data is seen
+than the configured block-size, an error will be raised. To be able to access such
+data, you must read the whole file in one shot (and it must fit in memory).
+
+Note that, currently, ``http://`` and ``https://`` are treated as separate protocols,
+and cannot be mixed.
+
 Developer API
 ~~~~~~~~~~~~~
 


Test patch:
diff --git a/dask/bytes/tests/test_bytes_utils.py b/dask/bytes/tests/test_bytes_utils.py
index 1972e0b16..aed29a6a9 100644
--- a/dask/bytes/tests/test_bytes_utils.py
+++ b/dask/bytes/tests/test_bytes_utils.py
@@ -91,7 +91,8 @@ def test_infer_storage_options():
     assert so.pop('username') == 'User-name'
     assert so.pop('host') == 'Node-name.com'
 
-    assert infer_storage_options('http://127.0.0.1:8080/test.csv')['host'] == '127.0.0.1'
+    u = 'http://127.0.0.1:8080/test.csv'
+    assert infer_storage_options(u) == {'protocol': 'http', 'path': u}
 
     # For s3 and gcs the netloc is actually the bucket name, so we want to
     # include it in the path. Test that:
diff --git a/dask/bytes/tests/test_http.py b/dask/bytes/tests/test_http.py
new file mode 100644
index 000000000..ce877c524
--- /dev/null
+++ b/dask/bytes/tests/test_http.py
@@ -0,0 +1,135 @@
+import os
+import pytest
+import requests
+import subprocess
+import time
+
+from dask.bytes.core import open_files
+from dask.compatibility import PY2
+from dask.utils import tmpdir
+
+files = ['a', 'b']
+
+
+@pytest.fixture(scope='module')
+def dir_server():
+    with tmpdir() as d:
+        for fn in files:
+            with open(os.path.join(d, fn), 'wb') as f:
+                f.write(b'a' * 10000)
+
+        if PY2:
+            cmd = ['python', '-m', 'SimpleHTTPServer', '8999']
+        else:
+            cmd = ['python', '-m', 'http.server', '8999']
+        p = subprocess.Popen(cmd, cwd=d)
+        timeout = 10
+        while True:
+            try:
+                requests.get('http://localhost:8999')
+                break
+            except requests.exceptions.ConnectionError:
+                time.sleep(0.1)
+                timeout -= 0.1
+                if timeout < 0:
+                    raise RuntimeError('Server did not appear')
+        yield d
+        p.terminate()
+
+
+def test_simple(dir_server):
+    root = 'http://localhost:8999/'
+    fn = files[0]
+    f = open_files(root + fn)[0]
+    with f as f:
+        data = f.read()
+    assert data == open(os.path.join(dir_server, fn), 'rb').read()
+
+
+@pytest.mark.parametrize('block_size', [None, 99999])
+def test_ops(dir_server, block_size):
+    root = 'http://localhost:8999/'
+    fn = files[0]
+    f = open_files(root + fn)[0]
+    data = open(os.path.join(dir_server, fn), 'rb').read()
+    with f as f:
+        # these pass because the default
+        assert f.read(10) == data[:10]
+        f.seek(0)
+        assert f.read(10) == data[:10]
+        assert f.read(10) == data[10:20]
+        f.seek(-10, 2)
+        assert f.read() == data[-10:]
+
+
+def test_ops_blocksize(dir_server):
+    root = 'http://localhost:8999/'
+    fn = files[0]
+    f = open_files(root + fn, block_size=2)[0]
+    data = open(os.path.join(dir_server, fn), 'rb').read()
+    with f as f:
+        # it's OK to read the whole file
+        assert f.read() == data
+
+    # note that if we reuse f from above, because it is tokenized, we get
+    # the same open file - where is this cached?
+    fn = files[1]
+    f = open_files(root + fn, block_size=2)[0]
+    with f as f:
+        # fails becasue we want only 12 bytes
+        with pytest.raises(ValueError):
+            assert f.read(10) == data[:10]
+
+
+def test_errors(dir_server):
+    f = open_files('http://localhost:8999/doesnotexist')[0]
+    with pytest.raises(requests.exceptions.RequestException):
+        with f:
+            pass
+    f = open_files('http://nohost/')[0]
+    with pytest.raises(requests.exceptions.RequestException):
+        with f:
+            pass
+    root = 'http://localhost:8999/'
+    fn = files[0]
+    f = open_files(root + fn, mode='wb')[0]
+    with pytest.raises(NotImplementedError):
+        with f:
+            pass
+    f = open_files(root + fn)[0]
+    with f as f:
+        with pytest.raises(ValueError):
+            f.seek(-1)
+
+
+def test_files(dir_server):
+    root = 'http://localhost:8999/'
+    fs = open_files([root + f for f in files])
+    for f, f2 in zip(fs, files):
+        with f as f:
+            assert f.read() == open(os.path.join(dir_server, f2), 'rb').read()
+
+
+@pytest.mark.network
+def test_parquet():
+    dd = pytest.importorskip('dask.dataframe')
+    pytest.importorskip('fastparquet')  # no pyarrow compatability FS yet
+    df = dd.read_parquet([
+        'https://github.com/Parquet/parquet-compatibility/raw/'
+        'master/parquet-testdata/impala/1.1.1-NONE/'
+        'nation.impala.parquet']).compute()
+    assert df.n_nationkey.tolist() == list(range(25))
+    assert df.columns.tolist() == ['n_nationkey', 'n_name', 'n_regionkey',
+                                   'n_comment']
+
+
+@pytest.mark.network
+def test_bag():
+    # This test pulls from different hosts
+    db = pytest.importorskip('dask.bag')
+    urls = ['https://raw.githubusercontent.com/weierophinney/pastebin/'
+            'master/public/js-src/dojox/data/tests/stores/patterns.csv',
+            'https://en.wikipedia.org']
+    b = db.read_text(urls)
+    assert b.npartitions == 2
+    b.compute()