Distributed Computing and Actors in Python with Ray

Ray is an open-source framework designed to scale Python applications seamlessly from a single machine to a multi-node cluster. This article explores how Ray facilitates distributed computing through stateless tasks and stateful actor-based parallelism, examining its underlying architecture, the @ray.remote decorator, its distributed object store, and how it handles concurrency and scheduling to simplify scalable system design.

The Core Abstractions: Tasks and Actors

Python's standard concurrency tools often struggle with distributed multi-node environments and CPU-bound workloads due to the Global Interpreter Lock (GIL) and serialization overhead. Ray resolves this by introducing two primary primitives that map cleanly onto distributed infrastructure: Tasks and Actors.

1. Ray Tasks (Stateless Distributed Functions)

Ray tasks represent stateless, asynchronous computations executed across worker nodes.

• Implementation: Any standard Python function is converted into a distributed task by decorating it with @ray.remote.
• Asynchronous Execution: Invoking a task using the .remote() method immediately returns an ObjectRef (a future) rather than blocking for the result.
• Dynamic Computation Graphs: Tasks can spawn other tasks dynamically, allowing fine-grained task graphs that scale automatically without centralized bottlenecks.

import ray

ray.init()

@ray.remote
def process_data(batch):
    return [x * 2 for x in batch]

# Non-blocking asynchronous execution
futures = [process_data.remote([i]) for i in range(10)]
results = ray.get(futures)

2. Ray Actors (Stateful Distributed Classes)

While tasks handle stateless data processing, stateful workloads require actors. Ray implements the Actor model by wrapping Python classes with @ray.remote.

• State Retention: An actor is an instantiated process with dedicated resources that retains state across multiple method calls.
• Message Passing: Invoking a method on an actor (actor_handle.method.remote()) submits a message to the actor's internal queue.
• Sequential Execution: Within a single actor instance, methods execute sequentially, preventing race conditions on internal state without requiring explicit locks.
• Parallel Actor Pools: Multiple actor instances can run concurrently across different cores or nodes, enabling actor-based parallelism for simulations, stateful streaming, or machine learning model serving.

@ray.remote
class Counter:
    def __init__(self):
        self.count = 0

    def increment(self):
        self.count += 1
        return self.count

counter = Counter.remote()
future = counter.increment.remote()
print(ray.get(future))  # Output: 1

Architectural Enablers of Ray's Performance

Ray's ability to coordinate tasks and actors efficiently relies on its distributed engine:

Distributed Object Store (Plasma)

Ray runs an in-memory object store on each cluster node using shared memory. When tasks return large data structures (such as NumPy arrays), the data is stored in the local object store. Other workers on the same machine access this data using zero-copy deserialization via Apache Arrow, drastically reducing memory overhead. For inter-node operations, Ray manages data transfers automatically.

Decentralized Scheduling

Instead of relying on a single master scheduler that can become saturated, Ray uses a two-level distributed scheduling system. Each node runs a local scheduler (raylet) that assigns work locally when resources allow. If local resources are exhausted, tasks and actor placement requests are forwarded to global schedulers, ensuring high-throughput task scheduling at scale.

Unified Resource Management

Ray permits fine-grained specification of hardware requirements (such as CPU, GPU, and custom resources) directly in the decorator:

@ray.remote(num_cpus=2, num_gpus=0.5)
def gpu_task():
    pass

The scheduler automatically places the task or actor on a node with the required capacity, enabling heterogeneous resource sharing across a cluster.

Summary

Ray bridges the gap between simple Python scripting and complex distributed engineering. By transforming standard functions into asynchronous distributed tasks and classes into stateful distributed actors, Ray provides a unified programming model that eliminates the boilerplate of distributed systems, allowing developers to scale parallel workloads with minimal architectural changes.