ray.data map_batches leak dead ray::IDLE

[happyxhw]

ray.data map_batches leak dead ray::IDLE

[happyxhw]

using yield, not return, can not work too !

https://docs.ray.io/en/latest/data/api/doc/ray.data.Dataset.map_batches.html

import numpy as np
import ray


class Embedding(object):
    """
    embedding
    """

    def __call__(self, item):
        """
        ray call
        """
        for i in range(0, 100):
            yield {'vector': np.random.uniform(low=0.5, high=13.3, size=(68000,)).tolist()}


@ray.remote(num_cpus=0)
def debug():
    ds = ray.data.from_items(list(range(0, 200)))
    ds = ds.map_batches(
        Embedding,
        batch_size=10,
        concurrency=(1, 4),
        num_cpus=0,
    )
    for item in ds.iter_batches():
        pass


if __name__ == '__main__':
    future = debug.remote()
    ray.get(future)
    ray.shutdown()

[happyxhw]

try to kill by pid can release the memory

import numpy as np
import ray
import ray.runtime_context


class Embedding(object):
    """
    embedding
    """

    def __call__(self, item):
        """
        ray call
        """
        for i in range(0, 100):
            yield {'vector': np.random.uniform(low=0.5, high=13.3, size=(380,)).tolist()}


@ray.remote(num_cpus=0)
def debug():
    ds = ray.data.from_items(list(range(0, 200)))
    ds = ds.map_batches(
        Embedding,
        batch_size=10,
        concurrency=(1, 4),
        num_cpus=0,
    )
    for item in ds.iter_batches():
        pass
    return ray.runtime_context.get_runtime_context().get_worker_id()


if __name__ == '__main__':
    future = debug.remote()
    worker_id = ray.get(future)

    import os
    import signal
    import time

    from ray.experimental.state.api import get_worker

    worker = get_worker(worker_id)
    pid = worker.pid
    ray.shutdown()

    time.sleep(10)
    os.kill(pid, signal.SIGKILL)

[bveeramani]

Hey @happyxhw, looks like this is expected behavior: https://discuss.ray.io/t/why-is-there-a-lot-of-ray-idle-in-my-ray-process/10395

[happyxhw]

Hey @happyxhw, looks like this is expected behavior: https://discuss.ray.io/t/why-is-there-a-lot-of-ray-idle-in-my-ray-process/10395

I have read this discussion. I know "ray::IDLE" is something similar to a progress in a pool. But in this issue the "ray::IDLE" is dead without a "ID". It never goes away while consuming too much memory, and can not be reused forever.

[bveeramani]

I ran the reproduction script with Ray 2.9.2, but I wasn't able to reproduce the issue. All of the IDLE processes have a PID.

@rynewang do you have a hunch about what might be going on here?