[Serve][Core] Fix serve_long_running memory leak by fixing GCS pubsub asyncio task leak

[architkulkarni]

Why are these changes needed?

Debugged with @simon-mo and @scv119 . The Serve long running test was failing due to a memory leak in dashboard.py. The root cause was in the GCS pubsub code, with the _close: asyncio.Event object adding millions of waits every few minutes without the waits ever being killed, causing the _close._waiters queue to grow without bound. The root cause is when awaiting with FIRST_COMPLETED, the caller is responsible for killing the unfinished task.

This PR:

• Fixes the memory leak by canceling the close task if it wasn't done. (Contributed by @simon-mo)

This PR also adds some side improvements to the release test:

• Use lower-memory instances so that memory leaks aren't hidden by the instances having a lot of available memory
• Gracefully handle the case where the wrk fails, which previously caused the release test output to be overwritten in a tight loop, which led to a hard-to-interpret errors being surfaced to the release test infrastructure
• Use different ports for the dashboard agents on the multiple cluster_utils virtual nodes to prevent port conflict

Related issue number

Closes #28977
May address #26568

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[architkulkarni]

@simon-mo Do you mind editing the PR description if the description of the memory issue is inaccurate?

[architkulkarni]

@scv119 @iycheng Is it feasible to add a unit test for the memory issue?

[scv119]

great findings! I'll let @iycheng to help the unit test part. one possible way is to open up the _close()._waiters and ensure we have O(1) instead of O(n) number of waiters.

[simon-mo]

One thing you can do is to check asyncio.Task.all_tasks() is not growing indefinitely

[architkulkarni]

At @sihanwang41 's suggestion, I tested it out again, but this time without the change in the first two lines where we removed ensure_future, and the leak was still fixed. @simon-mo what's the benefit of changing ensure_future to create_task()?

[simon-mo]

future cannot be cancelled, task can. (this is the API in Python 3.6 if i recall...)... also ensure_future in Python 3.6 creates a task anyway, which is a confusing API.

[architkulkarni]

Ah I see, I was testing on 3.8 so I didn't run into that

[architkulkarni]

One thing you can do is to check asyncio.Task.all_tasks() is not growing indefinitely

Thanks, added this as a unit test

[architkulkarni]

Long running test passed

[architkulkarni]

windows runtime env tests flaky on master
windows serve:test_api was broken on master
windows serve:tutorial_rllib was broken on master
windows test_args was broken on master
windows test_multi_node was broken on master
test_dashbaord was broken on master
test_client flaky on master
rllib cartpole tests broken on master

Java test failed due to randomly chosen port conflict, but I couldn't find evidence of flakiness on the tracker. Restarting the Java test to be safe

[architkulkarni]

Java passed on retry