调度#

对于每个任务或 Actor，Ray 将选择一个节点来运行它，调度决策基于以下因素。

资源#

每个任务或 Actor 都有指定的资源需求。基于此，节点可以处于以下状态之一

可行：节点拥有运行该任务或 Actor 所需的资源。根据这些资源的当前可用性，存在两种子状态
- 可用：节点拥有所需的资源，且当前空闲。
- 不可用：节点拥有所需的资源，但这些资源当前被其他任务或 Actor 占用。
不可行：节点不具备所需的资源。例如，仅有 CPU 的节点对于 GPU 任务来说是不可行的。

资源需求是硬性要求，这意味着只有可行的节点才有资格运行该任务或 Actor。如果存在可行的节点，Ray 将根据下文讨论的其他因素，选择可用节点或等待不可用节点变为可用。如果所有节点都不可行，则任务或 Actor 在集群中添加可行的节点之前无法被调度。

调度策略#

任务或 Actor 支持 scheduling_strategy 选项来指定在可行节点中决定最佳节点的策略。当前支持的策略如下。

“DEFAULT”#

"DEFAULT" 是 Ray 使用的默认策略。Ray 将任务或 Actor 调度到一组排名前 k 的节点上。具体来说，节点按以下方式排序：首先倾向于已经调度了任务或 Actor 的节点（为了局部性），然后倾向于资源利用率低的节点（为了负载均衡）。在排名前 k 的组中，随机选择节点以进一步改善负载均衡，并缓解大型集群中的冷启动延迟。

在实现上，Ray 根据逻辑资源的利用率计算集群中每个节点的得分。如果利用率低于阈值（由 OS 环境变量 RAY_scheduler_spread_threshold 控制，默认为 0.5），则得分为 0，否则为资源利用率本身（得分 1 表示节点完全利用）。Ray 从得分最低的前 k 个节点中随机选择最佳节点进行调度。k 的值是（集群中的节点数 * RAY_scheduler_top_k_fraction 环境变量）和 RAY_scheduler_top_k_absolute 环境变量中的最大值。默认情况下，它是节点总数的 20%。

目前，Ray 特别处理不要求任何资源的 Actor（即 num_cpus=0 且没有其他资源），通过在集群中随机选择一个节点而不考虑资源利用率。由于节点是随机选择的，不要求任何资源的 Actor 实际上会 SPREAD 到整个集群中。

@ray.remote
def func():
    return 1


@ray.remote(num_cpus=1)
class Actor:
    pass


# If unspecified, "DEFAULT" scheduling strategy is used.
func.remote()
actor = Actor.remote()
# Explicitly set scheduling strategy to "DEFAULT".
func.options(scheduling_strategy="DEFAULT").remote()
actor = Actor.options(scheduling_strategy="DEFAULT").remote()

# Zero-CPU (and no other resources) actors are randomly assigned to nodes.
actor = Actor.options(num_cpus=0).remote()

“SPREAD”#

"SPREAD" 策略将尝试将任务或 Actor 分散到可用节点之间。

@ray.remote(scheduling_strategy="SPREAD")
def spread_func():
    return 2


@ray.remote(num_cpus=1)
class SpreadActor:
    pass


# Spread tasks across the cluster.
[spread_func.remote() for _ in range(10)]
# Spread actors across the cluster.
actors = [SpreadActor.options(scheduling_strategy="SPREAD").remote() for _ in range(10)]

PlacementGroupSchedulingStrategy#

PlacementGroupSchedulingStrategy 将任务或 Actor 调度到放置组所在的位置。这对于 Actor gang 调度很有用。更多详细信息请参阅放置组。

NodeAffinitySchedulingStrategy#

NodeAffinitySchedulingStrategy 是一种底层策略，允许将任务或 Actor 调度到由其节点 ID 指定的特定节点上。soft 标志指定如果指定的节点不存在（例如，节点死亡）或由于不具备运行任务或 Actor 所需的资源而不可行，是否允许在其他位置运行该任务或 Actor。在这些情况下，如果 soft 为 True，任务或 Actor 将被调度到不同的可行节点上。否则，任务或 Actor 将失败并抛出 TaskUnschedulableError 或 ActorUnschedulableError。只要指定的节点存活且可行，任务或 Actor 只会在那里运行，无论 soft 标志如何。这意味着如果节点当前没有可用资源，任务或 Actor 将等待资源可用。此策略应仅在其他高级调度策略（例如放置组）无法提供所需的任务或 Actor 放置时使用。它有以下已知限制

它是一种底层策略，阻止智能调度器进行优化。
由于任务或 Actor 创建时必须知道节点 ID，因此无法充分利用自动扩缩集群。
特别是在多租户集群中，很难做出最佳的静态放置决策：例如，应用程序不知道同一节点上还调度了什么。

@ray.remote
def node_affinity_func():
    return ray.get_runtime_context().get_node_id()


@ray.remote(num_cpus=1)
class NodeAffinityActor:
    pass


# Only run the task on the local node.
node_affinity_func.options(
    scheduling_strategy=ray.util.scheduling_strategies.NodeAffinitySchedulingStrategy(
        node_id=ray.get_runtime_context().get_node_id(),
        soft=False,
    )
).remote()

# Run the two node_affinity_func tasks on the same node if possible.
node_affinity_func.options(
    scheduling_strategy=ray.util.scheduling_strategies.NodeAffinitySchedulingStrategy(
        node_id=ray.get(node_affinity_func.remote()),
        soft=True,
    )
).remote()

# Only run the actor on the local node.
actor = NodeAffinityActor.options(
    scheduling_strategy=ray.util.scheduling_strategies.NodeAffinitySchedulingStrategy(
        node_id=ray.get_runtime_context().get_node_id(),
        soft=False,
    )
).remote()

局部性感知调度#

默认情况下，Ray 优先选择具有本地大型任务参数的可用节点，以避免通过网络传输数据。如果存在多个大型任务参数，则优先选择具有最多本地对象字节的节点。这优先于 "DEFAULT" 调度策略，这意味着 Ray 将尝试在局部性首选节点上运行任务，而无论节点资源利用率如何。但是，如果局部性首选节点不可用，Ray 可能会在其他地方运行任务。当指定了其他调度策略时，它们具有更高的优先级，并且不再考虑数据局部性。

注意

局部性感知调度仅适用于任务，不适用于 Actor。

@ray.remote
def large_object_func():
    # Large object is stored in the local object store
    # and available in the distributed memory,
    # instead of returning inline directly to the caller.
    return [1] * (1024 * 1024)


@ray.remote
def small_object_func():
    # Small object is returned inline directly to the caller,
    # instead of storing in the distributed memory.
    return [1]


@ray.remote
def consume_func(data):
    return len(data)


large_object = large_object_func.remote()
small_object = small_object_func.remote()

# Ray will try to run consume_func on the same node
# where large_object_func runs.
consume_func.remote(large_object)

# Ray will try to spread consume_func across the entire cluster
# instead of only running on the node where large_object_func runs.
[
    consume_func.options(scheduling_strategy="SPREAD").remote(large_object)
    for i in range(10)
]

# Ray won't consider locality for scheduling consume_func
# since the argument is small and will be sent to the worker node inline directly.
consume_func.remote(small_object)

调度#

资源#

调度策略#

“DEFAULT”#

“SPREAD”#

PlacementGroupSchedulingStrategy#

NodeAffinitySchedulingStrategy#

局部性感知调度#

更多关于 Ray 调度#