Ray: A Distributed Framework for Emerging AI Applications

arXiv:1712.05889v1 [cs.DC] 16 Dec 2017

Philipp Moritz∗, Robert Nishihara,∗ Stephanie Wang, Alexey Tumanov, Richard Liaw,
Eric Liang, William Paul, Michael I. Jordan, Ion Stoica
UC Berkeley

Abstract

with learning to operate continuously within an uncertain environment [44]. RL-based applications have already led to remarkable results, such as Google’s AlphaGo beating a human world champion [43], and are
finding their way into self-driving cars, UAVs [33], and
robotic manipulation [23, 47].
There are three characteristics that distinguish RL applications from traditional supervised learning applications. First, they often rely heavily on simulations to
explore states and discover the consequences of actions.
A simulator could encode the rules of a computer game
[11], the Newtonian dynamics of a physical system such
as a robot [45], or the hybrid dynamics of virtual environments. This generally requires massive amounts of
computation; for example, a realistic application might
perform hundreds of millions of simulations. Second,
the computation graph of an RL application is heterogeneous and evolves dynamically. A simulation can take
from a few milliseconds to a few minutes, and the result of a simulation can determine the parameters of future simulations. Third, many RL applications, such as
robotic control or autonomous driving, require actions
to be taken quickly in response to a constantly changing
environment. Furthermore, to choose the best action, the
application may need to perform more simulations in real
time. In summary, we need a computation framework
that supports heterogeneous and dynamic computation
graphs, while handling millions of tasks per second with
millisecond-level latencies.
Existing cluster computing frameworks fall short
of adequately satisfying these requirements. MapReduce [18], Apache Spark [50], Dryad [25], Dask [38],
and CIEL [32] support neither the throughput nor the latencies required by general RL applications, while TensorFlow [5], Naiad [31], MPI [21], and Canary [37] generally assume static computation graphs.1

The next generation of AI applications will continuously
interact with the environment and learn from these interactions. These applications impose new and demanding systems requirements, both in terms of performance
and flexibility. In this paper, we consider these requirements and present Ray—a distributed system to address
them. Ray implements a dynamic task graph computation model that supports both the task-parallel and the
actor programming models. To meet the performance
requirements of AI applications, we propose an architecture that logically centralizes the system’s control state
using a sharded storage system and a novel bottom-up
distributed scheduler. In our experiments, we demonstrate sub-millisecond remote task latencies and linear
throughput scaling beyond 1.8 million tasks per second.
We empirically validate that Ray speeds up challenging
benchmarks and serves as both a natural and performant
fit for an emerging class of reinforcement learning applications and algorithms.

1

Introduction

Artificial intelligence is currently emerging as the
workhorse technology for a range of real-world applications [27]. To date, however, these applications have
largely been based on a fairly restricted supervised learning paradigm in which a model is trained offline and
deployed to serve predictions online. As the field matures, it will be necessary to consider a broader setting
than standard supervised learning. Instead of making
and serving a single prediction, machine learning applications must increasingly operate in dynamic environments, react to changes in the environment, and take sequences of actions to accomplish a goal [6, 34]. These
broader requirements are naturally framed within the
paradigm of reinforcement learning (RL), which deals
∗ equal

1 While deep learning frameworks like TensorFlow or MXNet are
critical components of RL systems, they are not sufficient on their own.
Existing distributed RL applications [3, 20, 39] that use these systems

contribution

1

In this paper, we propose Ray, a cluster computing
framework that satisfies these requirements. To support
the heterogeneous and dynamic workloads imposed by
these applications, Ray implements a dynamic task graph
computation model, similar to CIEL [32]. However, Ray
also provides an actor programming abstraction on top
of this execution model, in addition to the task-parallel
abstraction provided by CIEL. The actor abstraction enables Ray to support stateful components, such as thirdparty simulators.
To achieve the stringent performance targets while
supporting dynamic computation graphs, Ray employs
a new distributed architecture that is horizontally scalable. The architecture is based on two key ideas. First,
we store all of the control state of the system in a global
control store, which enables all of the other components
in the system to be stateless. As a result, each component
can easily be scaled horizontally and restarted in the case
of failures. In turn, the global control store can be scaled
via sharding and made fault tolerant via replication.
Second, we introduce a new bottom-up distributed
scheduler, where tasks are submitted by workers and
drivers to local schedulers (there is one local scheduler
per node). Local schedulers may choose to schedule
tasks locally or to forward tasks to a replicated global
scheduler. This decreases task latency by allowing local
decisions, and increases the system throughput by reducing the burden on the global scheduler. We make the
following contributions:

simple RL application to illustrate the key requirements
for Ray.
An RL system consists of an agent that interacts repeatedly with the environment (see Figure 1(a)). The
goal of the agent is to learn a policy that maximizes some
reward. A policy is a mapping from the state of the environment to an action to take. The definitions of environment, agent, state, action, and reward are applicationspecific (Table 1).
Figure 2 shows an example of the pseudocode used
by an agent to learn a policy. A typical procedure consists of two steps: (1) evaluate the current policy, and
(2) improve the policy. To evaluate the policy, the pseudocode invokes rollout(environment, policy) to generate a set of rollouts, where a rollout is a trajectory of
states and rewards collected by interacting with the environment using environment.step(action). An action is
computed given the current policy and environment state
via policy.compute(state). As trajectories are generated,
train policy() uses completed trajectories to improve the
current policy via policy.update(trajectories). This process is repeated until the policy converges.
While simple, this application illustrates the key requirements of emerging AI applications. We group these
requirements into three categories.
Flexibility. The flexibility of a system is typically
measured in terms of the diversity of the workloads it can
support. We consider two aspects of flexibility: the heterogeneity of concurrently executing tasks and the generality and dynamicity of the execution graph.
Concurrent, heterogeneous tasks. Concurrent tasks
can be heterogeneous along three dimensions:

• We specify the systems requirements for emerging
AI applications: support for (a) heterogeneous, concurrent computations, (b) dynamic task graphs, (c)
high-throughput and low-latency scheduling, and
(d) transparent fault tolerance.

• Functionality. In the case of a robot, assessing the
environment’s state (e.g., environment.step(action))
involves processing the inputs of multiple sensors,
such as video, microphone, and radar. This requires
running multiple tasks in parallel, each implementing a different computation (see Figure 1(b)).

• We provide the actor abstraction–in addition to the
task-parallel programming abstraction–on top of a
dynamic task graph computation model.

• Duration. The time it takes to compute a trajectory
can vary significantly (see rollout(policy, environment)). For example, in the case of a game, it could
take just a few actions (moves) to lose the game, or
it could take hundreds of moves to win it.

• We propose a horizontally scalable architecture to
meet the above requirements, and build Ray, a cluster computing system that realizes this architecture.

2

• Resource types. Computing an action by evaluating
the policy (e.g., policy.compute(state)) is in many
cases implemented by a deep neural network, which
typically requires the use of GPUs. On the other
hand, most of the other computations use CPUs.

Motivation and Requirements

While Ray can support a variety of workloads—as it provides both the task-parallel and the actor abstractions—
we focus on RL workloads in this paper, as they are representative of emerging AI applications, and were the
primary driver behind Ray’s design. Here, we consider a

Note that these requirements are not naturally satisfied by
the Bulk Synchronous Parallel (BSP) model [46], which
is implemented by many of today’s popular cluster computing frameworks [18, 50]. With BSP, all tasks within

rely on ad-hoc solutions or other distributed frameworks to implement
task-parallelism and fault tolerance and to integrate stateful simulators.

2

(a)

(b)

(c)

Figure 1: (a) An RL system. (b) Task graph for processing sensor inputs. (c) Task graph for learning policy.
Application
Go (game)
Atari Pong
Robot moving an object

Environment
Board and opponent
Program
Physical world

Agent
Player
Player
Control program

State
Board position
Sequence of video frames
Object and robot pose

Action
Place a piece
Control joystick
Actuate joints

Reward
Game outcome
Pong score
Object moved

Table 1: Example RL applications.
// generate a trajectory under a given policy
rollout(policy, environment):
trajectory ← []
state ← environment.initial state()
while (not environment.has terminated()):
action ← policy.compute(state)
state, reward ← environment.step(action)
trajectory.append(state, reward)
return trajectory

the environment and compute a new action in a matter
of milliseconds. Similarly, simulations might take on
the order of milliseconds as well. As such, we need to
be able to schedule tasks in less than one millisecond.
Otherwise the scheduling overhead could be significant.
Given that clusters with tens of thousands of cores are
common, we need to be able to schedule hundreds of
thousands or even millions of tasks per second. Consider a cluster consisting of 100 servers, each with 32
cores, and suppose each task takes 5ms to execute. To
fully utilize this cluster, we would need to schedule 640K
tasks/sec.

// learn a policy in a given environment
train policy(environment):
policy ← initial policy()
while (policy has not converged):
trajectories ← []
// generate k rollouts and use them to update policy
for i from 1 to k:
trajectories.append(rollout(policy, environment))
policy = policy.update(trajectories)
return policy

