Передача пользовательского объекта между подпроцессами

Question

У меня есть три файла следующим образом

""" main.py """
import time
import ray

class LocalBuffer(dict):
    def __call__(self):
        return self

@ray.remote
class Worker():
    def __init__(self, learner):
        self.local = LocalBuffer()
        self.learner = learner

    def sample(self):
        for i in range(10):
            self.local.update({
                'state': [1, 2, 3]
            })
            print(self.local)
            self.learner.update_buffer.remote(self.local)

@ray.remote
class Learner():
    def __init__(self):
        self.buffer = {}

    def update_buffer(self, local_buffer):
        print(local_buffer)
        self.buffer['state'] = local_buffer['state']

ray.init()

learner = Learner.remote()
worker = Worker.remote(learner)

worker.sample.remote()

time.sleep(10)

Приведенный выше код будет работать нормально, если я удалю весь код, связанный с ray.Если нет, то что-то не так происходит.Сообщение об ошибке говорит, что нет state в local_buffer в update_buffer.Я знаю, что ошибка возникает из-за LocalBuffer, определенного в worker.py - если я определю Worker.local как встроенный dict, все будет хорошо.Но почему я не могу использовать LocalBuffer?Мне это нужно, и я понятия не имею, как заставить его работать.

Обновление

Я вижу, в чем проблема.Причина в том, что worker и learner находятся в разных процессах.И пользовательский объект, такой как self.local, не может быть передан между процессами.Для этой конкретной проблемы я могу избавиться от этой проблемы, приведя self.local к dict, когда self.local передано self.learner.update_buffer.Я пытался импортировать LocalBuffer в learner.py, но это не сработало.Возможно, мне нужно больше узнать о многопроцессорности, чтобы понять это.Буду очень признателен, если кто-нибудь захочет дать мне полезную информацию.

Answer 1

Мы должны сделать LocalBuffer актера луча, чтобы заставить его работать.Следующий код работает как нужно.

import ray

@ray.remote
class LocalBuffer(dict):
    # have to redefine these functions in order to make it work with ray
    def __getitem__(self, k):
        return super().__getitem__(k)

    def update(self, d):
        super().update(d)

    def __call__(self):
        # cannot return self since self is a ray actor
        return dict(super().items())


@ray.remote
class Worker():
    def __init__(self, learner):
        self.local = LocalBuffer.remote()
        self.learner = learner

    def sample(self):
        for i in range(10):
            id = self.local.update.remote({
                'state': [1, 2, 3]
            })
            print(ray.get(self.local.__call__.remote()))
            self.learner.update_buffer.remote(self.local)


@ray.remote
class Learner():
    def __init__(self):
        self.buffer = {}

    def update_buffer(self, local_buffer):
        print(ray.get(local_buffer.__call__.remote()))
        self.buffer['state'] = ray.get(local_buffer.__getitem__.remote('state'))
        print('learner buffer', self.buffer)

ray.init()

learner = Learner.remote()
worker = Worker.remote(learner)

ray.get(worker.sample.remote())