Allow P2P to store data in-memory

[hendrikmakait]

• Supersedes [RFC]: Allow in-memory buffering of p2p shuffle results #7618
• Includes and blocked by Allow ResourceLimiter to be unlimited #8276

• Tests added / passed
• Passes pre-commit run --all-files

[github-actions]

Unit Test Results

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

       27 files  +       1         27 suites  +1   15h 49m 14s ⏱️ + 1h 8m 8s
  3 932 tests +     56    3 811 ✔️ +     58     115 💤 ±  0    6 ❌  - 2 
49 387 runs  +2 020  47 017 ✔️ +1 956  2 340 💤 +62  30 ❌ +2 

For more details on these failures, see this check.

Results for commit 255e477. ± Comparison against base commit a8e5dab.

This pull request removes 51 and adds 107 tests. Note that renamed tests count towards both.

distributed.shuffle.tests.test_merge ‑ test_merge[all-inner]
distributed.shuffle.tests.test_merge ‑ test_merge[all-left]
distributed.shuffle.tests.test_merge ‑ test_merge[all-outer]
distributed.shuffle.tests.test_merge ‑ test_merge[all-right]
distributed.shuffle.tests.test_merge ‑ test_merge[none-inner]
distributed.shuffle.tests.test_merge ‑ test_merge[none-left]
distributed.shuffle.tests.test_merge ‑ test_merge[none-outer]
distributed.shuffle.tests.test_merge ‑ test_merge[none-right]
distributed.shuffle.tests.test_merge ‑ test_merge[some-inner]
distributed.shuffle.tests.test_merge ‑ test_merge[some-left]
…

distributed.shuffle.tests.test_memory_buffer ‑ test_basic
distributed.shuffle.tests.test_memory_buffer ‑ test_many[1000]
distributed.shuffle.tests.test_memory_buffer ‑ test_many[100]
distributed.shuffle.tests.test_memory_buffer ‑ test_many[2]
distributed.shuffle.tests.test_memory_buffer ‑ test_read_before_flush
distributed.shuffle.tests.test_merge ‑ test_merge[all-False-inner]
distributed.shuffle.tests.test_merge ‑ test_merge[all-False-left]
distributed.shuffle.tests.test_merge ‑ test_merge[all-False-outer]
distributed.shuffle.tests.test_merge ‑ test_merge[all-False-right]
distributed.shuffle.tests.test_merge ‑ test_merge[all-True-inner]
…

♻️ This comment has been updated with latest results.

[hendrikmakait]

This can be refactored to avoid most of the baggage it inherits from ShardsBuffer. I'd do this in a follow-up since I expect us to rework buffers anyhow and don't want to spend time on API compatibility with the DiskShardsBuffer until then.

[hendrikmakait]

This is ready for review; I'm running some A/B tests to see the impact.

[hendrikmakait]

A/B test results on modified tests:

• Some are up to 40% faster.
• Some are unaffected.
• Some failed to run because the workers ran OOM.

[mrocklin]

Some failed to run because the workers ran OOM

This seems bad to me

I've been playing with TPC-H recently and have been really appreciating how Dask is able to keep running even when other systems break down because we didn't have as much memory as we thought we did.

I'd encourage you to generate scale-100 on your personal computer and then run local TPC-H and see what happens with this PR. My guess is that it isn't immediately obvious how to do this. I should write up instructions probably.

[hendrikmakait]

This seems bad to me

This is the first iteration of diskless shuffling. It is explicitly "opt-in" by requiring you to set a config option. Falling back to disk in case a worker unexpectedly runs out of memory has always been considered out of scope for this iteration.

[mrocklin]

Ah, sorry. Thanks for the clarification.

…

On Thu, Oct 19, 2023 at 9:57 AM Hendrik Makait ***@***.***> wrote: This seems bad to me This is the first iteration of diskless shuffling. It is explicitly "opt-in" by requiring you to set a config option. Falling back to disk in case a worker unexpectedly runs out of memory has always been considered out of scope for this iteration. — Reply to this email directly, view it on GitHub <#8279 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTD6IBI2WW2TRFHHGYTYAE5U5AVCNFSM6AAAAAA6E7IYX6VHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMYTONZRGE3DINBRGM> . You are receiving this because you commented.Message ID: ***@***.***>