Skip to content

Chapter 01: Remote Tasks, ObjectRefs & Asynchronous Execution

  • Topic Focus • @ray.remote Functions, Futures, Asynchronous Concurrency, and Batch Evaluation
  • Interactive Challenges • 6 Hands-on Exercises
  • Launch Playground in Wasm →

1. Architectural Overview & Control Plane Mechanics

In Ray, Tasks represent stateless, asynchronous functions executed across worker processes. When a task is invoked using .remote(), Ray immediately returns an ObjectRef (a future) without blocking the calling thread.

flowchart LR
    Driver["Driver Program<br/><code>compute.remote(x)</code>"] -->|"1. Request Worker Lease"| Raylet["Local Raylet<br/>(Scheduler)"]
    Driver -.->|"Immediate Future Return"| ObjectRef["ObjectRef [Future]"]
    Raylet -->|"2. Assign Leased Worker"| Worker["Worker Process<br/>(Executes Task)"]
    Worker -->|"3. Zero-Copy Store"| Plasma["Plasma Object Store<br/>(/dev/shm)"]
    Driver -->|"4. ray.get(ref)"| Plasma

    style Driver fill:#1e293b,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    style Raylet fill:#1e293b,stroke:#f59e0b,stroke-width:2px,color:#f8fafc
    style Worker fill:#0f172a,stroke:#34d399,stroke-width:2px,color:#f8fafc
    style Plasma fill:#1e1e38,stroke:#c084fc,stroke-width:2px,color:#f8fafc
    style ObjectRef fill:#0f172a,stroke:#818cf8,stroke-width:1px,stroke-dasharray: 5 5,color:#f8fafc

Diagram Walkthrough & Core Concepts: - Non-blocking Dispatch: The driver requests a worker lease from the local Raylet and immediately receives an ObjectRef future without waiting for the computation to finish. - Decoupled Worker Execution: The Raylet schedules the task on an available worker process, which executes the Python function in isolation. - Shared Memory Object Store: Worker return values are written directly to the local node's Plasma Object Store in shared memory (/dev/shm), enabling zero-copy deserialization when ray.get() is resolved.

2. Annotated Python Code Anatomy & API Reference

Below is a production pattern for remote tasks, parallel batching, and async wait operations:

import ray
from typing import List

# 1. Initialize Ray cluster connection or local embedded engine
ray.init(ignore_reinit_error=True)

# 2. Decorate stateless function with @ray.remote
@ray.remote(num_cpus=1)
def process_partition(partition_id: int, records: List[dict]) -> dict:
    """Processes a discrete data partition asynchronously across workers."""
    transformed = [r["val"] * 2 for r in records if "val" in r]
    return {"partition_id": partition_id, "count": len(transformed), "sum": sum(transformed)}

# 3. Launch parallel asynchronous tasks
data_batches = [[{"val": i} for i in range(100)] for _ in range(4)]
futures: List[ray.ObjectRef] = [
    process_partition.remote(idx, batch) for idx, batch in enumerate(data_batches)
]

# 4. Resolve futures concurrently with ray.get or non-blocking ray.wait
ready_refs, unready_refs = ray.wait(futures, num_returns=len(futures), timeout=5.0)
results = ray.get(ready_refs)

Key API Parameter Reference

  • @ray.remote(num_cpus=1, num_gpus=0): Declares the task and its scheduling resource requirements.
  • func.remote(*args, **kwargs): Dispatches the task asynchronously and returns an ObjectRef.
  • ray.get(object_refs): Synchronously blocks until the referenced object(s) are resolved and deserialized.
  • ray.wait(object_refs, num_returns=1, timeout=None): Non-blocking partition of finished vs. pending futures.

3. Production Best Practices & Hardening Guidelines

  1. Avoid Fine-Grained Tasks: Ensure each remote task executes for at least 10–50ms to amortize scheduling and IPC serialization overhead.
  2. Never Call ray.get() in Loops: Calling ray.get() inside a for loop serializes execution; launch all .remote() calls first, then resolve in bulk.
  3. Pass ObjectRefs Directly: Pass ObjectRefs into downstream tasks rather than calling ray.get() on the driver; Ray will resolve dependencies automatically on the target worker.
  4. Specify Explicit Resource Demands: Always set num_cpus or memory limits to prevent worker over-subscription.
  5. Handle Timeouts in ray.wait: Always provide a bounded timeout when waiting on dynamic tasks to avoid driver deadlocks.

4. Troubleshooting & Diagnostic Workflows

  1. Worker Process Exhaustion:
  2. Symptom: Cluster hangs or scheduler logs "Too many pending tasks".
  3. Fix: Avoid infinite unconstrained task launches; use a semaphore or ray.wait() pipeline to bound concurrent in-flight tasks.
  4. Driver Memory Spike during ray.get():
  5. Symptom: Out-of-memory error on driver node.
  6. Fix: Process partitions in batches or stream results with generator tasks instead of resolving large datasets to the driver.
  7. Serialization Failures (TypeError: cannot pickle...):
  8. Symptom: Task submission fails with serialization errors.
  9. Fix: Ensure closures and arguments do not capture non-serializable objects (e.g., active database sockets or open file descriptors).

5. Hands-on Practice Exercises

Exercise ID Goal / Topic Playground Link
basics01 Ray Init & First Remote Task Open Exercise basics01 →
basics02 ObjectRefs and ray.get() Open Exercise basics02 →
basics03 Parallel Pipeline Execution Open Exercise basics03 →
basics04 Passing ObjectRefs to Tasks Open Exercise basics04 →
basics05 Dynamic Completion with ray.wait() Open Exercise basics05 →
basics06 Multiple Returns in Remote Tasks Open Exercise basics06 →