Ray Tasks and Actors in Python

Ray Core provides two useful primitives for parallel and distributed Python. A remote function runs as a stateless task. A remote class creates a stateful actor. Calls to both return object references immediately, allowing work to be submitted before any result is requested.

The example below models six coffee orders because the work is easy to follow. It runs on one machine and demonstrates Ray's scheduling model. It does not establish that Ray is the best choice for this workload or that the same speedup will continue across a cluster.

The example was tested with Ray 2.52.1, so that version is used here for reproducibility.

Starting a Local Ray Runtime

python -m pip install "ray==2.52.1"

With no existing Ray runtime, ray.init() starts one locally. The num_cpus value declares four logical CPU resources. These resources guide scheduling but do not reserve or isolate four physical CPU cores[0]. Each task in the example explicitly requests one logical CPU, so at most four of those tasks can run at once.

Tasks and Object References

Calling a remote function schedules work and returns an ObjectRef. Calling ray.get() blocks until the referenced result is available. Submitting every task before calling ray.get() gives the scheduler an opportunity to run independent tasks concurrently[1].

The same make_drink function is used for the sequential and Ray executions. Prices are stored as integer cents to avoid floating-point arithmetic for money.

Actors and State

An actor is a dedicated worker process that retains state between method calls. Methods on the synchronous actor below execute one at a time. That makes the two counter updates safe within this actor, although async or threaded actors can execute calls concurrently[2].

The actor requests one logical CPU explicitly. It is created only after the task timing has finished, so it does not reduce the four task slots used in the comparison.

import time
import ray

ray.init(num_cpus=4, include_dashboard=False)

orders = [
    {"drink": "Cappuccino", "price_cents": 450},
    {"drink": "Latte", "price_cents": 500},
    {"drink": "Espresso", "price_cents": 300},
    {"drink": "Americano", "price_cents": 350},
    {"drink": "Mocha", "price_cents": 550},
    {"drink": "Flat White", "price_cents": 475},
]


def make_drink(order):
    time.sleep(order.get("delay_seconds", 1))
    return order


make_drink_remote = ray.remote(num_cpus=1)(make_drink)


@ray.remote(num_cpus=1)
class CashRegister:
    def __init__(self):
        self.total_revenue_cents = 0
        self.drinks_sold = 0

    def record_sale(self, price_cents):
        self.total_revenue_cents += price_cents
        self.drinks_sold += 1

    def get_summary(self):
        return {
            "drinks_sold": self.drinks_sold,
            "total_revenue_cents": self.total_revenue_cents,
        }


# Give Ray an opportunity to start four workers before measuring.
warmup_order = {"drink": "Warm-up", "price_cents": 0, "delay_seconds": 0.1}
ray.get([make_drink_remote.remote(warmup_order) for _ in range(4)])

sequential_start = time.perf_counter()
sequential_drinks = [make_drink(order) for order in orders]
sequential_seconds = time.perf_counter() - sequential_start

parallel_start = time.perf_counter()
drink_refs = [make_drink_remote.remote(order) for order in orders]
parallel_drinks = ray.get(drink_refs)
parallel_seconds = time.perf_counter() - parallel_start

register = CashRegister.remote()
sale_refs = [
    register.record_sale.remote(drink["price_cents"])
    for drink in parallel_drinks
]
ray.get(sale_refs)
summary = ray.get(register.get_summary.remote())

print(f"Sequential: {sequential_seconds:.2f} seconds")
print(f"Ray tasks:  {parallel_seconds:.2f} seconds")
print(summary)

ray.shutdown()

The sequential run takes about six seconds. Four task slots process the same six one-second waits in two batches, so the Ray run takes about two seconds after warm-up. Exact times depend on the machine and Ray version.

What the Timing Means

time.sleep() makes the scheduling pattern visible, but it represents waiting rather than CPU-intensive computation. Threads or asyncio would usually be simpler for a small workload that only waits. Ray becomes more useful when independent work is expensive, must use different resources, or needs to run across processes or machines.

Task submission, serialization, object transfer, and scheduling all add overhead. Small tasks can therefore become slower when distributed. Ray recommends batching work when individual tasks are too fine-grained[3].

Moving Beyond One Machine

A multi-node deployment requires a running cluster and a connection through ray.init(address=...) or the Ray Jobs API. Dependencies, files, storage, resource requests, and data movement must also be configured for the cluster. The task and actor APIs remain familiar, but deployment is not automatic.

Actor state is not durable by default. A crashed actor is not restarted unless max_restarts is configured, and restarting it reruns the constructor rather than restoring application state. Important state requires checkpointing or external storage[4].