Сериализация будущего луча и возвращение результатов позже

Question

Я обернул Ray в веб-API (используя ray start --head и uvicorn с ray.init). Теперь я пытаюсь:

1. Отправить задание Ray (через API) и сериализовать будущее и вернуться к пользователю
2. Позже, позвольте пользователю нажать API для посмотреть, завершена ли задача, и вернуть результаты

Кикер состоит в том, что будучи многопоточным, я не могу гарантировать, что следующий вызов будет из того же потока. Вот то, что я наивно хотел бы сработать:

id = my_function.remote()
id_hex = id.hex()

Затем в другом запросе / вызове:

id = ray._raylet.ObjectID(binascii.unhexlify(id_hex))
ray.get(id)

Теперь это никогда не заканчивается (время истекает), хотя я знаю будущее завершен, и код работает, если я запустил его в том же потоке, что и исходный запрос.

Я предполагаю, что это связано с использованием другой инициализации Ray.

Есть ли способ заставить Рэя «refre sh» фьючерсного результата от Redis?

Answer 1

Получение objectID прямо на вашем пути может вызвать неожиданное поведение из-за механизма подсчета / оптимизации ссылок Ray. Одна рекомендация - использовать «отдельный актер». Вы можете создать отдельного актера и делегировать ему вызов. Отдельные актеры выживут при жизни Рэя (если вы его не убьете), поэтому вам не нужно беспокоиться о проблемах, о которых вы упомянули. Да. Это может сделать программу немного медленнее, так как для этого требуется 2 перехода, но я полагаю, что эти накладные расходы не будут иметь значения для вашей рабочей нагрузки (модель представления клиента).

https://docs.ray.io/en/latest/advanced.html?highlight=detached#detached -акторы

ray.remote
class TaskInvocator:
    def __init__(self):
        self.futures = {}
    def your_function(self):
        object_id = real_function.remote()
        self.futures[object_id.hex()] = object_id
    def get_result(self, hex_object_id):
        return ray.get(self.futures[hex_object_id])

TaskInvocator.remote(detached=True, name='invocator-1')
task_invocator = ray.util.get_actor('invocator-1')
object_id = task_invocator.your_function.remote()
blah blah...
result = ray.get(task_invocator.get_result.remote(object_id.hex()))

Answer 2

Сериализация будущего луча в внеполосной строке с последующей десериализацией не поддерживается. Причина этого в том, что эти фьючерсы - это больше, чем просто идентификаторы, с ними связано много состояний в различных системных компонентах. управляет временем существования этих задач. Когда вы запускаете задачу, вы передаете ее ObjectID актеру. Затем, когда пользователь обращается к конечной точке, чтобы проверить, завершено ли она, он проверяет связь с актором, который ищет соответствующий ObjectID и вызывает для него ray.wait ().