Ease of development. Since writing parallel applications is non-trivial, and since ML developers prefer to
focus on their applications rather than on systems programming, simplifying development is paramount for the
success of such a system.
Deterministic replay and fault tolerance. The ability
to deterministically replay a job dramatically simplifies
debugging. Transparent fault tolerance obviates the need
for users to handle faults explicitly. It also enables users
to use cheap preemptible resources (e.g., spot instances
on AWS), leading to substantial cost savings when running in a public cloud.
Easy parallelization of existing algorithms. This involves providing a simple API and supporting existing
languages, tools, and libraries. First, we need to provide support for Python, as Python is the language of
choice for AI developers. Second, we need to provide
tight integration with the wide range of available thirdparty libraries. These libraries include simulators such as
OpenAI gym [13], DeepMind Lab [10], and the Mujoco
physics simulator [45] as well as deep learning frameworks like TensorFlow [5], Theano [12], PyTorch [4],
and Caffe [26]. As we will see, this requires augmenting
the task-parallel model with an actor-like abstraction to
wrap these third-party services.

Figure 2: Pseudocode for a typical RL training application.
the same stage2 typically perform the same computations
(albeit on different data partitions) and take roughly the
same amounts of time.3
Dynamic task graphs. Consider the train policy()
function. Though not shown in Figure 2, it would be natural to update the policy as soon as a subset of rollouts
finish (instead of waiting for all of them) and to launch
new rollouts to maintain a pool of executing rollouts as in
Figure 1(c). This makes the execution graph dynamic, as
we cannot predict the order in which rollouts will complete or which rollouts will be used for a particular policy
update.
Performance. In the case of a robot interacting with
the physical environment, we need to infer the state of
2A

stage is the unit of parallelism in BSP.
straggler mitigation techniques depend on this assumption [8].
3 Many

3

3

Programming and Computation Model

Ray provides basic support for stateful components in
the form of actors. In Ray, an actor is a stateful process
that exposes a set of methods that can be invoked as remote functions and that executes these methods serially.

Ray implements a dynamic task graph computation
model. On top, Ray provides both an actor and a taskparallel programming abstraction. This dual abstraction
differentiates Ray from related systems, such as CIEL
which only provides a task-parallel abstraction, and Orleans, which primarily provides an actor abstraction [14].

3.1

3.2

Computation Model

Ray employs a dynamic task graph computation
model [19], in which the execution of both remote functions and actor methods is automatically triggered by
the system when their inputs become available. In
this section, we describe how the computation graph
(Figure 3(b)) is constructed from a user program (Figure 3(a)). This program uses the API in Table 2 to implement the pseudocode from Figure 2.
Ignoring actors first, there are two types of nodes in
a computation graph: data objects and remote function
invocations, or tasks. There are also two types of edges:
data edges and control edges. Data edges capture the
dependencies between data objects and tasks. More precisely, if data object D is an output of task T , we add a
data edge from T to D. Similarly, if D is an input to T ,
we add a data edge from D to T . Control edges capture
the computation dependencies that result from nested remote functions (Section 3.1): if task T1 invokes task T2 ,
then we add a control edge from T1 to T2 .
Actor method invocations are also represented as
nodes in the computation graph. They are identical to
tasks with one key difference. To capture the state dependency across subsequent method invocations on the
same actor, we add a third type of edge: a stateful edge.
If method M j is called right after method Mi on the same
actor, then we add a stateful edge from Mi to M j . Thus,
all methods invoked on the same actor object form a
chain connected by stateful edges (Figure 3(b)). This
chain captures the order in which these methods were
invoked.
Stateful edges help us embed actors in an otherwise
stateless task graph, as they capture the implicit data dependency between successive method invocations sharing the internal state of an actor. Stateful edges also enable us to maintain lineage. As in other dataflow systems [50], we track data lineage to enable reconstruction. By explicitly including stateful edges in the lineage graph, we can easily reconstruct lost data, whether
produced by remote functions or actor methods (Section 4.2.3).

Programming Model and API

At its core, Ray provides a task-parallel programming
model. Table 2 shows Ray’s API. When a remote function is invoked, a future that represents the result of the
task is returned immediately. Futures can be retrieved using ray.get() and composed, i.e., a future can be passed
as an argument into another remote function. This allows the user to express parallelism while capturing data
dependencies.
Remote functions operate on immutable objects, and
are expected to be stateless and side-effect free: their
outputs are determined solely by their inputs. This
implies idempotence, which simplifies fault tolerance
through function re-execution on failure.
To satisfy the requirements for heterogeneity, flexibility, and ease of development given in Section 2, we augment the task-parallel programming model in four ways.
First, to handle concurrent tasks with heterogeneous
durations, we introduce ray.wait(). This call takes in
a list of futures and returns the subset whose results
are available, either after a timeout or when at least k
are available. In contrast, ray.get() blocks until all requested futures are available. This is highly beneficial
for RL applications, as simulations may have widely different durations, but complicates fault tolerance due to
introduced nondeterminism.
Second, to handle resource-heterogeneous tasks, we
enable developers to specify resource requirements so
that the Ray scheduler can efficiently manage resources.
The resources specified for a remote function are only
allocated during the function’s execution.
Third, to improve flexibility, we enable nested remote functions, meaning that remote functions can invoke other remote functions. This is also critical for
achieving high scalability (see Section 4), as it enables
multiple processes to invoke remote functions in parallel (otherwise the driver becomes a bottleneck for task
invocations).
Finally, and most importantly, for ease of development
and efficiency, we enhance our programming model with
an actor abstraction. One limitation we encountered
early in our development with stateless tasks was the
inability to wrap third-party simulators, which do not
expose their internal state. To address this limitation,

4

Architecture

4.1

Application Layer

The application layer consists of three components:
• Driver: A process executing the user program.
4

Name
futures list = f.remote(args)
obj list = ray.get(futures list)
futures list done =
ray.wait(futures list, k, timeout)
actor = Class.remote(args)
futures list = actor.method.remote(args)

Description
Execute function f () remotely. f () takes either object values or futures as arguments,
and returns a list of futures. This is a non-blocking call.
Return the values associated with a list of futures. This is a blocking call.
Given a list of futures, return the futures whose corresponding tasks have completed
as soon as either k of the tasks have completed or the timeout expires.
Instantiate class Class as a remote actor, and return a reference to it. Call a method
on the remote actor and return a list of futures. This is a non-blocking call.

Table 2: Ray API
T0
train_policy

@ray.remote
def create_policy():
# Initialize the policy randomly.
return policy

@ray.remote(num_gpus=2)
def update_policy(policy, *rollouts):
# Update the policy.
return policy

@ray.remote(num_gpus=1)
@ray.remote
class Simulator(object):
def train_policy():
def __init__(self):
# Create a policy.
# Initialize the environment.
policy_id = create_policy.remote()
self.env = Environment()
# Create 10 actors.
def rollout(self, policy, num_steps):
simulators = [Simulator.remote() for _ in range(10)]
observations = []
# Do 100 steps of training.
observation = self.env.current_state()
for _ in range(100):
for _ in range(num_steps):
# Perform one rollout on each actor.
action = compute(policy, observation)
rollout_ids = [s.rollout.remote(policy) for s in simulators]
observation = self.env.step(action)
# Update the policy with the rollouts.
observations.append(observation)
policy_id = update_policy.remote(policy_id,
return observations
*rollout_ids)
return ray.get(policy_id)

A10
Simulator

T1
create_policy

A20
Simulator

A11
rollout

policy1

A21
rollout
rollout21

rollout11
T2
update_policy
policy2

A12
rollout

rollout22

rollout12

…
object
data edges

A22
rollout

T3
update_policy

…

control edges

(a)

…
task/method
stateful edges

(b)

Figure 3: (a) Python code implementing the example in Figure 2 in Ray. Note that @ray.remote indicates remote functions and
actors. Invocations of remote functions and actor methods return futures, which can be passed to subsequent remote functions or
actor methods to encode task dependencies. Each actor has an environment object self.env shared between all of its methods. (b)
The task graph corresponding to an invocation of train policy.remote(). Remote function calls and the actor method calls correspond
to tasks in the task graph. The figure shows two actors. The method invocations for each actor (the tasks labeled A1i and A2i ) have
stateful edges between them indicating that they share the mutable actor state. There are control edges from train policy to the tasks
that it invokes. To train multiple policies in parallel, we could call train policy.remote() multiple times.

• Worker: A stateless process that executes tasks
(remote functions) invoked by a driver or another
worker. Workers are started automatically and assigned tasks by the system layer. When a remote
function is declared, the function is automatically
published to all workers. A worker executes tasks
serially.

contrast, an actor is a stateful process in that the results of
a method invocation can depend on the previous methods
executed by that actor.

4.2

System Layer

The system layer enables us to meet the performance and
fault tolerance goals, as discussed in Section 2, by employing an architecture in which each component is horizontally scalable and fault-tolerant. This layer consists
of three major components: a global control store, a distributed scheduler, and a distributed object store.

• Actor: A stateful process that executes, when invoked, the methods it exposes. Unlike a worker,
an actor is explicitly instantiated by a worker or a
driver. Like workers, actors execute methods serially.
Note that workers are stateless in that they do not
maintain local state across tasks. Assuming deterministic execution, invoking the same remote function with
the same arguments will return the same results no matter whether it is executed on the same worker or not. In

4.2.1

Global Control Store (GCS)

At its core, our system leverages a global control store
(GCS), which stores all up-to-date metadata and control
state information in the system. This includes (1) a speci5

