Deep Dive - Ray

A single LLM product can need thousands of CPU preprocessing tasks, a multi-node training job, a fleet of rollout actors, and several GPU-serving replicas. Building a separate distributed system for every stage wastes engineering time. Ray gives all of them the same small execution vocabulary.

That vocabulary is the project's main contribution. A Python function becomes a remote task, a class becomes a stateful actor, and every remote result becomes an object reference. Ray's higher-level libraries build data, training, tuning, reinforcement-learning, and serving systems from those three pieces.^{[1]Reference 1Ray: An AI Compute Enginehttps://github.com/ray-project/ray}

Why Ray exists

Traditional batch systems work well when a job is a fixed directed acyclic graph. AI workloads are often less tidy. Reinforcement-learning workers create new work after observing environments, serving routers change destinations as replicas scale, and tuning services stop trials from partial results. Dependencies appear while the program is running.

The original Ray paper targeted that mix of task-parallel and actor-based computation. It paired a dynamic execution engine with distributed scheduling and fault-tolerant storage, then showed that one interface could express workloads that previously needed separate systems.^{[2]Reference 2Ray: A Distributed Framework for Emerging AI Applicationshttps://www.usenix.org/conference/osdi18/presentation/moritz}

Ray Core keeps its public model compact:

Tasks are best for stateless work. Actors keep mutable state between calls and serialize methods by default. Object references let downstream work depend on values without forcing the driver to download those values first.^{[3]Reference 3What is Ray Core?https://docs.ray.io/en/latest/ray-core/walkthrough.html}

Three primitives create a dynamic graph

The first visual follows a small request through Ray Core. The driver submits work and receives references immediately. Raylets on cluster nodes schedule workers against declared resources. Workers produce immutable objects or update actor state. The driver doesn't need to know which machine holds each result.

Here is the smallest API shape. embed.remote returns references, while the actor keeps a counter across calls:

The expected output is 3. Ray resolves each top-level object-reference argument before the actor method runs. No explicit loop downloads each vector into the driver.

That concise code hides several contracts. num_cpus=1 is a scheduling request, not an operating-system limit. The actor's process can still allocate too much memory. The list of references preserves dependencies, but returning a giant Python object can still fill the object store. Distribution removes manual socket code; it doesn't remove capacity planning.

Scheduling happens at two levels

Each node runs a raylet. Local raylets decide whether work can run on their node and coordinate with peers when it can't. A global control service (GCS) stores cluster-level control state such as node membership and actor metadata. Workers execute user code, while the shared-memory object store holds larger immutable values.

Ray schedules logical resources. A task asking for one GPU is admitted only where one GPU slot is available. Custom resources can represent accelerator type, rack, license, or another placement constraint. The scheduler doesn't infer hidden requirements from code, so an undeclared GPU or excessive heap allocation can overcommit a node.

Placement groups reserve bundles of resources together. A tensor-parallel server might request four bundles with one GPU each and require them to be packed onto one node. A pipeline-parallel model could spread bundles across nodes. The placement strategy belongs to model topology, not to Ray alone.

Object references are data-plane handles

An ObjectRef behaves like a future, but it also names distributed data. Small values can travel directly between processes. Larger values enter each node's shared-memory object store, where eligible local buffers can avoid another full copy. Ray transfers objects between nodes when consumers need them.

The ownership system separates object data from object metadata. A worker that creates an object reference owns its metadata even if another node stores the bytes. The NSDI Ownership paper moved this metadata away from one central bottleneck and used decentralized reference counting for low-latency fine-grained work.^{[4]Reference 4Ownership: A Distributed Futures System for Fine-Grained Taskshttps://www.usenix.org/conference/nsdi21/presentation/cheng}

Ownership shapes recovery. If stored bytes disappear after a node failure, Ray can reconstruct task output from lineage. It can't recover an object created with ray.put after its owner dies, and replay assumes the generating task is deterministic and idempotent.^{[5]Reference 5Object Fault Tolerancehttps://docs.ray.io/en/latest/ray-core/fault_tolerance/objects.html}

Consider a task that charges a credit card and returns a receipt. Blind replay could charge twice. The task needs an idempotency key or an external transaction record. Ray can retry execution, but application semantics decide whether retry is safe.

Ray libraries share one substrate

Ray's libraries aren't separate cluster managers. They use Core primitives and can pass objects or actors across library boundaries.

Ray Data streams blocks through read, transform, shuffle, and write operators rather than materializing a whole corpus in the driver. Its batch-inference paths can keep CPU preprocessing feeding GPU replicas while backpressure limits in-flight blocks.^{[6]Reference 6Ray Datahttps://docs.ray.io/en/latest/data/data.html}

Ray Train creates a worker group, assigns ranks, and integrates distributed training frameworks. It handles launch, environment setup, checkpoints, and failure coordination around framework code. PyTorch, DeepSpeed, or another trainer still owns gradient math and collectives.^{[7]Reference 7Ray Trainhttps://docs.ray.io/en/latest/train/train.html}

Tune represents trials as distributed work and reacts to intermediate metrics. RLlib builds rollout, learner, and replay components on actors. Their value comes from composition: the same cluster can preprocess data, launch a training group, compare trials, and publish a serving deployment without translating every asset through a new scheduler.

Ray Serve LLM wraps inference engines

vLLM and SGLang optimize model execution inside an engine. Ray Serve LLM exposes OpenAI-compatible requests through OpenAiIngress. Its LLMEngine protocol lets vLLM or SGLang engines plug into LLMServer, while Ray places, scales, routes, and monitors instances.^{[8]Reference 8Ray Serve LLM Architecture Overviewhttps://docs.ray.io/en/latest/serve/llm/architecture/overview.html}

Serve deployments become actors, replicas become instances of those actors, and deployment handles connect components. An ingress deployment can expose an OpenAI-compatible API. Router policies select replicas based on queue state, session affinity, prefix reuse, or a custom rule. Autoscaling changes replica count from observed load.

For large models, one logical engine replica can span several GPU workers through tensor or pipeline parallelism. Serve LLM also supports patterns such as prefill-decode disaggregation, data-parallel attention, expert parallelism, and multi-LoRA routing. Ray coordinates resources and request movement; the configured engine remains responsible for continuous batching, PagedAttention or RadixAttention, and GPU execution.^{[8]Reference 8Ray Serve LLM Architecture Overviewhttps://docs.ray.io/en/latest/serve/llm/architecture/overview.html}

Where Ray helps LLM systems

Ray is strongest when one workload crosses several machines or execution modes:

1. Corpus processing. Ray Data reads object storage, applies CPU-heavy parsing, then streams tokenized batches to GPU workers.
2. Distributed fine-tuning. Ray Train creates workers and checkpoint plumbing around PyTorch, FSDP, DeepSpeed, or Hugging Face training code.
3. Reinforcement learning. Actors hold environments, rollout engines, reward services, and learners with different resource shapes.
4. Model selection. Tune allocates trials, stops weak candidates, and records intermediate results.
5. Online serving. Serve deploys routers and engine replicas, then scales them against traffic.
6. Offline inference. Ray Data streams large input collections through vLLM or another batch predictor.

Ray offers less value for one fixed single-node script, a tiny queue with ordinary web workers, or a serving engine that already fits cleanly in one process. Distribution adds packaging, cluster, observability, retry, and resource-model costs. Use it when those costs buy real parallelism or unify multiple distributed phases.

Failure modes are part of the design

Object-store and heap pressure

The object store has a bounded shared-memory budget. Ray can spill objects to disk when it fills, but spilling converts a memory path into storage I/O. Worker heap memory is separate. A worker can exhaust node memory even when the object store looks healthy.

The frequent anti-pattern is ray.get on a huge list of references. That asks the driver to materialize every value at once. Process results incrementally with ray.wait, keep data in Ray Data blocks, or aggregate remotely.

Actor recovery isn't automatic state recovery

Actors don't restart by default. Enabling restarts reruns the constructor; it doesn't restore mutable in-memory state. A durable actor needs an external checkpoint, replay log, or reconstruction function. Actor method retries also need idempotency.

Resource declarations can lie

Ray only sees declared resources. A task marked num_cpus=1 can start 32 native threads. A model actor that declares one GPU can allocate host memory sized for four. Validate actual process usage and use concurrency limits inside each actor.

Control-plane availability needs a plan

The GCS is central to cluster control. Production clusters need supported high-availability configuration and cluster recovery. KubeRay can recreate nodes and clusters, but transient queues and in-process actor state still disappear. A restarted cluster isn't equivalent to an uninterrupted request stream.

Dynamic graphs are harder to debug

A remote exception surfaces when a reference is resolved, sometimes far from submission. Task timelines, actor logs, resource views, object-memory reports, and distributed traces need correlated IDs. Keep unit logic runnable outside Ray so debugging doesn't always require a cluster.

Strengths and weaknesses

Ray should be compared with the right layer. Kubernetes manages containers and cluster reconciliation. Spark specializes in data-parallel processing. Celery handles task queues. vLLM and SGLang run LLM kernels and schedules. Ray can sit on Kubernetes, move data like a general runtime, queue dynamic tasks, and host vLLM engines, but it doesn't erase those systems' specialized roles.

Project identity

Ray began in UC Berkeley's RISELab in 2016 and 2017. Its OSDI paper was authored by researchers including Philipp Moritz, Robert Nishihara, Stephanie Wang, Michael I. Jordan, and Ion Stoica.^{[2]Reference 2Ray: A Distributed Framework for Emerging AI Applicationshttps://www.usenix.org/conference/osdi18/presentation/moritz} Anyscale was founded in 2019 by Ray's creators to build a managed production platform around the open-source runtime.^{[9]Reference 9About Anyscalehttps://www.anyscale.com/about}

The project now uses open governance under the Linux Foundation's PyTorch Foundation umbrella.^{[10]Reference 10Ray | PyTorchhttps://pytorch.org/projects/ray/} Committers, a technical steering committee, and lead maintainers govern technical work.^{[11]Reference 11Ray Project Governancehttps://docs.ray.io/en/latest/ray-governance/index.html} Anyscale remains a major contributor and commercial operator, but Ray's public governance and code aren't the same thing as the hosted Anyscale product.^{[9]Reference 9About Anyscalehttps://www.anyscale.com/about}

The research lineage explains today's design:

Only the first two are necessary to understand Core. The later projects show why a general runtime can host specialized libraries without turning every workload into one monolithic scheduler.

A practical reading path through the repository

Start with public boundaries instead of opening scheduler internals at random:

1. Read python/ray/__init__.py and remote-function or actor decorators to see user-facing handles.
2. Follow task submission into CoreWorker bindings and object references.
3. Inspect raylet scheduling and resource bookkeeping.
4. Trace one object from worker return through ownership metadata and object storage.
5. Open one library, such as Serve, and identify which components become actors.
6. Compare a local vLLM engine with its Ray Serve wrapper to mark orchestration versus execution ownership.

Pin the repository commit while reading. Ray's library APIs and internal C++ paths move faster than its three-primitives mental model.

What to remember

• Ray Core is a distributed execution substrate built from tasks, actors, and object references.
• Object references preserve dependencies and locality; premature ray.get calls often create driver bottlenecks.
• Logical resource requests guide placement but don't enforce process memory or thread limits.
• Ray Data, Train, Tune, RLlib, and Serve compose because they share Core primitives.
• Ray Serve LLM orchestrates vLLM or SGLang engines; it doesn't replace their token schedulers or kernels.
• Retry safety, actor state, object ownership, and GCS recovery need explicit production contracts.
• Ray is useful when distribution or cross-library composition pays for its operational complexity.