Actor np.array属性在更改后不会更新； Ray for Python

如何解决Actor np.array属性在更改后不会更新； Ray for Python

我在参数服务器设置中遇到一些奇怪的行为。 可能是代码本身，或者可能是我误用了。

我有一个用numpy数组实例化的服务器（它也可以完成一些工作）。在将数组作为参数传递后，我将其复制。服务器作为参数传递给多个工作程序。阵列在工作人员之间划分，他们将更改提交给阵列，但是阵列只能由服务器更新。

我猜想与标准参数服务器唯一的区别是“参数”全部容纳在同一阵列的各个部分中，并且存在规则的离散时间点，在这些时间点上，工作人员停止工作，服务器进行某些处理。

问题：有时候服务器对阵列进行更新（是其自身的属性）时，这些更新将丢失/丢弃/不保留。而且我无法真正分辨出为什么保留一些更新而丢失一些更新的原因或模式。

更奇怪的是，该数组在服务器的update方法内部进行了更新（即，我可以在该方法中打印或返回该数组，并查看它是否已正确更新），但是该数组没有当服务器稍后通过其他方法/句柄访问时，不再显示更新。

我相当有信心这不是一般的代码问题，因为我可以简单地将数组从服务器返回到主线程中，然后稍后将其传递回服务器，更新数组即可解决问题（此操作是我目前的解决方法；很明显，这需要处理时间。）

这可能是时间问题（可能是工人在更新阵列之前访问服务器），但是我进行了阵列更新并插入了单独的阻塞线，所以我不确定为什么会这样。

下面是代码的简化版本。不幸的是，它不能重现错误，但是也许有人可以在我的结构中发现错误。

该数组只有2d（10000，20，dtype = np.int64），而且我使用的是8核Macbook，所以感觉好像不大是个问题，但是也许？

import numpy as np
import ray

@ray.remote
class Server:
    def __init__(self,arr):
        self.arr = np.copy(arr)
        self.batch_size = 100
    
    def get_arr(self):
        return self.arr
    
    def get_batch(self,id_):
        return self.arr[self.batch_size*id_:self.batch_size*(1+id_)]

    def worker_changes(self,batch,id_):
        self.arr[self.batch_size*id_:self.batch_size*(1+id_)] = batch
    
    def server_update(self,i):
        self.i = i
    
    def arr_update(self):
        self.arr[:,5] = self.arr[:,6:9].sum(axis=1)
        
@ray.remote
class Worker:
    def __init__(self,server,id_):
        self.server = server
        self.id = id_

    def work(self,i):
        batch_readonly = ray.get(self.server.get_batch.remote(self.id))
        batch = np.copy(batch_readonly)
        batch[:,0] = batch[:,:3].sum(axis=1)
        self.server.worker_changes.remote(batch,self.id)

ray.init()
size = 10000
arr = np.random.randint(0,1000,size=(size,20))


server = Server.remote(arr)
workers = [Worker.remote(server,id_) for id_ in range(5)]
for i in range(5):
    ray.get(server.server_update.remote(i=i))
    ray.get([worker.work.remote(i) for worker in workers])
    ray.get(server.arr_update.remote())

newarr = ray.get(server.get_arr.remote())
checkarr = np.copy(arr[0,:3])
for i in range(5):
    checkarr[0] = checkarr.sum()


checkarr == newarr[0,:3]
>>> array([ True,True,True])

当我运行上面的精简版本时，我得到了您期望的答案（真，真，真）。当我在项目中运行类似的结构时，在那里会出现一些错误。

无论如何，只是想知道以前是否有人见过类似的问题。