Auto-fail tasks with deps larger than the worker memory

[crusaderky]

• 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.

[github-actions]

Unit Test Results

See test report for an extended history of previous test failures. This is useful for diagnosing flaky tests.

       21 files  +       17         21 suites  +17   10h 49m 29s ⏱️ + 9h 55m 16s
  3 795 tests +     415    3 688 ✔️ +     519     107 💤  -    104  0 ❌ ±0 
36 687 runs  +31 420  34 888 ✔️ +30 068  1 799 💤 +1 352  0 ❌ ±0 

Results for commit f3881d2. ± Comparison against base commit ef59142.

♻️ This comment has been updated with latest results.

[fjetter]

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

This is only true for array workloads atm