python-3.x - 如何在python的类中使用光线并行性？

Question

我想使用光线任务方法而不是光线角色方法来并行化类中的方法。原因是后者似乎需要更改类的实例化方式(如图所示 here )。下面是一个玩具代码示例，以及错误

import numpy as np
import ray


class MyClass(object):
    
    def __init__(self):
        ray.init(num_cpus=4)
    
    @ray.remote
    def func(self, x, y):
        return x * y
    
    def my_func(self):
        a = [1, 2, 3]
        b = np.random.normal(0, 1, 10000)
        result = []
        # we wish to parallelise over the array `a`
        for sub_array in np.array_split(a, 3):
            result.append(self.func.remote(sub_array, b))
        return result

mc = MyClass()
mc.my_func()
>>> TypeError: missing a required argument: 'y'

出现错误是因为 ray 似乎不“知道”该类，因此它需要一个参数 self .
如果我们不使用类，代码工作正常:

@ray.remote
def func(x, y):
    return x * y

def my_func():
    a = [1, 2, 3, 4]
    b = np.random.normal(0, 1, 10000)
    result = []
    # we wish to parallelise over the list `a`
    # split `a` and send each chunk to a different processor
    for sub_array in np.array_split(a, 4):
        result.append(func.remote(sub_array, b))
    return result


res = my_func()
ray.get(res)
>>> [array([-0.41929678, -0.83227786, -2.69814232, ..., -0.67379119,
        -0.79057845, -0.06862196]),
 array([-0.83859356, -1.66455572, -5.39628463, ..., -1.34758239,
        -1.5811569 , -0.13724391]),
 array([-1.25789034, -2.49683358, -8.09442695, ..., -2.02137358,
        -2.37173535, -0.20586587]),
 array([ -1.67718712,  -3.32911144, -10.79256927, ...,  -2.69516478,
         -3.1623138 ,  -0.27448782])]```

正如预期的那样，我们看到输出是一个包含 4 个数组的列表。我怎样才能得到 MyClass使用射线处理并行性？

Answer 1

一些提示:

一般建议您只使用 ray.remote python中函数或类的装饰器(未绑定(bind)方法)。

您应该非常小心地拨打 ray.init在函数的构造函数内部，因为 ray.init是 不是 幂等(这意味着如果您实例化 MyClass 的多个实例，您的程序将失败)。相反，您应该确保 ray.init只在你的程序中运行一次。

我认为有 2 种方法可以实现您与 Ray 一起追求的结果。
你可以搬家 func在类之外，所以它变成了一个函数而不是一个绑定(bind)方法。请注意，在这种方法中 MyClass将被序列化，这意味着更改 func使到 MyClass不会反射(reflect)在函数之外的任何地方。在您的简化示例中，这似乎不是问题。这种方法可以最容易地实现最多的并行性。

@ray.remote
def func(obj, x, y):
    return x * y


class MyClass(object):
    def my_func(self):
        ...
        # we wish to parallelise over the array `a`
        for sub_array in np.array_split(a, 3):
            result.append(func.remote(self, sub_array, b))
        return result

您可以考虑的另一种方法是使用 async actors .在这种方法中，射线actor 将通过asyncio 处理并发，但是这个 comes with the limitations of asyncio .

@ray.remote(num_cpus=4)
class MyClass(object):
    async def func(self, x, y):
        return x * y
    
    def my_func(self):
        a = [1, 2, 3]
        b = np.random.normal(0, 1, 10000)
        result = []
        # we wish to parallelise over the array `a`
        for sub_array in np.array_split(a, 3):
            result.append(self.func.remote(sub_array, b))
        return result