Figure 5: Bottom-up distributed scheduler. Tasks are submitted bottom-up, from drivers and workers to a local scheduler and forwarded to the global scheduler only if needed (Section 4.2.2). The thickness of each arrow is proportional to its
request rate.

Figure 4: Ray’s architecture consists of two parts: an application layer and a system layer. The application layer implements
the API and the computation model described in Section 3, the
system layer implements task scheduling and data management
to satisfy the performance and fault-tolerance requirements.

schedulers (e.g., Canary [37]); (3) parallel scheduling,
where multiple global schedulers schedule tasks concurrently on all worker nodes (e.g., Sparrow [36]). Unfortunately, none of these approaches addresses Ray’s demands. Batch scheduling still requires the global scheduler to handle every task, which limits its scalability, hierarchical scheduling assumes the task graph is known
in advance (i.e., the graph is static), and parallel scheduling assumes that every global scheduler schedules independent jobs. In contrast, we require a highly scalable
scheduler to handle dynamic task graphs, possibly generated by a single job.
Like existing hierarchical scheduling solutions, we
employ a global scheduler and per-node local schedulers.
However, unlike prior solutions, tasks created on a node
are submitted to the node’s local scheduler first, not to a
global scheduler (Figure 5). A local scheduler schedules
tasks locally, unless the node is overloaded, or it cannot satisfy a task’s requirements (e.g., lacks GPU), or
a task’s inputs are remote. If a local scheduler doesn’t
schedule a task, it sends the task to the global scheduler. To determine the load, the local scheduler checks
the current length of its task queue. If the length exceeds some configurable threshold, it concludes that the
local node is overloaded. The value of this threshold enables the scheduling policy to span the continuum from
centralized, when all tasks are handed over to the global
scheduler, to decentralized, when all tasks are handled
locally.
Each local scheduler sends periodic heartbeats (e.g.,
every 100ms) to the GCS containing its load information. The GCS records this information and forwards
it to the global scheduler. Upon receiving a task, the
global scheduler uses the latest load information from every node along with the locations and sizes of the task’s
inputs (from the GCS’s object metadata) to decide which
node to assign the task to. If the global scheduler be-

fication of every task, (2) the code for every remote function, (3) the computation graph, (4) the current locations
of all objects, and (5) every scheduling event. The GCS
also provides publish-subscribe infrastructure to facilitate communication between components.
By storing and managing the entire control state in a
centralized fashion, the GCS enables every other component to be stateless. This not only simplifies the support
for fault tolerance (i.e., on failure, components restart
and read the latest state from the GCS), but also makes it
easy to horizontally scale every other component, as all
the state shared by the component’s replicas or shards is
accessible via the GCS.
To scale the GCS, we use sharding. Since we can associate pseudo-random IDs to virtually every data entry
in the GCS (e.g., objects, tasks, functions), it is relatively
easy to balance the load across multiple shards. To provide fault-tolerance, we use a hot replica for each shard.
Centralizing the system control information allows us
to easily build debugging, profiling, and visualization
tools on top of the GCS. The minimalist tools we’ve built
so far have already proven useful in our development.
4.2.2

Bottom-Up Distributed Scheduler

Many existing cluster computing frameworks, such
as Apache Spark [50], CIEL [32], Dryad [25] and
Hadoop [49], implement a centralized scheduler. While
this simplifies the design, it hurts scalability.
There are several approaches to improve scheduling
scalability: (1) batch scheduling, where the scheduler
submits tasks to worker nodes in batches to amortize
fixed overheads related to task submission (e.g., Drizzle [48]); (2) hierarchical scheduling, where a global
scheduler partitions the task graph across per-node local
6

N1

comes a bottleneck, we can instantiate more replicas and
have each local scheduler randomly pick a replica to send
its tasks to. This makes our scheduler architecture highly
scalable.
4.2.3

@ray.remote
def add(a, b):
return a + b

0

idC = add.remote(a, b)
c =ray.get(id)

ida
idb

1

@ray.remote
def add(a, b):
return a + b

Object Table

Object store

9

N1
N2

Object store

6

7

ida a

To minimize task latency, we implement an in-memory
distributed storage system to store the inputs and outputs
of every task. This allows workers and actors to share
data efficiently. On each node, we implement the object store via shared memory. This allows zero-copy data
sharing between tasks running on the same node. Additionally, we use Apache Arrow [1], an efficient memory
layout that is becoming the de facto standard in data analytics.
If a task’s inputs are not local, the inputs are replicated to the local object store on the same node before
execution. Tasks also write all outputs to the local object
store. Replication eliminates the potential bottleneck due
to hot data objects and minimizes task execution time as
a task only reads and writes data in local memory. This
increases throughput for computation-bound workloads,
a profile shared by many AI applications.
As with existing cluster computing frameworks, such
as Apache Spark [50], and Dryad [25], the object store
is limited to immutable data, which significantly simplifies the system design by obviating the need for complex
consistency protocols (as it avoids concurrent updates),
and by simplifying support for fault tolerance.
For simplicity, our object store does not build in support for distributed objects, that is, each object fits on a
single node. Distributed objects like large matrices or
trees can be implemented at a higher level (e.g., the application level) as collections of futures.

Worker

Function Table

@ray.remote
def add(a, b):
return a + b

In-Memory Distributed Object Store

N2

Global Control State (GCS)

Driver

3

Local Scheduler

2

Global Scheduler

8

ida a idb b

5

Local Scheduler

4

(a) Executing a task remotely
N1

N2

Global Control State (GCS)

Driver

Worker

Function Table

@ray.remote
def add(a, b):
return a + b

@ray.remote
def add(a, b):
return a + b

idC = add.remote(a, b)
c =ray.get(id)

7

Object Table

2

1
ida a idc c

@ray.remote
def add(a, b):
return a + b

5

ida
idb
idc

N1
N2
N2, N1

3
4
idc c ida a idb b

6
Local Scheduler

Global Scheduler

Local Scheduler

(b) Returning the result of a remote task

Figure 6: An end-to-end example that adds a and b and returns
c. Solid lines are data plane operations and dotted lines are
control plane operations. (a) The function add() is registered
with the GCS by node 1 (N1), invoked on N1, and executed
on N2. (b) N1 gets add()’s result using ray.get(). The Object
Table entry for c is created in step 4 and updated in step 6 after
c is copied to N1.

been lost. By walking backwards along data and stateful
edges we reach A10 which has no input.4 Thus, to reconstruct rollout12 , we need to re-instantiate the actor by
executing A10 , and then executing the methods A11 and
A12 in order.
Note that for any object whose lineage includes stateful edges, reconstruction will require reinstantiating the
actor (e.g., A10 ) and replaying a possibly long chain of
stateful edges (e.g., A11 , A12 , etc). Since actors are commonly used to wrap third-party simulators, which have
a finite lifetime, we expect these chains to be bounded.
However, we’ve also found actors to be useful for managing more general forms of state. To improve recovery
time in such cases, we checkpoint the actor’s state periodically and allow the actor to recover from checkpoints.
For low latency, we keep objects entirely in memory
and evict them as needed to disk using a least-recentlyused eviction policy.
In Section A, we briefly explain how our design satisfies the requirements outlined in Section 2.

Object reconstruction. Component failure may result in object loss, which Ray recovers from through
lineage re-execution. Ray tracks lineage by recording
task dependencies in the GCS during execution. This is
similar to the solution employed by other cluster computing systems like Apache Spark [50] and CIEL [32].
Also, like these systems, Ray assumes that objects are
immutable and operators (i.e., remote functions and actor methods) are deterministic. However, unlike these
systems, Ray adds support for stateful operator (i.e., actor) reconstruction. By integrating stateful edges directly
into the computation graph, we can leverage the same reconstruction mechanism for both remote functions and
actors.
To reconstruct a lost object, we walk backward along
data and stateful edges until we find tasks whose inputs
are all present in the object store. We then replay the
computation subgraph rooted at these inputs. Consider
the example in Figure 3(b), and assume that rollout12 has

4 Since it has no input, all of its inputs are trivially in the object store.

7

4.3

Putting Everything Together

The bottom-up distributed scheduler (Section 4.2.2) is
3.2KLoC and will undergo significant development as
we refine Ray’s scheduling policies. In this section,
we focus on the implementation details for achieving
the performance targets dictated by real-time AI applications: (a) scheduler performance, (b) object store performance, and (c) end-to-end system performance.

