Ray’s remote functions are surprisingly not just glorified background jobs, but full-fledged, first-class citizens in a distributed system.
Let’s see Ray in action. Imagine you have a computationally intensive task, like calculating Fibonacci numbers, and you want to run it on a remote machine.
import ray
# Initialize Ray
ray.init()
# Define a remote function
@ray.remote
def fib(n):
if n < 2:
return n
return fib.remote(n-1) + fib.remote(n-2)
# Invoke the remote function
result_ref = fib.remote(10)
# Retrieve the result
result = ray.get(result_ref)
print(f"Fibonacci(10) = {result}")
ray.shutdown()
When you run this, ray.init() spins up a local Ray cluster (or connects to an existing one). The @ray.remote decorator transforms the fib function. Calling fib.remote(10) doesn’t execute fib(10) immediately; instead, it returns an ObjectRef, a future-like object representing the promise of a result. The actual computation happens asynchronously on a Ray worker. ray.get(result_ref) then blocks until the result is available and fetches it. Notice how fib.remote(n-1) and fib.remote(n-2) are themselves remote calls, demonstrating how Ray functions can recursively launch other remote tasks, building a directed acyclic graph (DAG) of computations.
Now, let’s talk about actors. While remote functions are stateless, actors are stateful, essentially distributed objects. Think of them as persistent workers that can maintain internal state and respond to method calls.
import ray
ray.init()
@ray.remote
class Counter:
def __init__(self):
self.value = 0
def increment(self):
self.value += 1
return self.value
def get_value(self):
return self.value
# Create an actor instance
counter_actor = Counter.remote()
# Call actor methods
obj_ref1 = counter_actor.increment.remote()
obj_ref2 = counter_actor.increment.remote()
current_value_ref = counter_actor.get_value.remote()
# Retrieve results
results = ray.get([obj_ref1, obj_ref2, current_value_ref])
print(f"Increment results: {results[:2]}")
print(f"Final value: {results[2]}")
ray.shutdown()
Here, Counter.remote() instantiates the Counter class as a remote actor. Each call to increment.remote() or get_value.remote() sends a message to the actor. The actor processes these messages sequentially, maintaining its self.value. The key difference from remote functions is that the state (self.value) persists across method calls. This allows for building complex distributed applications where components need to maintain information over time, like managing shared resources or coordinating distributed processes. Ray handles the serialization, network communication, and scheduling of these actor method calls.
The surprising aspect of Ray’s design is how it unifies these two paradigms—stateless tasks and stateful actors—under a single, consistent API. You define them using decorators (@ray.remote) and invoke them using .remote(), always getting back ObjectRefs. This uniform interface simplifies distributed programming significantly. You don’t need separate libraries or frameworks for different types of distributed computations. Ray’s scheduler is responsible for placing tasks and actors on available nodes, managing dependencies, and handling failures.
A common point of confusion is understanding the lifecycle of an ObjectRef. When you call some_function.remote(), you get an ObjectRef immediately. This doesn’t mean the computation has finished, or even started. It simply signifies that a task has been scheduled. The actual execution happens asynchronously. If you call ray.get() on an ObjectRef that hasn’t completed, your program will block until the result is ready. However, if you have multiple ObjectRefs, you can pass them all to a single ray.get() call, which will efficiently wait for all of them to complete and return their results in a list, preserving the order of the input ObjectRefs. This is crucial for performance, as it allows Ray to fetch results in batches.
The next step is understanding how to manage the resources these tasks and actors consume.