Shuffle metrics 2/4: Add background metrics

[crusaderky]

• Closes Use metering for P2P shuffling instrumentation #7943

Please read: #7943 (comment)
There are two commits in this PR. All but the last are the previous PRs in the chain.

[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  ±  0         27 suites  ±0   11h 40m 34s ⏱️ + 2m 18s
  3 947 tests +  3    3 835 ✔️ ±  0     110 💤 +1  2 ❌ +2 
49 649 runs  +27  47 346 ✔️ +21  2 301 💤 +4  2 ❌ +2 

For more details on these failures, see this check.

Results for commit faa5bd7. ± Comparison against base commit 4d41d32.

♻️ This comment has been updated with latest results.

[fjetter]

pretty sure this is a SyntaxError since the break is outside a loop

[fjetter]

I'm a bit confused how this isn't lighting up CI

[crusaderky]

because it breaks the inner while size < self.max_message_size:, not the outer one

[fjetter]

I imagine this list call could have a non-trivial impact on performance since it forces the entire list of shards to be serialized and the serialized bytes to be retained before anything is written to disk, doesn't it?

[crusaderky]

In theory, yes.
In practice, no, because it uses pickle buffers. There shouldn't be any use cases where the buffer will be temporarily deep-copied (as we do not cater for object dtypes). So the only thing that will be temporarily kept in memory is the pickle metadata, which is always trivial (~120 bytes per shard, while the worst use case we currently care to monitor is 8 kiB worth of buffer per shard).

This said, I reworked the code to remove the list, so that we avoid any and all doubts.

[crusaderky]

I suggest to revert this. I'm hesitant to factor out a while loop body with loop control statements into a function. This is a non-trivial refactoring and objectively not necessarily simpler to understand

reverted

[hendrikmakait]

Thanks, @crusaderky!