A simple example to create an Ray actor to run in a loop

ray_multi.py

import time

import ray


@ray.remote(num_cpus=2)
class C(object):
    def __init__(self) -> None:
        self.stop = 0

    def set_stop(self):
        self.stop = 1
    
    def get_stop(self):
        return self.stop


@ray.remote(num_cpus=2)
class A(object):
    def run(self):
        d = ray.get_actor("D")
        self.data = ray.get(d.get_data.remote())

    def get_data(self):
        return self.data


@ray.remote(num_cpus=2)
class Data(object):
    def __init__(self, max_size=100) -> None:
        self.max_size = max_size
        self.data = []

    def append(self, v):
        self.data.append(v)

    def get_data(self):
        return self.data


@ray.remote(num_cpus=2)
class B(object):
    def run(self):
        iter = 0
        print(f'start b.run!')
        c = ray.get_actor("C")
        d = ray.get_actor("D")
        while True:
            iter += 1
            time.sleep(20)
            stop = ray.get(c.get_stop.remote())
            if stop == 1:
                print(f'in b.run stop: {stop}')
                data = ray.get(d.get_data.remote())
                print(f'in b: {data}')
                break


if __name__ == '__main__':
    ray.init()

    c = C.options(name="C").remote()
    d = Data.options(name='D').remote(max_size=10000)
    b = B.options(name="B").remote()
    a = A.options(name="A").remote()

    ws = [b.run.remote()]
    print("ws:", ws)

    for i in range(10):
        print(f'Append {i}')
        d.append.remote(i)
    
    time.sleep(20)
    c.set_stop.remote()
    time.sleep(20)
    print(f"d data: {ray.get(d.get_data.remote())}")
    a.run.remote()
    print(ray.get(c.get_stop.remote()))
    print(ray.get(a.get_data.remote()))
    ray.wait(ws)
    ray.shutdown()