Don't pile up context_meter callbacks

[crusaderky]

• Closes Metrics introduce non-trivial overhead #7949

Reduce overhead of fine performance metrics when there is a very long chain of task finished -> task started events on the worker.

from time import time
import dask
from distributed import Client, wait

dask.config.set({"distributed.scheduler.worker-saturation": float("inf")})
c = Client(n_workers=2, threads_per_worker=2)

def inc(x):
    return x + 1

t0 = time()
N = 10_000
futs = c.map(inc, range(N))
wait(futs)
t1 = time()
print((t1 - t0) / N * c.cluster.scheduler.total_nthreads)

release	time per task
2023.3.1	1.2ms
2023.3.2	6.5ms
2023.6.1	8.1ms
this PR	1.5ms

Note that real-life impact is a lot more modest than what the benchmark would suggest, as you would only experience noticeable degradation when

1. scheduler-side queueing is turned off, or
2. there's a very large amount of non-rootish tasks all ready at the same time

neither of which should be common use cases.

[crusaderky]

As an aside, this whole incident would have been prevented by

• [WIP] Asynchronous SpillBuffer #7686

[github-actions]

Unit Test Results

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

       20 files  ±    0         20 suites  ±0   10h 39m 44s ⏱️ - 50m 50s
  3 701 tests +    3    3 591 ✔️ +    4     106 💤 ±  0  4 ❌  - 1 
34 974 runs   - 792  33 262 ✔️  - 749  1 708 💤  - 40  4 ❌  - 3 

For more details on these failures, see this check.

Results for commit 6a595fe. ± Comparison against base commit 9b9f948.

[fjetter]

scheduler-side queueing is turned off, or
there's a very large amount of non-rootish tasks all ready at the same time
neither of which should be common use cases.

Well, this does affect real world problems as well regardless of root-ish settings. For instance, if you add a trivial dependency, nothing is classified as root-ish anymore and you are in the same worst-case situation, e.g.

def inc(x, foo):
    return x + 1

t0 = time()
N = 10_000
foo = c.submit(inc, 0, foo=0, pure=False)
futs = c.map(inc, range(N), foo=foo)

(Note: this problem is also suffering a different problem, namely the tasks are all assigned to the same worker. I'm working on this problem in parallel but it also shows the executor overhead)

---

I'm a bit disappointed that this was not detected by our benchmarks suite. I guess our examples are too simple. would you mind adding a parametrized version of test_large_map that disables queuing. This would've picked up this regression early on.

[crusaderky]

For instance, if you add a trivial dependency, nothing is classified as root-ish anymore and you are in the same worst-case situation, e.g.

def inc(x, foo):
    return x + 1

t0 = time()
N = 10_000
foo = c.submit(inc, 0, foo=0, pure=False)
futs = c.map(inc, range(N), foo=foo)

You are hitting an edge case, where you should have two TaskGroups but you end up with only one because the dependency and the dependents accidentally share the same prefix AND the keys are not tuples. This causes the TaskGroup to have itself in its dependencies, which in turns causes is_rootish to erroneously believe that all 10,001 tasks in the group have 10,001 dependencies.

If you were using dask.array, dask.dataframe, or dask.bag OR the dependency was using a different function, you'd have two different TaskGroups and all the tasks would be rootish (because there are less than 5 dependencies).

We could fix this specific use case by changing Client.map to generate tuple keys.

I guess our examples are too simple. would you mind adding a parametrized version of test_large_map that disables queuing. This would've picked up this regression early on.

I don't think it's a good idea. Queuing impacts way, way more than just this specific issue, so if we actually care the whole test suite should run with and without queuing. I believe there was implicit consensus that that would be overkill as users shouldn't tamper with that setting and it's only there to let users revert the change in a hurry in case we realize that we did a catastrophic design mistake that hamstrings a specific use case.

I'm a bit disappointed that this was not detected by our benchmarks suite.

I disagree. I think that the benchmarks suite shows that this is a negligible problem if you don't go digging for edge cases (the big assumption here is that worker-saturation: inf is an edge case, and again if it's not meant to be then the whole test suite should run twice).

[fjetter]

If you were using dask.array, dask.dataframe, or dask.bag OR the dependency was using a different function, you'd have two different TaskGroups and all the tasks would be rootish (because there are less than 5 dependencies).

No point in arguing about this. This is a problem, it needs to be fixed. It is very possible to pile up tasks on workers, even with queuing.

I don't think it's a good idea. Queuing impacts way, way more than just this specific issue, so if we actually care the whole test suite should run with and without queuing. I believe there was implicit consensus that that would be overkill as users shouldn't tamper with that setting and it's only there to let users revert the change in a hurry in case we realize that we did a catastrophic design mistake that hamstrings a specific use case.

I'm not willing to invest the time and money to run the entire test suite with this being parametrized. This specific test was added to measure task latency and it missed this regression. Disabling this config reinstates this measurement.