Make BatchedSend restartable

[gjoseph92]

This is #6272, with my PR comments addressed.

Closes #5481, #6228, #5480

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

[gjoseph92]

The changes from fb8f1c9 are a driveby (@graingert and I discussed how the current logic is messed up in potentially allowing the wrong client batchedsend to be deleted and closed). I'm happy to pull them out to a separate PR.

[gjoseph92]

I didn't do anything about my concern in #6272 (comment) because I don't know if it's valid or not

[github-actions]

Unit Test Results

       15 files   -        1         15 suites   - 1   7h 5m 29s ⏱️ - 18m 38s
  2 797 tests +     26    2 710 ✔️ +     17    79 💤 +    1    8 ❌ +  8 
20 743 runs   - 1 387  19 806 ✔️  - 1 303  910 💤  - 111  27 ❌ +27 

For more details on these failures, see this check.

Results for commit d49d5ca. ± Comparison against base commit 9f02e7a.

♻️ This comment has been updated with latest results.

[gjoseph92]

This is currently plagued by a few reliably failing tests:

distributed/tests/test_client.py::test_retire_workers
distributed/tests/test_scheduler.py::test_retire_workers_close
distributed/tests/test_client.py::test_reconnect_timeout
distributed/tests/test_tls_functional.py::test_retire_workers
distributed/tests/test_stress.py::test_stress_scatter_death

I can't reproduce these failures locally. Additionally, when I restrict CI to only run these tests, they pass: https://github.com/dask/distributed/runs/6428523276?check_suite_focus=true

Interestingly, test_retire_workers seems to be leaking file descriptors:

================================ RESOURCE LEAKS ================================
distributed/tests/test_client.py::test_retire_workers leaking fds: leaked 3 file descriptor(s) (16->19)
distributed/tests/test_tls_functional.py::test_retire_workers leaking fds: leaked 1 file descriptor(s) (19->20)

[gjoseph92]

Update: the retire_workers tests do fail on Windows, even when run in isolation. test_reconnect_timeout (which is the one that fails most reliably normally) is still not failing in isolation.

[gjoseph92]

Thanks to 40e37fa, we can now at least see that the asyncio.Task being leaked from test_reconnect_timeout is coming from the Client: https://github.com/dask/distributed/runs/6429598771?check_suite_focus=true#step:11:1324

[fjetter]

This should be no longer relevant after #6361