在 ray.io 中存储和检索对象答案

【问题标题】：Storing and retrieving object in ray.io在 ray.io 中存储和检索对象
【发布时间】：2020-07-13 12:00:15
【问题描述】：

我有一个光线集群在如下机器上运行：

ray start --head --redis-port=6379

我有两个文件需要在集群上运行。制作人 p_ray.py：

import ray
ray.init(address='auto', redis_password='5241590000000000')


@ray.remote
class Counter(object):
    def __init__(self):
        self.n = 0

    def increment(self):
        self.n += 1

    def read(self):
        return self.n


counters = [Counter.remote() for i in range(4)]
[c.increment.remote() for c in counters]
futures = [c.read.remote() for c in counters]
print(futures, type(futures[0]))
obj_id = ray.put(ray.get(futures))
print(obj_id)
print(ray.get(obj_id))
while True:
    pass

消费者c_ray.py：

import ray
ray.init(address='auto', redis_password='5241590000000000')
[objs] = ray.objects()
print('OBJ-ID:', objs, 'TYPE:', type(objs))
print(ray.get([objs]))

我的意图是存储来自生产者的期货对象并在消费者中检索它。我可以在消费者中检索对象 ID。然而，消费者的获取永远不会返回。

我做错了什么？

如何解决我的要求？

【问题讨论】：

标签： actor consumer ray publisher

【解决方案1】：

这种特殊情况可能是一个错误（我不是 100% 确定）。我在 Ray Github 上创建了一个问题。

但是，这不是获取 p_ray.py 创建的对象的好方法。如果你有很多对象，管理起来会非常复杂。您可以使用detached actor 实现类似的功能。 https://ray.readthedocs.io/en/latest/advanced.html#detached-actors。

这个想法是创建一个分离的actor，它可以被同一集群中运行的任何驱动程序/工作人员检索。

p_ray.py

import ray
ray.init(address='auto', redis_password='5241590000000000')

@ray.remote
class DetachedQueue:
    def __init__(self):
        self.dict = {}
    def put(key, value):
        self.dict[key] = value
    def get(self):
        return self.dict


@ray.remote
class Counter(object):
    def __init__(self):
        self.n = 0

    def increment(self):
        self.n += 1

    def read(self):
        return self.n


queue = DetachedQueue.remote(name="queue_1", detached=True)
counters = [Counter.remote() for i in range(4)]
[c.increment.remote() for c in counters]
futures = [c.read.remote() for c in counters]
print(futures, type(futures[0]))
queue.put.remote("key", ray.get(futures)))
while True:
    pass

c_ray.py：

import ray
ray.init(address='auto', redis_password='5241590000000000')
queue = ray.util.get_actor("queue_1")
print(ray.get(queue.get.remote()))

【讨论】：