Figure 6 illustrates how Ray works end-to-end with a
simple example that adds two objects a and b, which
could be scalars or matrices, and returns result c. The remote function add() is automatically registered with the
GCS upon initialization and distributed to every worker
in the system (step 0 in Figure 6(a)).
Figure 6(a) shows the step-by-step operations triggered by a driver invoking add.remote(a, b), where a
and b are stored on nodes N1 and N2, respectively. The
driver submits add(a, b) to the local scheduler (step 1),
which forwards it to a global scheduler (step 2).5 Next,
the global scheduler looks up the locations of add(a, b)’s
arguments in the GCS (step 3) and decides to schedule
the task on node N2, which stores argument b (step 4).
The local scheduler at node N2 checks whether the local object store contains add(a, b)’s arguments (step 5).
Since the local store doesn’t have object a, it looks up
a’s location in the GCS (step 6). Learning that a is stored
at N1, N2’s object store replicates it locally (step 7). As
all arguments of add() are now stored locally, the local
scheduler invokes add() at a local worker (step 8), which
accesses the arguments via shared memory (step 9).
Figure 6(b) shows the step-by-step operations triggered by the execution of ray.get() at N1, and of add()
at N2, respectively. Upon ray.get(idc )’s invocation, the
driver checks the local object store for the value c, using
the future idc returned by add() (step 1). Since the local
object store doesn’t store c, it looks up its location in the
GCS. At this time, there is no entry for c, as c has not
been created yet. As a result, N1’s object store registers
a callback with the Object Table to be triggered when c’s
entry has been created (step 2). Meanwhile, at N2, add()
completes its execution, stores the result c in the local
object store (step 3), which in turn adds c’s entry to the
GCS (step 4). As a result, the GCS triggers a callback to
N1’s object store with c’s entry (step 5). Next, N1 replicates c from N2 (step 6), and returns c to ray.get() (step
7), which finally completes the task.
While this example involves a large number of RPCs,
in many cases this number is much smaller, as most tasks
are scheduled locally, and the GCS replies are cached by
the global and local schedulers.

5

Bottom-up distributed scheduler. We implement
both the local and global scheduler as event-driven,
single-threaded processes. Internally, local schedulers
maintain cached state for local object metadata, tasks
waiting for inputs, and tasks ready for dispatch to a
worker. As object dependencies become available, tasks
become ready for dispatch. Worker availability triggers
dispatch of as many tasks as possible under the node’s
capacity constraints.
The local scheduler sends periodic heartbeats (every
100ms) to the global schedulers, routed through the GCS
via a publish-subscriber mechanism, containing the dispatch queue length and resource availability. This enables the global scheduler to balance the load across
nodes.
Object store. Ray’s object store is also implemented
as a single-threaded event loop. It uses shared memory
so workers on the same node can read data without copying it. Objects are immutable. An object is only made
visible after a worker finishes creation. To minimize object creation overhead, the store pre-allocates a pool of
large memory-mapped files. We use a SIMD-like memory copy to maximize the throughput of copying data
from a worker to the object store’s shared memory. We
also parallelize computation of an object’s content hash,
which is used to detect non-deterministic computations.
Ray uses Apache Arrow [1] to achieve high performance
when serializing/deserializing Python objects.
Global control store. We implement Ray’s Global
Control Store (GCS) using one Redis [40] key-value
store per shard (Redis can be easily swapped with other
key-value stores). We shard the GCS tables by object
and task IDs to scale, and we replicate every shard for
fault tolerance. As we scale up the experiment, we distribute the shards across multiple nodes. Though our
GCS implementation uses multiple Redis servers, our
performance and fault tolerance requirements could also
be met by existing systems like RAMCloud [35].

Implementation

Ray is implemented in ≈ 40K lines of code (LoC), 72%
in C++ for the system layer, 28% in Python for the
application layer. The object store and our zero-copy
serialization libraries have been factored out as standalone projects that can be used independently of Ray.

Lastly, Ray’s monitor tracks system component liveness and reflects component failures in the GCS. Tasks
and objects on failed cluster nodes are marked as lost,
and objects are later reconstructed with lineage information, as necessary.

5 Note

that the N1 local scheduler could also decide to schedule the
task locally.

8

tasks (thousands)
10 20 30 40 50 60

number of nodes

5

10
nodes

15

20

Figure 8: Ray maintains balanced load. A driver on the first
node submits 100K tasks, which are rebalanced by the global
scheduler across the 21 available nodes.

a linear fashion, leveraging the GCS and bottom-up distributed
scheduler. Ray reaches 1 million tasks per second throughput
with 60 m4.16xlarge nodes and processes 100 million tasks in
under a minute. We omit x ∈ {70, 80, 90} due to cost.

20000

15

IOPS

15000

Evaluation

10

10000

5

5000

In this section, we demonstrate three key points. First,
we examine the scalability of the system as a whole as
well as the performance of its individual components
(Section 6.1). Second, we demonstrate robustness and
fault tolerance (Section 6.2). Third, we demonstrate that
Ray is a natural fit for reinforcement learning applications, both in terms of performance and ease of development (Section 6.3).
All experiments were run on
Amazon Web Services. The specific instance types are
reported below.

6.1

executed tasks
submitted tasks

100

Figure 7: End-to-end scalability of the system is achieved in

6

100
80
60
40
20
00

0

throughput (GB/s)

tasks per second (millions)

1.8
1.6
1.4
1.2
1.0
0.8
0.6
0.4
0.2
0.0

1KB 10KB 100KB 1MB 10MB 100MB 1GB
object size

0

Figure 9: Object store write throughput and IOPS. From a
single client, throughput exceeds 15GB/s (red) for large objects and 18K IOPS (cyan) for small objects on a 16 core instance (m4.4xlarge). It uses 8 threads to copy objects larger
than 0.5MB and 1 thread for small objects. Bar plots report
throughput with 1, 2, 4, 8, 16 threads. Results are averaged
over 5 runs.

Scalability and Performance

Object store performance. We track two metrics for
object store performance: IOPS (for small objects) and
write throughput (for large objects) in Figure 9. As object size increases, the write throughput from a single
client reaches 15GB/s. For larger objects, copying the
object from the client dominates the time spent on object
creation. For smaller objects, completion time is dominated by serialization overhead and IPC between the
client and object store. The object store peaks at 18K
IOPS, which corresponds to 56µs per operation.

End-to-end scalability. One of the key benefits of the
Global Control Store (GCS) is the ability to horizontally
scale the system. We evaluate this ability in this section. In Fig. 7, we benchmark an embarrassingly parallel
workload, increasing the cluster size on the x-axis. We
observe near-perfect linearity in progressively increasing
task throughput. Ray exceeds 1 million tasks per second
throughput at 60 nodes and continues to scale linearly
beyond 1.8 million tasks per second at 100 nodes. The
rightmost datapoint shows that Ray can process 100 million tasks in less than a minute (54s). Variability (shown
with black error bars) is minimal. As expected, increasing task duration reduces throughput proportionally to
mean task duration, but the overall scalability remains
linear.
The global scheduler’s primary responsibility is to
maintain balanced load throughout the system. In Figure 8, 100K tasks submitted on a single node are rebalanced across the available resources. Note that the
node where the load originates processes more tasks as it
maximizes utilization of the local node before forwarding tasks to the global scheduler.

6.2

Fault Tolerance

Recovering from object failures. In Figure 10, we
demonstrate Ray’s ability to transparently recover from
worker node failures and elastically scale. The driver
submits rounds of tasks where each task is dependent on
a task in the previous round. As worker nodes are killed
(at 25s, 50s, 100s), the surviving local schedulers automatically trigger reconstruction of the lost objects. During periods of reconstruction, the tasks originally submitted by the driver stall, since their dependencies cannot be
satisfied. However, the overall task throughput remains
9

Original tasks
Reexecuted tasks

0

50

100

150

Time since start (s)

200

method calls. The initial throughput is comparable to
that without checkpointing. After node failure, the majority of reconstruction is done by executing checkpoint
tasks to reconstruct the actor’s state (t = 210-270s). As
a result, only 500 tasks need to be re-executed, and new
method calls stall for 60s, versus 10K re-executions and
120s without checkpointing, respectively. In the future,
we hope to further reduce actor reconstruction time, e. g.
by allowing user annotations for read-only methods.
Overhead from GCS replication. To make the GCS
fault tolerant, we replicate each of the database shards.
When a client writes to one of the shards of the GCS,
it duplicates the writes to all replicas. For workloads in
which we artificially make the GCS the bottleneck by reducing the number of GCS shards, the overhead of twoway replication is less than 10%. In most real workloads
the slowdown is undetectable.

60
50
40
30
20
10
0

Number of nodes

Throughput (tasks/s)

2000
1500
1000
500
0

Figure 10: Fully transparent fault tolerance for distributed
tasks. The dashed line represents the number of nodes in the
cluster. Curves show the throughput of new tasks (cyan) and
re-executed tasks (red). The driver continually submits and retrieves rounds of 10000 tasks. Each task takes 100ms and depends on a task in the previous round. Each task has input and
output of size 100KB.

600
500
400
300
200
100
0

Original tasks
Reexecuted tasks

Throughput (tasks/s)

Throughput (tasks/s)

700

100 200 300 400

Time since start (s)
(a) without checkpointing

700
600
500

Original tasks
Reexecuted tasks
Checkpoint tasks

6.3

Given the diverse and demanding requirements of reinforcement learning applications described in Section 2, reinforcement learning algorithms today are implemented on top of special-purpose ad-hoc systems that
typically require substantial engineering effort to develop
and which do not generalize to other algorithms.
In this section, we implement two types of reinforcement learning algorithms in Ray and show that we are
able to match or outperform the performance of specialized systems built specifically for these algorithms. Furthermore, using Ray to distribute these algorithms over
clusters requires changing only a few lines of code in serial implementations of the algorithms.
In addition, we test Ray in a latency-sensitive setting
in which Ray is used to control a simulated robot under
varying real-time requirements.

400
300
200
100
0

RL Applications

100 200 300 400

Time since start (s)
(b) with checkpointing

