我正在使用Ray来并行化一些计算，但它似乎正在积累溢出。
我不介意它将对象溢出到我的硬盘驱动器，但如果这意味着使用 +130 GiB 来处理大约 1.6 GiB 的模拟，我会这样做。
贝娄是正在发生的事情的痕迹：
Number of steps: 55 (9,091 simulations each)
0%
[2m[36m(raylet)[0m Spilled 3702 MiB, 12 objects, write throughput 661 MiB/s. Set RAY_verbose_spill_logs=0 to disable this message.
[2m[36m(raylet)[0m Spilled 5542 MiB, 17 objects, write throughput 737 MiB/s.
2%
[2m[36m(raylet)[0m Spilled 9883 MiB, 33 objects, write throughput 849 MiB/s.
5%
[2m[36m(raylet)[0m Spilled 16704 MiB, 58 objects, write throughput 997 MiB/s.
13%
[2m[36m(raylet)[0m Spilled 32903 MiB, 124 objects, write throughput 784 MiB/s.
29%
[2m[36m(raylet)[0m Spilled 66027 MiB, 268 objects, write throughput 661 MiB/s.
53%
[2m[36m(raylet)[0m Spilled 131920 MiB, 524 objects, write throughput 461 MiB/s.
60%
这是我正在运行的代码：
def get_res_parallel(simulations, num_loads=num_cpus):
load_size = simulations.shape[0] / num_loads
simulations_per_load = [simulations[round(n * load_size): round((n+1) * load_size)]
for n in range(num_loads)]
# 2D numpy arrays
results = ray.get([get_res_ray.remote(simulations=simulations)
for simulations in simulations_per_load])
return np.vstack(results)
MAX_RAM = 6 * 2**30 # 6 GiB
def get_expected_res(simulations, MAX_RAM=MAX_RAM):
expected_result = np.zeros(shape=87_381, dtype=np.float64)
bytes_per_res = len(expected_result) * (64 // 8)
num_steps = simulations.shape[0] * bytes_per_res // MAX_RAM + 1
step_size = simulations.shape[0] / num_steps
print(f"Number of steps: {num_steps} ({step_size:,.0f} simulations each)")
for n in range(num_steps):
print(f"r{n / num_steps:.0%}", end="")
step_simulations = simulations[round(n * step_size): round((n+1) * step_size)]
results = get_res_parallel(simulations=step_simulations)
expected_result += results.mean(axis=0)
print(f"r100%")
return expected_result / num_steps
在具有16 GiB内存，Ray 2.0.0和Python 3.9.13的Mac M1上运行。
问题
鉴于我的代码，这是正常行为吗？
我该怎么做才能解决此问题？强制垃圾回收？
你知道get_res_ray返回的数组的预期大小吗？
Ray 会溢出远程任务返回的对象以及传递给远程任务的对象，因此在这种情况下，有两个可能的地方会导致内存压力：
ObjectRefs由get_res_ray.remote返回simulations传给了get_res_ray.remote.由于它们很大，Ray 会自动将它们放入本地对象存储中，以减小任务定义的大小。
如果这些对象的总大小大于计算机上 RAM 的 30%(这是 Ray 对象存储的默认大小)，则可能会溢出。不建议增加对象存储的大小，因为这可能会导致函数的内存压力。
但是您可以尝试在每次迭代中处理更少的内容和/或尝试更快地发布ObjectRefs。特别是，您应该尝试尽快释放上一次迭代中的results，以便 Ray 可以为您 GC 对象。您可以使用它们后调用del results来执行此操作。
这是一个完整的建议，它将通过将数组结果馈送到另一个任务而不是将它们放在驱动程序上来执行相同的操作。这通常是一种更好的方法，因为它避免了在驱动程序上增加内存压力，并且你不太可能意外地将结果固定在驱动程序的内存中。
@ray.remote
def mean(*arrays):
return np.vstack(arrays).mean(axis=0)
def get_res_parallel(simulations, num_loads=num_cpus):
load_size = simulations.shape[0] / num_loads
simulations_per_load = [simulations[round(n * load_size): round((n+1) * load_size)]
for n in range(num_loads)]
# 2D numpy arrays
# Use the * syntax in Python to unpack the ObjectRefs as function arguments.
result = mean.remote(*[get_res_ray.remote(simulations=simulations)
for simulations in simulations_per_load])
# We never have the result arrays stored in driver's memory.
return ray.get(result)
MAX_RAM = 6 * 2**30 # 6 GiB
def get_expected_res(simulations, MAX_RAM=MAX_RAM):
expected_result = np.zeros(shape=87_381, dtype=np.float64)
bytes_per_res = len(expected_result) * (64 // 8)
num_steps = simulations.shape[0] * bytes_per_res // MAX_RAM + 1
step_size = simulations.shape[0] / num_steps
print(f"Number of steps: {num_steps} ({step_size:,.0f} simulations each)")
for n in range(num_steps):
print(f"r{n / num_steps:.0%}", end="")
step_simulations = simulations[round(n * step_size): round((n+1) * step_size)]
expected_result += get_res_parallel(simulations=step_simulations)
print(f"r100%")
return expected_result / num_steps