Удаленные вызовы блокируются при использовании методов в объекте Actor? - PullRequest
Новая
       15

Удаленные вызовы блокируются при использовании методов в объекте Actor?

1 голос
/ 12 июня 2019

Выполнение следующего не будет работать одновременно, вместо этого оно сначала выполнит Run1 и заблокирует его до завершения, прежде чем выполнит Run2. 

@ray.remote
class Test:
    def __init__(self):
        pass

    def Run1(self):
        print('Run1 Start')
        sleep(5)
        print('Run1 End')

    def Run2(self):
        print('Run2')

ray.init()
test = Test.remote()
test.Run1.remote()
test.Run2.remote()

sleep(10)

Выход: 

(pid=8109) Run1 Start
(pid=8109) Run1 End
(pid=8109) Run2

Это немного неожиданно. Как я могу обеспечить одновременное выполнение методов?

РЕДАКТИРОВАТЬ, ЧТОБЫ УЗНАТЬ СЛЕДУЮЩИЕ КОММЕНТАРИИ:

Использование двухпоточного подхода, похоже, не работает. Приведенный ниже код постоянно приводит к разрыву каналов из PyArrow. Я хотел бы запустить как метод self.PreloadSamples, так и методы self.Optimize последовательно. Класс BufferActor собирает и предоставляет пакетные образцы с помощью метода GetSamples (), украшенного @ ray.remote. Поскольку данные на графическом процессоре не сериализуемы, это нужно делать на стороне объекта Optimizer, и я хочу убедиться, что это делается параллельно, а не последовательно в отношении оптимизации.

См. Ниже полностью изолированную версию проблемы, которая реплицирует проблемы примерно через 1 минуту работы: 

import torch
import ray
import threading
from time import sleep


def Threaded(fn):
    def wrapper(*args, **kwargs):
        thread = threading.Thread(target=fn, args=args, kwargs=kwargs)
        thread.start()
        return thread
    return wrapper

@ray.remote
class BufferActor():
    def __init__(self):
        pass

    def GetSamples(self):
        return torch.randn(32, 100)


@ray.remote(num_gpus=1)
class OptimizerActor():
    def __init__(self, bufferActor):
        self.bufferActor = bufferActor
        self.samplesOnGPU = list()

        self.PreloadSamples()
        self.Optimize()

    @Threaded
    def PreloadSamples(self):
        #this retrieves a batch of samples (in numpy/torch format on CPU)
        if (len(self.samplesOnGPU) < 5):
            samples = ray.get(self.bufferActor.GetSamples.remote())

            self.samplesOnGPU.append(samples.to('cuda'))

            print('Samples Buffer: %s' % len(self.samplesOnGPU))
        else:
            sleep(0.01)

        self.PreloadSamples()

    @Threaded
    def Optimize(self):
        if (len(self.samplesOnGPU) > 0):
            samples = self.samplesOnGPU.pop(0)
            print('Optimizing')

            #next we perform loss calc + backprop + optimizer step (not shown)

        sleep(0.01)
        self.Optimize()



ray.init()

bufferActor = BufferActor.remote()
optimizerActor = OptimizerActor.remote(bufferActor)

sleep(60*60)

1 Ответ

1 голос
/ 13 июня 2019

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

...