[Core] Memory leak when using asyncio

What happened + What you expected to happen

When running the script below, the memory utilization of the Supervisor actor increases slowly over time. This usually leads to OOM for more complex applications.

This memory increase does not happen when we use worker_task = asyncio.wrap_future(worker.run.remote().future())instead of worker_task = worker.run.remote()

Versions / Dependencies

Python 3.8.19
Ray 2.10.0
Numpy 1.24.4

Reproduction script

import asyncio
import numpy as np
import ray
from ray.util.queue import Queue

@ray.remote(num_cpus=1)
class Worker:
    def __init__(self, queue):
        self.queue = queue

    async def run(self):
        for i in range(1000):
            status = np.ones((1024*1024)) * i
            self.queue.put(status)
            await asyncio.sleep(0.1)

@ray.remote(num_cpus=1)
class Supervisor:
    def __init__(self, workers, queue):
        self.workers = workers
        self.queue = queue
        self.worker_tasks = []

    async def check_workers(self):
        should_finish = False
        while not should_finish:
            done, pending = await asyncio.wait(self.worker_tasks, timeout=0.001)
            if len(done) == len(self.workers):
                should_finish = True
                break
            
            if not self.queue.empty():
                status_update = await self.queue.get_async()
                print(status_update)

    async def run(self):
        for worker in self.workers:
            worker_task = worker.run.remote()
            # worker_task = asyncio.wrap_future(worker.run.remote().future())
            self.worker_tasks.append(worker_task)
        await self.check_workers()

async def main():
    status_queue = Queue()
    worker_a = Worker.remote(status_queue)
    worker_b = Worker.remote(status_queue)
    supervisor = Supervisor.remote([worker_a, worker_b], status_queue)
    await supervisor.run.remote()

asyncio.run(main())

Issue Severity

High: It blocks me from completing my task.