• Partially closes Scheduler gather should warn or abort requests if data is too large #7964

This check is somewhat cruder than what was described in the issue above.

I tried setting up a system that graciously copes with heterogeneous clusters where a few workers mount much more memory than the rest (a somewhat common design). However

1. it was resulting in unnecessarily complicated and fragile implementation (it would add new restrictions to tasks, which in turn would flip rootish tasks to non-rootish), and
2. it didn't escape the fact that today dask/distributed has no clue about the output size of a task (whereas dask/dask does). In other words, the scheduler can't know in advance if the task is

• a reduction that requires sum(dts.nbytes for dts in ts.dependencies) heap memory, or
• a concatenation that requires twice as much, or
• something else that requires arbitrary amounts of temporary memory (possibly oom-larger than the inputs).

So in the end I opted for a crude order-of-magnitude check and, in the case of heterogenous clusters, explicitly expect the user to use worker or resource constraints.

Note

This catches client.gather(client.compute(collection)), but not collection.compute() or dask.compute(collection), as only the first one has a finalizer; the other two fetch the individual chunks directly from the original workers to the client.