Currently with Distributed we have Futures in that context. However there doesn't exist a first class Future object within Dask itself. It would be useful to have a first class Future object within Dask. This would enable a wider set of algorithms and provided some needed flexibility for various problems people run into today.

First having a Future object would be useful in cases where we want to submit some work to start and then construct later operations with the knowledge that these results will soon become available (one example might be writing these to disk). Second this can be a useful way to prioritize elements of a workflow where we know what we want to prioritize (like working with large chunks of memory first). Third can help in cases where we want to interleave work between subtasks (recentering, normalizing, binning, etc.). Fourth this can be useful when the results of tasks are needed to more informatively generate new work (working with unknown chunks, selection based on data values, etc.). Fifth this can be a useful way of cutting down on large task graph handling issues (this is something other distributed libraries use #3514 (comment) ). Sixth use cases that involve some form of iteration where the intermediate result needs to be checked to see if the task is done. There may be more ways this functionality could be used, but hopefully this provides some ideas on where/how it might help.

Additionally have gone through the issue tracker and identified issues/cases where this would either solve or help solve the underlying problem. This is not necessarily comprehensive. Also some issues may require more work on top of this. However in all cases this seems to provide a meaningful base to build on. Here's the list:

• Scheduler fail case: centering data with dask.array #874
• with_dependencies method for adding arbitrary dependencies #1519
• ENH: Delayed variant of persist (pin?)  #2156 (at least practically)
• Multiprocess writes using to_hdf5 #3074 (useful for)
• Safely pass a file-like object to dask store #3255 (useful for)
• Very large dask arrays #3514
• Convention for automatic rechunking #3587 (comment)
• compute on sliced array wastes memory #3595 (comment)
• Handling Dask Arrays with vindex #4156 (useful for)
• Allow UDF style array storage with store-like function #4149 (useful for)
• Handling da.histogram when range is not specified #4544

Note this likely (though happy to be proven wrong) depends on moving to concurrent.futures as documented in issue ( #6220 ). May also need secede from PR ( dask/distributed#471 ) and rejoin from PR ( dask/distributed#1462 ).