Figure 11: Fully transparent fault tolerance for actor methods.
The driver continually submits tasks to the actors in the cluster.
At t = 200s, we kill 2 of the 10 nodes, causing 400 of the 2000
actors in the cluster to be recovered on the remaining nodes.

stable, fully utilizing the available resources until the lost
dependencies are reconstructed. Furthermore, as more
nodes are added back to the system at 210s, Ray is able
to fully recover to its initial throughput.
Recovering from actor failures. Next, we demonstrate Ray’s ability to transparently recover lost actors.
By encoding each actor’s method calls into the dependency graph, we can reuse the same object reconstruction mechanism as in Figure 10. The workload in Figure 11a demonstrates the extreme case where no intermediate actor state is saved. All previous method calls
for each lost actor must be re-executed serially (t = 210330s). Lost actors are automatically redistributed across
the available nodes, and throughput fully recovers after
reconstruction.
To improve reconstruction time for long-lived actors,
we provide transparent checkpointing of intermediate actor state. Figure 11b shows the same workload, but with
an automatic checkpoint task on each actor every 10

6.3.1

Evolution Strategies

To evaluate Ray on large-scale RL workloads, we implement the evolution strategies (ES) algorithm and compare to the reference implementation [39], which is a
special-purpose system built for this algorithm. It uses
a hierarchy of Redis servers as message buses and relies
on low-level multiprocessing libraries for sharing data.
As shown in Figure 12, a straightforward implementation on top of Ray is more scalable, scaling to 8192
physical cores, whereas the special-purpose system stops
running after 1024 cores. The Ray implementation runs
in a median time of 3.7 minutes, which is more than
twice as fast as the best published result (10 minutes).
The Ray implementation was also substantially simpler
to develop. Parallelizing a serial implementation using
Ray required modifying only 7 lines of code. In contrast,
10

Mean time to solve (minutes)

80

Reference ES
Ray ES

60

ing p2.16xlarge (GPU) and m4.16xlarge (high CPU) instances, each of which has 32 physical cores.
Ray’s API made it easy to take advantage of heterogeneous resources decreasing costs by a factor of 4.5
[2]. Ray tasks and actors can specify distinct resource
requirements, allowing CPU-only tasks to be scheduled
on cheaper high-CPU instances. In contrast, MPI applications often exhibit symmetric architectures, in which
all processes run the same code and require identical resources, in this case preventing the use of CPU-only machines for scale-out.
As shown in Figure 13, the Ray implementation outperforms the optimized MPI implementation in all experiments (hyperparameters listed in Section D) with a fraction of the GPUs. As with ES, we were able to parallelize
PPO using Ray with minimal changes to the structure of
the serial program.

40
20
0

256

512

x

1024
2048
Number of CPUs

x4096 x8192

Figure 12: Time to reach a score of 6000 in the Humanoid-

Mean time to solve (minutes)

v1 task [13]. The Ray ES implementation scales well to 8192
cores. The special-purpose system failed to run beyond 1024
cores. With 8192 cores, we achieve a median time of 3.7 minutes, which is over twice as fast as the best published result.
ES is faster than PPO (Section 6.3.2) on this benchmark but
exhibits greater runtime variance.

500

MPI PPO
Ray PPO

400

6.3.3

300

We show that Ray can meet soft real-time requirements
by controlling a simulated robot in real time. A Ray
driver runs the simulated robot and takes actions at fixed
time steps varying from 1 millisecond to 30 milliseconds
to simulate different real-time requirements. The driver
submits tasks that compute actions to be taken using a
policy that was trained offline. However, the actions are
only taken if they are received by the driver within the
relevant time step (otherwise the prior action is repeated).
Latency budgets for real robots are on the order of 10
milliseconds, and we find that even if we run the simulation faster than real time (using a 3 millisecond time
step), Ray is able to produce a stable walk. Table 3 shows
the fraction of tasks that did not arrive fast enough to be
used by the robot.

200
100
0

8x1

16x2

Controlling a Simulated Robot

32x4 64x8 128x16 256x32 512x64
CPUs x GPUs

Figure 13: Time to reach a score of 6000 in the Humanoid-v1
task [13]. The Ray PPO implementation outperforms a specialized MPI implementation [3] with fewer GPUs and at a fraction
of the cost. The MPI implementation required 1 GPU for every
8 CPUs, whereas the Ray version required at most 8 GPUs (and
never more than 1 GPU per 8 CPUs).

the reference implementation required several hundred
lines of code to develop a custom protocol for communicating tasks and data between workers and could not be
easily adapted to different algorithms or communication
patterns. We include pseudocode illustrating this point in
Section B.1.

Time budget (ms)
% actions dropped
Stable walk?

30
0
Yes

20
0
Yes

10
0
Yes

5
0
Yes

3
0.4
Yes

2
40
No

1
65
No

Table 3: Low-latency robot simulation results
6.3.2

Proximal Policy Optimization

To evaluate Ray on single node and small cluster RL
workloads, we implement Proximal Policy Optimization
(PPO) [41] in Ray and compare to a highly-optimized
reference implementation [3] that uses OpenMPI communication primitives.6 All experiments were run us-

7

Related Work

Dynamic task graphs. Ray is closely related to
CIEL [32]. They both support dynamic task graphs with
nested tasks, implement the futures abstraction, and provide lineage-based fault tolerance. However, they differ in two important aspects. First, Ray extends the task
model with an actor abstraction. Second, Ray employs
a fully distributed control plane and scheduler, instead
of relying on a single master. In addition, Ray adds the
ray.wait() method, employs an in-memory (instead of a

6 Both implementations use TensorFlow to define neural networks
but rely on the distributed frameworks (OpenMPI and Ray) for communication between processes. Though this algorithm could be implemented using TensorFlow alone, single-machine performance would be
slow due to the Python global interpreter lock and distributed versions
would look similar to MPI implementations (sending and receiving data
via queues).

11

file-based) object store, and extends an existing programming language (Python), while CIEL provides its own
scripting language (Skywriting). Ray is also closely related to Dask [38], which supports dynamic task graphs,
including a wait-like primitive, and employs the futures
abstraction in a Python environment. However, Dask
uses a centralized scheduler, doesn’t offer an actor-like
abstraction, and doesn’t provide fault tolerance.
Data flow systems. Popular dataflow systems, such
as MapReduce [18], Spark [51], and Dryad [25] have
widespread adoption for analytics and ML workloads,
but their computation model is more restrictive. Spark
and MapReduce implement the BSP execution model,
which assumes that tasks within the same stage perform
the same computation and take roughly the same amount
of time. Dryad relaxes this restriction but lacks support
for dynamic task graphs. Furthermore, none of these
systems provide an actor abstraction, nor implement a
distributed scalable control plane and scheduler. Finally,
Naiad [31] is a dataflow system that provides improved
scalability for some workloads, but only supports static
task graphs.
Actor systems. Orleans [14] provides a virtual actorbased abstraction. Actors are perpetual and their state
persists across invocations. For scaling out, Orleans also
allows multiple instances of an actor to run in parallel
when the actor operates on immutable state or has no
state. These stateless actors can act as tasks in Ray. However, unlike Ray, the Orleans developer must explicitly
checkpoint actor state and intermediate responses. In
addition, Orleans provides at-least-once semantics. In
contrast, Ray provides transparent fault tolerance and
exactly-once semantics, as each method call is logged in
the GCS and both arguments and results are immutable.
We find that in practice these limitations do not affect the
performance of our applications.
Erlang [9] and C++ Actor Framework (CAF) [16], two
other actor-based systems, also require the application to
explicitly handle fault tolerance. Also, Erlang’s global
state store is not suitable for sharing large objects such as
ML models, while CAF does not support data sharing.
Global control state and scheduling. The concept
of logically centralizing the control plane has been previously proposed in software defined networks (SDNs)
[15], distributed file systems (e.g., GFS [22]), resource
management (e.g., Omega [42]), and distributed frameworks (e.g., MapReduce [18], BOOM [7]), to name a
few. Ray draws inspiration from these pioneering efforts,
but provides significant improvements. In contrast with
SDNs, BOOM, and GFS which couple the control plane
data and computation, Ray decouples the storage of the
control plane information (e.g., GCS) from the logic implementation (e.g., schedulers). This allows both storage
and computation layers to scale independently, which is

key to achieving our scalability targets. Omega uses a
distributed architecture in which schedulers coordinate
via globally shared state. To this architecture, Ray adds
global schedulers to balance load across local schedulers,
and targets ms-level, not second-level, task scheduling.
Ray implements a unique distributed bottom-up
scheduler that is horizontally scalable, and can handle
dynamically constructed task graphs. Unlike Ray, most
existing cluster computing systems [18, 50, 32] use a
centralized scheduler architecture. While Sparrow [36]
is decentralized, its schedulers make independent decisions, limiting the possible scheduling policies, and all
tasks of a job are handled by the same global scheduler. Mesos [24] implements a two-level hierarchical
scheduler, but its top-level scheduler can be a bottleneck.
Canary [37] achieves impressive performance by having each scheduler instance handle a portion of the task
graph, but does not handle dynamic computation graphs.
Machine learning frameworks. TensorFlow [5] and
MXNet [17] target deep learning workloads and efficiently leverage both CPUs and GPUs. While they
achieve great performance for workloads consisting of
static DAGs of linear algebra operations, they have
limited support for more general workloads. TensorFlow Fold [29] provides some support for dynamic task
graphs, as well as MXNet through its internal C++ APIs,
but neither fully supports the ability to modify the DAG
during execution in response to task progress, task completion times, or faults. TensorFlow and MXNet in principle achieve generality by allowing the programmer to
simulate low-level message-passing and synchronization
primitives, but the pitfalls and user experience in this
case are similar to those of MPI. OpenMPI [21] can
achieve high performance, but it is relatively hard to program as it requires explicit coordination to handle heterogeneous and dynamic task graphs. Furthermore, it forces
the programmer to explicitly handle fault tolerance.

