Ray 蒙特卡洛估算 π 实战：用 Task 并行采样、Actor 跟踪进度

Ray 蒙特卡洛估算 π 实战：用 Task 并行采样、Actor 跟踪进度

【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray

本指南基于 Ray 官方教程 Monte Carlo Estimation of π，讲解如何用 Ray Core 的两类核心抽象——无状态的 Ray task 和有状态的 Ray actor——把一个蒙特卡洛估算 π 的程序写成分布式应用。读完后，你将掌握：如何用@ray.remote把普通 Python 函数变成可并行执行的远程任务、用 actor 聚合多个任务的状态、通过ObjectRef与ray.get()获取异步结果，以及如何调整任务数和采样量把程序从笔记本扩展到更大集群。

数学原理：用随机采样估算 π

教程的数学背景是经典的蒙特卡洛方法：在边长为 2 的正方形内（坐标范围[-1, 1] × [-1, 1]）随机撒点，统计落在以原点为圆心、半径为 1 的单位圆内的点的比例。圆的面积为 π，正方形的面积为 4，因此“圆内点数 / 总点数”的估计值乘以 4 即可近似 π。采样点越多，估计值越接近 π 的真实值。

上图是仓库文档中的示例示意图：仅 1000 个样本时 π 的估计值约为 3.164；教程默认配置下每个任务采样 1000 万个点，共 10 个任务，总计 1 亿个点，估计精度会显著提高。

整个程序的分工由 Ray 两类原语承担：

• Ray task：把“采样并统计圆内点数”这一无状态计算拆成多个任务并行执行；
• Ray actor：用一个ProgressActor跟踪所有采样任务的累计进度，供主程序周期查询。

教程原文说明该代码可以直接在笔记本上运行，也可以平滑扩展到更大集群以提升估算精度。

完整示例代码

教程的完整可运行代码位于 doc_code/monte_carlo_pi.py，文档各小节通过标记注释（如__starting_ray_start__）从该文件摘取片段。完整代码如下，后文按教程的小节顺序逐段讲解：

import ray import math import time import random ray.init() # fmt: off @ray.remote class ProgressActor: def __init__(self, total_num_samples: int): self.total_num_samples = total_num_samples self.num_samples_completed_per_task = {} def report_progress(self, task_id: int, num_samples_completed: int) -> None: self.num_samples_completed_per_task[task_id] = num_samples_completed def get_progress(self) -> float: return ( sum(self.num_samples_completed_per_task.values()) / self.total_num_samples ) # fmt: on # fmt: off @ray.remote def sampling_task(num_samples: int, task_id: int, progress_actor: ray.actor.ActorHandle) -> int: num_inside = 0 for i in range(num_samples): x, y = random.uniform(-1, 1), random.uniform(-1, 1) if math.hypot(x, y) <= 1: num_inside += 1 # Report progress every 1 million samples. if (i + 1) % 1_000_000 == 0: # This is async. progress_actor.report_progress.remote(task_id, i + 1) # Report the final progress. progress_actor.report_progress.remote(task_id, num_samples) return num_inside # fmt: on # Change this to match your cluster scale. NUM_SAMPLING_TASKS = 10 NUM_SAMPLES_PER_TASK = 10_000_000 TOTAL_NUM_SAMPLES = NUM_SAMPLING_TASKS * NUM_SAMPLES_PER_TASK # Create the progress actor. progress_actor = ProgressActor.remote(TOTAL_NUM_SAMPLES) # Create and execute all sampling tasks in parallel. results = [ sampling_task.remote(NUM_SAMPLES_PER_TASK, i, progress_actor) for i in range(NUM_SAMPLING_TASKS) ] # Query progress periodically. while True: progress = ray.get(progress_actor.get_progress.remote()) print(f"Progress: {int(progress * 100)}%") if progress == 1: break time.sleep(1) # Get all the sampling tasks results. total_num_inside = sum(ray.get(results)) pi = (total_num_inside * 4) / TOTAL_NUM_SAMPLES print(f"Estimated value of π is: {pi}") assert str(pi).startswith("3.14")

运行前的准备很简单：通过pip install -U ray安装 Ray 即可。

启动 Ray 集群

教程的第一步是引入依赖模块并启动一个本地 Ray 集群：

import ray import math import time import random ray.init()

