Thanks - this is a really good bug report @aljeshishe ... out of curiosity are you able to workaround by spinning up a fresh Ray Cluster (or, put in another way, any reason not to)?

@jjyao on the attempt to repro...

@aljeshishe it's probably best to set up a call to do a live debug when the cluster is in this bad state.

I can help coordinate some time here > @aljeshishe can you Slack me and Jiajun on Ray Slack and we can go from there?

Thanks - this is a really good bug report @aljeshishe ... out of curiosity are you able to workaround by spinning up a fresh Ray Cluster (or, put in another way, any reason not to)?

@jjyao on the attempt to repro...

Usually, I kill the docker container with head and then I do ray up(to avoid head instance recreation).
Now I have to do it almost every day.
Ok I'll contact you in slack later.

It is probably related to runtime environment, as if I don't pass runtime_env = dict(py_modules=[pandas]) to ray.init, there is no issue.

This is super interesting. afaik, we should only download this once and reuse, but based on what you are saying, it seems like this cause issues

I am curious if you use "pip" instead of pymodules (runtime_env={"pip": ["pandas"]}), you have the same issue ? Or as a workaround, you can probably just add pandas to your docker images.

@rkooo567
I've tried pip/working_dir/py_modules, same issue.

@jjyao finally found out how to reproduce the issue

To degrade cluster run:

import random
from pathlib import Path
import ray
import logging

logger = logging.getLogger(__name__)

def main():
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    submodule_path = Path(__file__).parent / "submodule"
    submodule_path.mkdir(exist_ok=True)
    (submodule_path / "__init__.py").write_text(f"a = {random.random()}")
    
    ray.init("ray://...:10001", 
             runtime_env=dict(py_modules=[submodule_path])
             )
    @ray.remote(scheduling_strategy='SPREAD')
    def task():
        pass

    print(ray.get(task.remote()))

        
if __name__ == "__main__":
    main()

run this script 200 times to create 200 envs:

for i in {0..200}; do python run.py; done;

Now lets check degraded state:

from ray.util.queue import Queue
import time
import ray
import pandas
import logging


logger = logging.getLogger(__name__)
NUM_TASKS = 3000
SLEEP_TIME = 1

def main():
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

    ray.init("ray://...:10001", 
             runtime_env=dict(py_modules=[pandas])
             )
    q = Queue()
    @ray.remote(scheduling_strategy='SPREAD')
    def task():
        q.put(1)
        
        ts = time.time()
        while time.time() - ts < 1: pass
        
        return 42


    
    logger.info("scheduling tasks")
    refs = [task.remote() for _ in range(NUM_TASKS)]
    logger.info("Done")

    ts = time.time()

    while True:
        time.sleep(SLEEP_TIME)
        items = len(q.get_nowait_batch(q.size()))
        new_ts = time.time()
        refs += [task.remote() for _ in range(items)]
        
        elapsed = new_ts - ts
        scheduling_speed = items / elapsed
        availible_cpus = ray.available_resources().get("CPU", 0)
        cluster_cpus = ray.cluster_resources().get("CPU", 0)
        logger.info(f"CPUs: {availible_cpus=:.2f} {cluster_cpus=:.2f} {scheduling_speed=:.2f}")
        
        ts = new_ts

        
if __name__ == "__main__":
    main()

As you may see in script output, lots of environments degrade cluster, so scheduling almost stops for some time

For comparison you may checkout scheduling on fresh cluster

There is a slight 2-second pause.

In an ideal world, I expect no scheduling degradation at all.

@jjyao have you managed to reproduce ?

@kevin85421 did you get a chance to look into this deeper?

Looks vaguely similar to #45373
I wonder if it's connected.

Hello, guys, is it going to be fixed?

I noticed in the repro script that the job is submitted from outside the cluster using Ray Client (the service running at port 10001 on the head node). I've heard that this submission method is not well-supported (though perhaps this has changed, I'm not sure)

I'm curious -- Does the issue persist if you run the job directly on the head node, or use Ray Job Submission (the other way of submitting jobs, using the dashboard port 8265 by default.)