8

Discussion and Experiences

Since we released Ray several months ago, over one hundred people have downloaded and used it. Here we discuss our experience developing and using Ray, as well as
some of the feedback we’ve received from early users.
API. In designing the API, we have emphasized minimalism. Initially we started with a basic task abstraction. Later, we added the wait() primitive to accommodate rollouts with heterogeneous durations and the actor
abstraction to accommodate third-party simulators and
amortize the overhead of expensive initializations. While
the resulting API is relatively low-level, it has proven
both powerful and simple to use. Indeed, some teams
report instructing developers to first write serial implementations and then to parallelize them using Ray.
12

To illustrate this point, next we briefly describe our experience with two other algorithms: Asynchronous Advantage Actor Critic (A3C) and hyperparameter search.
A3C [30] is a state-of-the-art RL algorithm which
leverages asynchronous policy updates to significantly
improve training times over previous algorithms. To
scale out this algorithm, we use a simple hierarchical
scheme where multiple instances of A3C are trained
in parallel and periodically aggregated to form an improved model. Implementing hierarchical A3C in Ray
was straightforward, requiring 20 lines of Python code
to extend the non-hierarchical version. Furthermore,
this simple extension improved performance on the same
hardware by 30%.
We were able to implement a state-of-the-art hyperparameter search algorithm [28] in roughly 30 lines of
Python code using Ray. Ray’s support for nested tasks
was critical because multiple experiments had to be run
in parallel, and each experiment typically used parallelism internally. The wait() primitive allowed us to process the results of experiments in the order that they completed and to adaptively launch new ones. The actor abstraction allowed us to pause and resume stateful experiments based on the progress of other experiments (see
Section B.3). In contrast, most existing implementations
have to wait for all experiments in the round to complete,
leading to inefficient resource utilization.
Ray’s API is still a work in progress. Based on early
user feedback, we are considering enhancing the API to
include higher level primitives, such as simple aggregation and map. These could also inform scheduling decisions in the Ray system layer (Section 4.2).
Limitations. Given the workload generality, specialized optimizations are hard. For example, we must make
scheduling decisions without full knowledge of the computation graph. Scheduling optimizations in Ray might
require more complex runtime profiling. In addition,
storing lineage for each task requires the implementation
of garbage collection policies to bound storage costs in
the GCS, a feature we are actively developing.
Fault tolerance. We are often asked if fault tolerance
is really needed for AI applications. After all, due to
the statistical nature of many AI algorithms, one could
simply ignore failed rollouts. Based on our experience,
our answer is an unqualified “yes”. First, the ability to
ignore failures makes applications much easier to write
and reason about. Second, our particular implementation
of fault tolerance via deterministic replay dramatically
simplifies debugging as it allows us to easily reproduce
most errors. This is particularly important since, due to
their stochasticity, AI algorithms are notoriously hard to
debug. Third, fault tolerance helps save money since it
allows us to run on cheap resources like spot instances on
AWS. Furthermore, as workloads scale, we expect fault

tolerance to become even more important. Of course,
this comes at the price of some overhead. However, we
found this overhead to be minimal for our target workloads.
GCS and Horizontal Scalability. The GCS dramatically simplified Ray development and debugging. Basic failure handling and horizontal scaling for all other
components took less than a week to implement. The
GCS enabled us to query the entire system state while
debugging (instead of having to manually expose internal component state). This helped us find numerous bugs
and generally understand system behavior.
The GCS is instrumental to Ray’s horizontal scalability. In the experiments reported in Section 6.1, we were
able to scale the results by adding more shards whenever
the GCS became a bottleneck. The GCS also enables the
global scheduler to scale by simply adding more replicas.
While currently we are manually configuring the number
of GCS shards and global schedulers, we are planning to
develop adaptive algorithms in the future. Due to these
advantages, we believe that centralizing control state will
be a key design component of future distributed systems.

9

Conclusion

Emerging AI applications present challenging computational demands. To meet these demands, Ray introduces a global control store and a bottom-up distributed
scheduler. Together, this architecture implements dynamic task graph execution, which in turn supports both
a task-parallel and an actor programming model. This
programming flexibility is particularly important for RL
workloads, which produce tasks diverse in resource requirements, duration, and functionality. Our evaluation
demonstrates linear scalability past 1M tasks per second,
transparent fault tolerance, and substantial performance
improvements on several contemporary RL workloads.
Thus, Ray provides a powerful combination of flexibility, performance, and ease of use for the development of
future AI applications.

References
[1] Apache Arrow. https://arrow.apache.org/.
[2] EC2 Instance Pricing.
pricing/on-demand/.

https://aws.amazon.com/ec2/

[3] OpenAI Baselines: high-quality implementations of reinforcement learning algorithms. https://github.com/openai/
baselines.
[4] PyTorch: Tensors and dynamic neural networks in python with
strong gpu acceleration. http://pytorch.org/.
[5] A BADI , M., BARHAM , P., C HEN , J., C HEN , Z., DAVIS , A.,
D EAN , J., D EVIN , M., G HEMAWAT, S., I RVING , G., I SARD ,

13

M., ET AL . TensorFlow: A system for large-scale machine learning. In Proceedings of the 12th USENIX Symposium on Operating
Systems Design and Implementation (OSDI). Savannah, Georgia,
USA (2016).

[19] D ENNIS , J. B., AND M ISUNAS , D. P. A preliminary architecture for a basic data-flow processor. In Proceedings of the 2Nd
Annual Symposium on Computer Architecture (New York, NY,
USA, 1975), ISCA ’75, ACM, pp. 126–132.

[6] AGARWAL , A., B IRD , S., C OZOWICZ , M., H OANG , L., L ANG FORD , J., L EE , S., L I , J., M ELAMED , D., O SHRI , G., R IBAS ,
O., S EN , S., AND S LIVKINS , A. A multiworld testing decision
service. arXiv preprint arXiv:1606.03966 (2016).

[20] D UAN , Y., C HEN , X., H OUTHOOFT, R., S CHULMAN , J., AND
A BBEEL , P. Benchmarking deep reinforcement learning for continuous control. In Proceedings of the 33rd International Conference on International Conference on Machine Learning - Volume
48 (2016), ICML’16, JMLR.org, pp. 1329–1338.

[7] A LVARO , P., C ONDIE , T., C ONWAY, N., E LMELEEGY, K.,
H ELLERSTEIN , J. M., AND S EARS , R. BOOM Analytics: exploring data-centric, declarative programming for the cloud. In
Proceedings of the 5th European conference on Computer systems (2010), ACM, pp. 223–236.

[21] G ABRIEL , E., FAGG , G. E., B OSILCA , G., A NGSKUN , T.,
D ONGARRA , J. J., S QUYRES , J. M., S AHAY, V., K AMBADUR ,
P., BARRETT, B., L UMSDAINE , A., C ASTAIN , R. H., DANIEL ,
D. J., G RAHAM , R. L., AND W OODALL , T. S. Open MPI:
Goals, concept, and design of a next generation MPI implementation. In Proceedings, 11th European PVM/MPI Users’ Group
Meeting (Budapest, Hungary, September 2004), pp. 97–104.

[8] A NANTHANARAYANAN , G., K ANDULA , S., G REENBERG , A.,
S TOICA , I., L U , Y., S AHA , B., AND H ARRIS , E. Reining in
the outliers in map-reduce clusters using mantri. In Proceedings
of the 9th USENIX Conference on Operating Systems Design and
Implementation (Berkeley, CA, USA, 2010), OSDI’10, USENIX
Association, pp. 265–278.

[22] G HEMAWAT, S., G OBIOFF , H., AND L EUNG , S.-T. The Google
file system. In ACM SIGOPS operating systems review (2003),
vol. 37, ACM, pp. 29–43.

[9] A RMSTRONG , J., V IRDING , R., W IKSTR ÖM , C., AND
W ILLIAMS , M. Concurrent programming in ERLANG.

[23] G U *, S., H OLLY *, E., L ILLICRAP, T., AND L EVINE , S.
Deep reinforcement learning for robotic manipulation with asynchronous off-policy updates. In IEEE International Conference
on Robotics and Automation (ICRA 2017) (2017).

[10] B EATTIE , C., L EIBO , J. Z., T EPLYASHIN , D., WARD , T.,
WAINWRIGHT, M., K ÜTTLER , H., L EFRANCQ , A., G REEN , S.,
VALD ÉS , V., S ADIK , A., ET AL . Deepmind lab. arXiv preprint
arXiv:1612.03801 (2016).