调用ray.init()后，Ray 会在当前进程所在机器上启动一个单节点集群（raylet、对象存储、GCS 等组件随之前台/后台拉起）。如果集群已经在运行，ray.init()会连接到已有集群。这一步的意义在于：后续所有@ray.remote定义的函数和类都不再直接在当前进程执行，而是被分发到 Ray 管理的 worker 进程中运行。

定义进度跟踪 Actor

教程的第二个小节定义了一个用于跟踪进度的 Ray actor：

@ray.remote class ProgressActor: def __init__(self, total_num_samples: int): self.total_num_samples = total_num_samples self.num_samples_completed_per_task = {} def report_progress(self, task_id: int, num_samples_completed: int) -> None: self.num_samples_completed_per_task[task_id] = num_samples_completed def get_progress(self) -> float: return ( sum(self.num_samples_completed_per_task.values()) / self.total_num_samples )

要点：

• Ray actor 本质上是“有状态服务”：持有实例（handle）的一方都可以调用它的方法。与无状态的 task 不同，actor 的方法调用共享同一份实例状态（这里的num_samples_completed_per_task字典）。
• 把普通 Python 类变成 Ray actor 的方式就是加@ray.remote装饰器，与把函数变成 task 的方式完全一致——两者共用同一个装饰器，Ray 依据被装饰对象是函数还是类来区分。
• 两个方法各司其职：report_progress()被各采样任务调用、按任务 ID 记录已完成采样数；get_progress()被主程序调用，返回“已完成采样总数 / 目标总数”的整体进度（0 到 1 之间）。

从源码结构看，ActorHandle类（见 python/ray/actor.py 第 2274 行起）的文档字符串明确写道：ActorHandle 有三种创建方式——对 ActorClass 调用.remote()、把 actor handle 作为参数传入 task（handle 的 fork）、或直接序列化 handle。这与教程中把progress_actor直接作为参数传给sampling_task.remote(...)的写法相互印证：task 接收到的 handle 是原 handle 的一份派生，指向同一个远程 actor 实例。

定义采样 Task

定义好 actor 后，教程的第三个小节定义真正干活的采样 task：

@ray.remote def sampling_task(num_samples: int, task_id: int, progress_actor: ray.actor.ActorHandle) -> int: num_inside = 0 for i in range(num_samples): x, y = random.uniform(-1, 1), random.uniform(-1, 1) if math.hypot(x, y) <= 1: num_inside += 1 # Report progress every 1 million samples. if (i + 1) % 1_000_000 == 0: # This is async. progress_actor.report_progress.remote(task_id, i + 1) # Report the final progress. progress_actor.report_progress.remote(task_id, num_samples) return num_inside

要点：

• Ray task 是无状态函数：通过@ray.remote装饰普通函数即可注册为远程函数，其.remote()调用即“Ray task”。
• 每个 task 在独立循环中对num_samples个点做(x, y) ~ Uniform(-1, 1)采样，用math.hypot(x, y) <= 1判断是否落在单位圆内（等价于x² + y² <= 1，且hypot对数值溢出更稳健），最后返回圆内点数。
• task 的参数类型标注为ray.actor.ActorHandle，这展示了从 task 内部调用 actor 方法的用法：progress_actor.report_progress.remote(task_id, i + 1)是异步调用，立即返回ObjectRef而不阻塞采样循环。这也是为什么代码注释强调 “This is async.”——每 100 万个样本上报一次进度，采样结束时再上报一次最终进度。
• 各 task 之间没有任何共享内存通信，唯一的协调通道就是这个共享的 progress actor。

关于 task 的更多机制（资源请求、ObjectRef 传递、取消等），仓库中有专门章节 Tasks 可延伸阅读；actor 的更多能力（资源声明、命名 actor、并发配置等）见 Actors。

创建 Progress Actor 实例

定义好 actor 类后，需要创建其实例：

# Change this to match your cluster scale. NUM_SAMPLING_TASKS = 10 NUM_SAMPLES_PER_TASK = 10_000_000 TOTAL_NUM_SAMPLES = NUM_SAMPLING_TASKS * NUM_SAMPLES_PER_TASK # Create the progress actor. progress_actor = ProgressActor.remote(TOTAL_NUM_SAMPLES)