[24] H INDMAN , B., KONWINSKI , A., Z AHARIA , M., G HODSI ,
A., J OSEPH , A. D., K ATZ , R., S HENKER , S., AND S TOICA ,
I. Mesos: A platform for fine-grained resource sharing in the
data center. In Proceedings of the 8th USENIX Conference on
Networked Systems Design and Implementation (Berkeley, CA,
USA, 2011), NSDI’11, USENIX Association, pp. 295–308.

[11] B ELLEMARE , M. G., NADDAF, Y., V ENESS , J., AND B OWL ING , M. The arcade learning environment: An evaluation platform for general agents. J. Artif. Intell. Res.(JAIR) 47 (2013),
253–279.

[25] I SARD , M., B UDIU , M., Y U , Y., B IRRELL , A., AND F ETTERLY, D. Dryad: Distributed data-parallel programs from
sequential building blocks. In Proceedings of the 2nd ACM
SIGOPS/EuroSys European Conference on Computer Systems
2007 (New York, NY, USA, 2007), EuroSys ’07, ACM, pp. 59–
72.

[12] B ERGSTRA , J., B REULEUX , O., BASTIEN , F., L AMBLIN ,
P., PASCANU , R., D ESJARDINS , G., T URIAN , J., WARDE FARLEY, D., AND B ENGIO , Y. Theano: A cpu and gpu math
compiler in python. In Proc. 9th Python in Science Conf (2010),
pp. 1–7.
[13] B ROCKMAN , G., C HEUNG , V., P ETTERSSON , L., S CHNEIDER ,
J., S CHULMAN , J., TANG , J., AND Z AREMBA , W. OpenAI
gym. arXiv preprint arXiv:1606.01540 (2016).

[26] J IA , Y., S HELHAMER , E., D ONAHUE , J., K ARAYEV, S., L ONG ,
J., G IRSHICK , R., G UADARRAMA , S., AND DARRELL , T.
Caffe: Convolutional architecture for fast feature embedding.
arXiv preprint arXiv:1408.5093 (2014).

[14] B YKOV, S., G ELLER , A., K LIOT, G., L ARUS , J. R., PANDYA ,
R., AND T HELIN , J. Orleans: Cloud computing for everyone.
In Proceedings of the 2nd ACM Symposium on Cloud Computing
(2011), ACM, p. 16.

[27] J ORDAN , M. I., AND M ITCHELL , T. M. Machine learning:
Trends, perspectives, and prospects. Science 349, 6245 (2015),
255–260.

[15] C ASADO , M., F REEDMAN , M. J., P ETTIT, J., L UO , J., M C KEOWN , N., AND S HENKER , S. Ethane: Taking control of the enterprise. SIGCOMM Comput. Commun. Rev. 37, 4 (Aug. 2007),
1–12.

[28] L I , L., JAMIESON , K., D E S ALVO , G., ROSTAMIZADEH ,
A., AND TALWALKAR , A. Hyperband: A novel banditbased approach to hyperparameter optimization. arXiv preprint
arXiv:1603.06560 (2016).

[16] C HAROUSSET, D., S CHMIDT, T. C., H IESGEN , R., AND
W ÄHLISCH , M. Native actors: A scalable software platform for
distributed, heterogeneous environments. In Proceedings of the
2013 workshop on Programming based on actors, agents, and
decentralized control (2013), ACM, pp. 87–96.

[29] L OOKS , M., H ERRESHOFF , M., H UTCHINS , D., AND N ORVIG ,
P. Deep learning with dynamic computation graphs. arXiv
preprint arXiv:1702.02181 (2017).
[30] M NIH , V., BADIA , A. P., M IRZA , M., G RAVES , A., L ILLI CRAP, T. P., H ARLEY, T., S ILVER , D., AND K AVUKCUOGLU ,
K. Asynchronous methods for deep reinforcement learning. In
International Conference on Machine Learning (2016).

[17] C HEN , T., L I , M., L I , Y., L IN , M., WANG , N., WANG , M.,
X IAO , T., X U , B., Z HANG , C., AND Z HANG , Z. MXNet: A
flexible and efficient machine learning library for heterogeneous
distributed systems. In NIPS Workshop on Machine Learning
Systems (LearningSys’16) (2016).

[31] M URRAY, D. G., M C S HERRY, F., I SAACS , R., I SARD , M.,
BARHAM , P., AND A BADI , M. Naiad: A timely dataflow system.
In Proceedings of the Twenty-Fourth ACM Symposium on Operating Systems Principles (New York, NY, USA, 2013), SOSP ’13,
ACM, pp. 439–455.

[18] D EAN , J., AND G HEMAWAT, S. MapReduce: Simplified data
processing on large clusters. Commun. ACM 51, 1 (Jan. 2008),
107–113.

14

[32] M URRAY, D. G., S CHWARZKOPF, M., S MOWTON , C., S MITH ,
S., M ADHAVAPEDDY, A., AND H AND , S. CIEL: A universal execution engine for distributed data-flow computing. In Proceedings of the 8th USENIX Conference on Networked Systems Design and Implementation (Berkeley, CA, USA, 2011), NSDI’11,
USENIX Association, pp. 113–126.

[47] VAN D EN B ERG , J., M ILLER , S., D UCKWORTH , D., H U , H.,
WAN , A., F U , X.-Y., G OLDBERG , K., AND A BBEEL , P. Superhuman performance of surgical tasks by robots using iterative learning from human-guided demonstrations. In Robotics
and Automation (ICRA), 2010 IEEE International Conference on
(2010), IEEE, pp. 2074–2081.

[33] N G , A., C OATES , A., D IEL , M., G ANAPATHI , V., S CHULTE ,
J., T SE , B., B ERGER , E., AND L IANG , E. Autonomous inverted helicopter flight via reinforcement learning. Experimental
Robotics IX (2006), 363–372.

[48] V ENKATARAMAN , S., PANDA , A., O USTERHOUT, K., G H ODSI , A., A RMBRUST, M., R ECHT, B., F RANKLIN , M., AND
S TOICA , I. Drizzle: Fast and adaptable stream processing at
scale. In Proceedings of the Twenty-Sixth ACM Symposium on
Operating Systems Principles (2017), SOSP ’17, ACM.

[34] N ISHIHARA , R., M ORITZ , P., WANG , S., T UMANOV, A.,
PAUL , W., S CHLEIER -S MITH , J., L IAW, R., N IKNAMI , M.,
J ORDAN , M. I., AND S TOICA , I. Real-time machine learning:
The missing pieces. In Workshop on Hot Topics in Operating
Systems (2017).

[49] W HITE , T. Hadoop: The Definitive Guide. O’Reilly Media, Inc.,
2012.
[50] Z AHARIA , M., C HOWDHURY, M., DAS , T., DAVE , A., M A ,
J., M C C AULEY, M., F RANKLIN , M. J., S HENKER , S., AND
S TOICA , I. Resilient distributed datasets: A fault-tolerant abstraction for in-memory cluster computing. In Proceedings of the
9th USENIX conference on Networked Systems Design and Implementation (2012), USENIX Association, pp. 2–2.

[35] O USTERHOUT, J., G OPALAN , A., G UPTA , A., K EJRIWAL , A.,
L EE , C., M ONTAZERI , B., O NGARO , D., PARK , S. J., Q IN , H.,
ROSENBLUM , M., ET AL . The RAMCloud storage system. ACM
Transactions on Computer Systems (TOCS) 33, 3 (2015), 7.
[36] O USTERHOUT, K., W ENDELL , P., Z AHARIA , M., AND S TO ICA , I. Sparrow: Distributed, low latency scheduling. In Proceedings of the Twenty-Fourth ACM Symposium on Operating
Systems Principles (New York, NY, USA, 2013), SOSP ’13,
ACM, pp. 69–84.

[51] Z AHARIA , M., X IN , R. S., W ENDELL , P., DAS , T., A RM BRUST, M., DAVE , A., M ENG , X., ROSEN , J., V ENKATARA MAN , S., F RANKLIN , M. J., G HODSI , A., G ONZALEZ , J.,
S HENKER , S., AND S TOICA , I. Apache Spark: A unified engine for big data processing. Commun. ACM 59, 11 (Oct. 2016),
56–65.

[37] Q U , H., M ASHAYEKHI , O., T EREI , D., AND L EVIS , P. Canary:
A scheduling architecture for high performance cloud computing.
arXiv preprint arXiv:1602.01412 (2016).
[38] ROCKLIN , M. Dask: Parallel computation with blocked algorithms and task scheduling. In Proceedings of the 14th Python in
Science Conference (2015), K. Huff and J. Bergstra, Eds., pp. 130
– 136.
[39] S ALIMANS , T., H O , J., C HEN , X., AND S UTSKEVER , I. Evolution strategies as a scalable alternative to reinforcement learning.
arXiv preprint arXiv:1703.03864 (2017).
[40] S ANFILIPPO , S. Redis: An open source, in-memory data structure store. https://redis.io/, 2009.
[41] S CHULMAN , J., W OLSKI , F., D HARIWAL , P., R ADFORD , A.,
AND K LIMOV, O. Proximal policy optimization algorithms.
arXiv preprint arXiv:1707.06347 (2017).
[42] S CHWARZKOPF, M., KONWINSKI , A., A BD -E L -M ALEK , M.,
AND W ILKES , J. Omega: Flexible, scalable schedulers for large
compute clusters. In Proceedings of the 8th ACM European Conference on Computer Systems (New York, NY, USA, 2013), EuroSys ’13, ACM, pp. 351–364.
[43] S ILVER , D., H UANG , A., M ADDISON , C. J., G UEZ , A.,
S IFRE , L., VAN D EN D RIESSCHE , G., S CHRITTWIESER , J.,
A NTONOGLOU , I., PANNEERSHELVAM , V., L ANCTOT, M.,
ET AL . Mastering the game of Go with deep neural networks
and tree search. Nature 529, 7587 (2016), 484–489.
[44] S UTTON , R. S., AND BARTO , A. G. Reinforcement Learning:
An Introduction, vol. 1. MIT press Cambridge, 1998.
[45] T ODOROV, E., E REZ , T., AND TASSA , Y. Mujoco: A physics
engine for model-based control. In Intelligent Robots and
Systems (IROS), 2012 IEEE/RSJ International Conference on
(2012), IEEE, pp. 5026–5033.
[46] VALIANT, L. G. A bridging model for parallel computation.
Communications of the ACM 33, 8 (1990), 103–111.

15

A

Satisfying the Requirements

In this section, we briefly explain how our design satisfies the requirements outlined in Section 2.
Flexibility: Ray extends the already general dynamic task model [32], by adding the ray.wait() primitive to efficiently handle tasks whose completion times are not known in advance, and the actor abstraction to handle third part
simulators and amortize expensive setups. Finally, Ray supports heterogeneous resources, such as GPUs.
Performance: Table 4 summarizes techniques for scaling each component and the associated overhead. The last
column shows the average number of requests/sec that each component should handle as a function of system and
workload parameters, where N is the number of nodes, s the number of GCS shards, g the number of global schedulers,
w the average number of tasks/sec generated by a node, and ε, the percentage of tasks submitted to the global scheduler.
For instance, on average, the local scheduler handles a number of requests proportional to w tasks/sec, where α1 ,
includes assigning the task to a worker or sending/receiving the task to/from the global scheduler. Similarly, the global
scheduler handles a number of request proportional to the average number of tasks/sec it receives (εNw/g) times α, a
factor that includes querying the GCS for task’s inputs’ locations; h represents the number of heartbeats/sec.
Note that given the cluster size, N, and the average load generated by a node, w, we could pick the number of GCS
shards, s and global schedulers, s to bound the load on each shard or global scheduler. This makes both GCS and the
global scheduler horizontally scalable.
Ease of development: Ray handles a variety of component failures, relieving the developer from writing complex software to handle failures. Ray achieves fault tolerance by using a variety of techniques, including stateless components,
replication and, replay, summarized in Table 5. This also enables a user to run the system on cheap infrastructures
where nodes may be preempted (e.g., AWS spot instances). The API shown in Table 2, while low-level, has proven to
be a good fit for the RL applications in our experience. The fact that the API is provided in Python, the most popular
language in the AI community, has been a big plus.
Component
Local Sched.
Object Store

Scaling Techniques
By default, use 1 per node.
By default, use 1 per node.

GCS

Use more shards (s).

Global Sched.

Use more replicas (g).

Data Object

Replicate objects to local
node before task execution.
Use nested tasks; spawn
tasks from workers.

Driver

Requests/sec
α1 w
α2 w
 

β Ns w + h Ng
s
 
α ε Ng s + h
N/A
N/A

Table 4: Achieving scalability in Ray. By increasing the number of GCS shards s and the number of global scheduler replicas
g, we can reduce the load on each GCS shard and global scheduler replica to a constant. N is the total number of nodes, w avg.
number of tasks/node/sec, h avg. number of node heartbeats/sec, and α, α1 , α2 , β , and ε are constants.

Component
Global Control Store (GCS)
Local and Global
Schedulers, Workers
Object Store, Actor
Driver

Description
Replicate each shard.
Stateless processes; restart and
read state from the GCS.
Use lineage stored in the GCS to
reconstruct objects (Section 4.2.3).
Restart it; can reuse objects already
present in the Object Store.

Table 5: Techniques for achieving fault tolerance in Ray

16

B

Ease of Development

B.1

Evolution Strategies

In this section, we give code examples to illustrate why the evolution strategies algorithm from Section 6.3.1 is so easy
to implement and customize on top of Ray and so difficult in the special-purpose reference system that we compare
to. Suppose we wish to parallelize the following serial Python code.
1
2
3
4
5
6
7

def f ():
return result
def g ( N ):
return [ f () for _ in range ( N )]
results = [ g ( N ) for _ in range ( M )]

Attempting to implement this code in MPI, distributed TensorFlow, or many specialized systems would involve
writing a single program that is run by all workers. As shown in the code below, programs would need to branch based
on the role of that worker and would likely only work for predetermined values of M and N.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

if WORKER_INDEX == 0:
results = []
for i in range ( M ):
child_index = 1 + i
results . append ( RECEIVE ( child_index ) , i )
elif 1 <= WORKER_INDEX < M :
i n t e rmediate_results = []
for i in range ( N ):
child_index = M + 1 + WORKER_INDEX + M * i
intermediate_results . append ( RECEIVE ( child_index ))
SEND ( intermediate_results , 0)
else :
result = f ()
parent_index = ( WORKER_INDEX - M - 1) % M
SEND ( result , parent_index )

This is complex for a number of reasons. Each worker has to decide what tasks to execute and has to explicitly
dictate logic for optimizations such as batching, which normally a task scheduler could take care of. Secondly, allowing
variable values of M and N would substantially complicate the program, but hard coding the values M and N will make
it even more difficult to extend the code to deeper hierarchies in the future.
Finally, this violates the lockstep assumptions that these frameworks make and would sacrifice performance by
restricting the algorithm to a predefined scheduling layout.
In contrast, the Ray version is clean, simple, and extensible.
1
2
3
4
5
6
7
8
9

@ray.remote
def f ():
return result
@ray.remote
def g ( N ):
return ray.get([ f.remote() for _ in range ( N )])
results = ray.get([ g.remote( N ) for _ in range ( M )])

We believe that Ray’s API will allow developers to easily develop more complex distributed schemes.

17

B.2

Tree Reduce

Consider the scenario where one wants to perform an aggregation operation on multiple data points in a distributed
fashion. With Ray, the programmer would be able to easily aggregate data in an efficient manner.
1
2
3
4
5
6
7
8

dataset = [ data1 , data2 , data3 , data4 , data5 , data6 , data7 , data8 ]
dataset_1 = [ aggregate . remote ( dataset [ i ] , dataset [ i +1])
for i in range (0 , 8 , step =2)]
dataset_2 = [ aggregate . remote ( dataset_1 [ i ] , dataset_1 [ i +1])
for i in range (0 , 4 , step =2)]
result = ray.get( aggregate . remote ( dataset_2 [0] , dataset_2 [1]))

Note that this implementation can be written in a more concise fashion.
1
2
3
4

while len ( data ) > 1:
data = data [2:] + [ aggregate.remote( data [0] , data [1])]
result = ray.get( data [0])

In contrast, systems such as MPI and Spark provide specialize tree-reduction operators (i.e. MPI Allreduce and
rdd.treeAggregate), since hierarchical computations are not easily expressed in their APIs.

18

B.3

Hyperparameter Search

Ray enables developers to build hyperparameter search programs in a readable and simple manner.
In this example, assume we have an experiment class with the following interface. We add a single decorator to the
class to convert it into an actor.
1
2
3
4
5
6
7
8

@ray.remote
class Experiment ():
def __init__ ( self , hyperparameters ):
# ...
def training_step ( self ):
# ...
return accuracy

Assume also that we have these helper functions predefined.
is promising(results) returns True if the model is doing well and False otherwise. In practice, this function will
may more information than the current result.
generate hyperparameters defines a queue for many hyperparameter configurations, enumerating the space that
the programmer wants to search over.
1
2
3
4
5

def is_promising ( results ):
return True
def g e n e r a t e_ h y p er p a r am e t e rs ():
return hyperparameter_queue

Here is a simple and straightforward implementation of a hyperparameter search program. Notice that more complicated management schemes can be implemented fairly easily, but we chose to leave those out in favor of readability.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

hp_queue = g e n e ra t e _ hy p e r pa r a m et e r s ()
experiment_list = [ Experiment.remote( next ( hp_queue )) for i in range (10)]
paused = []
current_exps = { exp . training_step.remote(): exp for exp in experiment_list }
while True :
ex pe riment_futures = current_exps . keys ()
[ ready_id ] , remaining = @ray.wait( experiment_futures , num_returns =1)
current_exp = current_exps [ ready_id ]
current_accuracy = ray.get( ready_id )
if is_promising ( current_accuracy ):
current_exps [ current_exp . training_step.remote()] = current_exp
else :
paused . append ( current_exp )
new_exp = Experiment.remote( next ( hp_queue ))
current_exps [ new_exp . training_step.remote()] = new_exp

19

C

Hyperparameters for Evolution Strategies
Hyperparameter
Noise standard deviation
Adam stepsize
L2 coefficient
Episodes per batch

D

Value
0.02
0.01
0.005
5000

Hyperparameters for Proximal Policy Optimization
Hyperparameter
Timesteps per batch
SGD minibatch size
SGD epochs per iteration
Adam stepsize
PPO clip param
GAE parameter (λ )
Discount (γ)

20

Value
320000
32768
20
1e-4
0.2
0.95
0.995