调用ActorClass.remote(...)时，参数会作为构造函数__init__的实参传入。Ray 会在某个远程 worker 进程中创建并运行该 actor，调用本身立即返回一个actor handle，后续凭它调用 actor 方法。

三个参数是理解如何“按集群规模调整”的关键：

默认配置下总采样量为 1 亿个点；在示例输出中，估算结果为 3.1412202，已经精确到小数点后三位。

异步执行采样任务

创建 actor 后，教程以小列表推导一次性提交全部采样任务：

# Create and execute all sampling tasks in parallel. results = [ sampling_task.remote(NUM_SAMPLES_PER_TASK, i, progress_actor) for i in range(NUM_SAMPLING_TASKS) ]

两个关键机制值得注意：

• sampling_task.remote(...)立即返回一个ObjectRef（Ray 中的 future），函数体随后在远程 worker 进程中异步执行。提交本身不做任何计算，因此 10 个任务的提交几乎瞬时完成。
• 得到的results是一个ObjectRef列表，此时任何结果都还没就绪。真正的取值留到最后通过ray.get(results)一次性完成。

注意progress_actor被作为参数传入了每个任务——如前所述，Ray 会对其做 handle fork/序列化，使 10 个任务都能指向同一个 progress actor。

周期查询进度

任务跑起来后，主程序通过 actor 的get_progress()方法周期性地观察整体进度：

# Query progress periodically. while True: progress = ray.get(progress_actor.get_progress.remote()) print(f"Progress: {int(progress * 100)}%") if progress == 1: break time.sleep(1)

这里展示了 actor 方法调用的完整范式：

1. progress_actor.get_progress.remote()是异步调用，立即返回ObjectRef，方法在远程 actor 进程中执行（actor 方法在同一个 actor 进程内串行执行，因此report_progress写入的字典不会出现并发写冲突）；
2. 阻塞式ray.get()取回该ObjectRef的实际值，即当前进度比例；
3. 进度达到 1 时退出循环，否则time.sleep(1)后再次查询。

这正是“有状态服务”模式的体现：多个采样任务不断往 actor 写入局部状态，主进程则把 actor 当作只读观测点轮询。教程运行时的输出形如：

Progress: 0% Progress: 15% Progress: 28% Progress: 40% Progress: 50% Progress: 60% Progress: 70% Progress: 80% Progress: 90% Progress: 100% Estimated value of π is: 3.1412202

进度以每个任务每 100 万采样点一次的粗粒度上报，所以百分比呈跳跃式增长而非严格线性；最后一步跳到 100% 是因为每个任务结束时都会上报最终进度。

汇总结果并计算 π

所有任务完成后，教程的最后一个小节收集结果并计算 π：

# Get all the sampling tasks results. total_num_inside = sum(ray.get(results)) pi = (total_num_inside * 4) / TOTAL_NUM_SAMPLES print(f"Estimated value of π is: {pi}")

这里用到一个便捷特性：ray.get()除了接受单个ObjectRef，还可以接受ObjectRef列表并返回对应的结果列表（阻塞到全部就绪）。随后按公式π ≈ 4 × (圆内点数 / 总点数)得出最终估计值。文件末尾还有一行assert str(pi).startswith("3.14")，用于在文档构建/测试中校验估计值确实落在 3.14x 区间，保证示例结果的确定性下限。

小结与扩展方向

这篇教程用一个不足百行的脚本串起了 Ray Core 最核心的三条 API 路径：

1. ray.init()启动集群；
2. @ray.remote装饰类得到 actor、装饰函数得到 task；
3. .remote()异步提交并返回ObjectRef/ActorHandle，ray.get()阻塞取值。

它同时演示了两种分布式协作模式：map 式并行（N 个无状态任务各算一段、结果求和合并）和共享有状态协调点（所有任务向同一个 actor 汇报、主进程轮询进度）。把NUM_SAMPLING_TASKS与NUM_SAMPLES_PER_TASK调大即可让同一份代码跑在更大的集群上：Ray 负责跨节点调度任务，代码本身无需感知节点拓扑。若希望了解 task 的资源请求、ObjectRef作为参数的依赖传递，或 actor 的命名、并发与故障恢复等更多细节，可继续阅读 Tasks 与 Actors 两篇文档，完整示例代码则可随时参考 monte_carlo_pi.py。

【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray