Yicheng Lu| August 25, 2026
We'd like to thank the following people for their contributions, benchmarks, and reviews to this effort: Edward Oakes, Mengjin Yan, Kartica Modi, Dhyey Shah, Josh Lee, Zac Policzer, Steve Alexander, Andrew Sy Kim (Google), Mao Yancan (ByteDance)
As Ray is used in more and more emerging AI workloads, we observed that users are pushing Ray Core to its limits from two directions:
Batch inference, shuffle, and other data pipelines: Workloads where the driver manages the lifecycle of millions of objects while running the scheduling loop on the critical path, which caps their sustained task throughput.Large scale RL and pre-/post-training: These run on clusters from 2,000 to 10,000 nodes, with tens of thousands of actors scheduled under network topology constraints. Startup time, and restart time after a failure, are critical for these users.
The bottlenecks we found kept falling into three kinds:
Lock contention between threads.
A single overwhelmed thread handling async work.
Scheduling on stale resource views.
After a round of Ray Core improvements, here is where these workloads stand compared to last year.
23% faster end to end for batch inference on 500 nodes.24% faster end to end for Ray Data shuffle.Now Ray can launch more actors on larger training clusters with topology constraints:
62× faster to get placement groups ready at 2,000 nodes.303× faster to get placement groups ready at 10,000 nodes.6.5× faster end to end launching actors at 2,000 nodes.Scales to 40,000 actors on a 10,000 node cluster, a scale it could not support a year ago.
The rest of this post walks through how we found each bottleneck and what we did about it. We start inside the driver with the data pipelines, then move to the training clusters at 10,000 nodes, and the three bottlenecks above keep coming back in both stories.
LinkWhen the Driver Becomes the Bottleneck #
It started with a puzzle from Ray Data. Batch inference and shuffle pipelines were losing time inside Ray Core calls, and their scheduling loop slowed down with them, hurting their sustained task throughput and end to end time. Yet when we measured those calls on their own, ray.wait
and the other Ray Core APIs were fast. So where was the time going?
To answer that, let's walk through the life of one object. Millions of objects take the same journey.
As shown in Figure 1, a Ray driver process has two threads that matter for this story. The Python thread runs your own code, like ray.wait
and your scheduling logic. The C thread, Ray Core's background thread, handles all the async work.
Looking at the Python thread part of Figure 1, you call f.remote()
to submit a task, get an object reference (ref) back, wait for the object to be ready (the real Ray Data scheduling loop uses a timeout and keeps checking), then pass the ref to other downstream tasks. Finally, when the object has been used and no one needs it anymore, you drop the ref.
Now, behind the scenes. When you call ray.wait,
a Ray Core API, it puts your thread to sleep waiting on the object's ready condition variable. The task runs on worker node A and stores its output object there. The ideal waiting time should be exactly your task's running time. The C thread gets the notification after the task finishes, and after doing the bookkeeping, including recording the object location, it wakes up the Python thread.
Now you pass the ref to the downstream task g. It gets scheduled on worker node B, which needs to know where the object is, so it subscribes to the object's location. The C thread recorded that location earlier, and now publishes it back. Node B pulls the object over from node A, and task g runs.
And the drop behind the scenes, brings the object's reference count to zero, which triggers the cleanup, including publishing a failure message to anyone still waiting for that object's location to stop waiting and give up pulling it, as well as freeing every copy in the cluster. With the background context above, let's look at the bottlenecks we found in the driver and how we solved them.
LinkTwo Threads, One Lock
As we can see from the lifecycle of the object above, both threads publish object updates, and there is a publish lock guarding all those publishing operations. As shown in Figure 2, when dropping the last ref, the Python thread needed to grab the lock. At the same time, the C thread needs the same lock to publish object locations.
Two threads, one lock, which leads to serious contention at scale. Due to the nature of batch inference, objects are being pulled every second, thus the C thread takes the publish lock constantly to publish object locations. At the same time, the objects are constantly being dropped after usage, the Python thread also needed to take the publish lock to publish failure messages. On a 500 node batch inference workload, the publish lock alone was eating 17.4% of the scheduling loop's time.
The fix for this kind of lock contention is to post all related work onto a single thread. We moved every publish, including that failure message, to the C thread, so the Python thread never needed to take the publish lock again and we saw the lock contention vanish from the profile.
LinkOne Overwhelmed Thread
And there is the second bottleneck, hiding in the middle of the object lifecycle journey, in the ray.wait
itself. As shown in Figure 3, the time you spend in ray.wait
is decided not only by how fast the object is produced, but also by how busy the C thread is. The blue part of your wait is from the task execution. The red part is the extra blocking time. When your object is ready, the C thread might also be dealing with many other objects' location updates and everything else, and all of those can sit ahead of the wake up in the C thread's operation queue. That queue delay turned out to be a top contributor to the slowness. So we went hunting and found two problems.
The first problem we found was a flood of tiny callbacks on the C thread. Each callback is cheap on its own, but if you post millions of them, the posting itself hurts performance. In one shuffle run, those callbacks came to over a million and occupied about 17% of the C thread's time, each one taking 0.04 ms to run after waiting 5.7 seconds for its turn.
Most of these callbacks turned out to be argument checks. Before a task can be submitted, every argument has to be confirmed as produced somewhere in the cluster, one callback per argument, and shuffle reduce tasks take a huge number of arguments. But reduce only starts after all map tasks are finished, so all of those arguments should have already been produced.
The issue is that the driver used to post the callback even when the argument was already ready at the moment the task was created, to avoid a potential deadlock. The fix makes sure all locking is safe, and only posts the async check when an argument is not already ready at the moment the task is created. That alone made shuffles 9.2% faster end to end.
The second problem jumped straight out of the profile. Two thirds of the C thread's time was just publishing object location updates out to other nodes. The investigation showed that every task on every node subscribed to the locations of its arguments. However, if the objects are already on the local node, it is not necessary for the task to subscribe to the locations of them. And as Ray schedules tasks close to their data, the possibility of the objects on the same node is pretty high. We made the improvements by skipping the request for any objects that are already local and it cut the driver's publishing traffic in half.
Together with many others not covered here, these fixes improved sustained task throughput by up to 1.6×. As a result, the batch inference pipeline now runs 23% faster end to end on 500 nodes, and the shuffle 24% faster.
In the longer term, we are working toward separating the C thread into multiple threads to further speed up Ray Core.
LinkScaling for Large Training Runs #
Apart from the above Ray Data workloads, more and more users tell us that actor startup and restart time is what limits their large scale RL and pre-/post-training workloads. Idle GPU time is an expensive waste, and they pay it multiple times, once at launch, and again at every failure, because an NCCL communicator is all-or-nothing and one dead node means restarting every actor.
They also care about network topology constraints, since the speed of data transfer between those actors plays a big role in the performance of the training runs, and the preferred way to express them is Ray's topology aware placement group scheduling. For example, when using NVIDIA GB200 or GB300 NVL72 racks, a common setup can be one placement group per rack of 18 nodes, one bundle per node, 4 actors inside each bundle, each corresponding to one GPU in the rack, 18 times 4 being exactly the 72 in NVL72.
So, to support RL and pre-/post-training workloads at their current and future scale, Ray first needs to scale towards the 10,000 node level, and then make placement group scheduling and actor scheduling fast enough.
LinkKeeping the GCS Stable at 10,000 Nodes
GCS, Ray's global control service, sits at the center of the control plane, doing critical work that includes:
Node management
Actor lifecycle management
Serving the internal key value store
Collecting demand for the autoscaler
Placement group scheduling
All of that used to share one main thread, and a lot of that work grows with the number of nodes. This is the second bottleneck again, one overwhelmed thread. And the GCS main thread was already busy long before any user workload even arrived. So when an actor or placement group workload came, the GCS soon became the bottleneck and was easily overwhelmed.
We made the GCS multithreaded, so it can scale with the cluster and carry the actor load. Broadcasting node status updates, serving the internal key value store, and collecting pending demand for the autoscaler each moved to its own thread.
We also lightened what stays on the GCS. Node change subscriptions went from O(workers) to O(nodes), placement group bundle information is now cached instead of queried from the GCS during actor submission, and enabling task events to move out of the GCS entirely, flowing straight to the dashboard instead.
Here is where the idle load stands now. On last year's Ray (2.51), a 10,000 node cluster with zero jobs submitted already kept the GCS main thread about 61% busy; on today's nightly, the same idle cluster sits at about 38%.
The result is a GCS that stays stable at 10,000 nodes. But this does not automatically mean scheduling gets faster, and that is a different problem.
LinkRay's Distributed Scheduling
Ray's scheduling is distributed. Every node is able to schedule, and that means every node has to know the whole cluster's resources. As shown in Figure 4, every node maintains its own local resource view, and that local view is the source of truth. Upon any change, the node snapshots its own resource view and sends the snapshot to the GCS. The GCS collects all the information and fans each change out to every node, so every node ends up holding the resource views of everyone else. Inside Ray, this shipping service is called the syncer.
Now, let's go through a simple task submission to see how Ray schedules. The submitter first sends the request to a node. In principle, this can be any node. In practice, it is usually the node the submitter sits on, or the node that holds most of the data the task needs. Let's say that node is node A. Node A first checks whether it can place the task locally. If not, it picks a node based on its resource view of the cluster, in this case node B. Node A then optimistically deducts the task's resources from its own view, and hands the decision back to the submitter. The submitter then sends the request to node B, and the same process continues. This time B accepts, allocates the resources for real, and the task runs on node B.
Notice what this design leans on. Node A makes its decision on its resource view, and that view can be stale, so the decision can be wrong. In that case, the submitter sends the request to node B, node B rejects it, and we reschedule. We rely on the syncer to deliver the latest resource views, and we rely on it to correct node A's wrong optimistic deduction. That reliance is where the third bottleneck lives. When the syncer falls behind, everything above is scheduling on a stale view.
LinkPlacement Groups: From Hours to Seconds
Placement group scheduling at scale was slow end to end. One cause came straight from the design above. Remember the optimistic deduction, and the syncer that corrects it when it goes wrong. There is one leak the syncer cannot cover, and Figure 6 walks through it from start to finish. From the moment node A hands the request back to the submitter, node A does not track it anymore, but it has already optimistically deducted node B's resources in its own view. The only way to correct a wrong deduction is a syncer update. So if the request gets dropped by the submitter, or the submitter dies, then nothing ever changes on node B, there is nothing to trigger the syncer, and node A's deduction leaks permanently.
To solve the problem in Ray, every three seconds, every node resets its locally modified view back to the last snapshot it received, and the GCS, which places placement groups on its own resource view, did the same. Normally that is harmless, because the syncer is far faster than three seconds, and fresh snapshots keep arriving. However, at large scale, the syncer is slow, and the insurance itself became the problem. Placement groups are usually created right at cluster start, so the snapshot every node reported said the cluster was empty at init. Every three seconds the reset made filled resources look free again, and the scheduler was not really scheduling at all, just sending placement groups to the already filled nodes until the syncer caught up.
But do we really need the reset for placement groups? It turns out placement group scheduling is different from the rest of Ray's scheduling. As Figure 7 shows, placement groups go through a two-phase commit and can only be scheduled by the GCS. The GCS first prepares, trying to reserve all the resources from the nodes, and commits only after all reservations succeed. Unlike an actor or task request, a placement group request is never forwarded through anyone else, and the GCS always knows whether it succeeded, thanks to the two-phase commit. So the GCS never needed the insurance at all. Simply removing the reset code for the GCS collapsed placement group scheduling at 10,000 nodes to seconds.
The other part of the slowness was the overhead introduced when checking the readiness of a placement group. pg.ready()
is the user facing API to check whether a placement group is ready. It sounds like a simple check. Under the hood it used to submit a dummy task into the placement group. A task getting scheduled into the group means the group must be ready, and because a task naturally returns a ref, pg.ready()
becomes an async API for free. But the shortcut had real costs. The dummy task inherited your job's runtime environment, so a simple resource check could sit waiting for a torch install to finish. And because the API hands back one ref per placement group, it naturally invites a serial loop. We found many users write it this way:
for pg in pgs:
ray.get(pg.ready())
But each ray.get
blocked on its own dummy task, so the checks ran one after another, submitting the next dummy task only after the previous one finished. A thousand groups meant a thousand sequential round trips. pg.ready()
is now a direct asynchronous query to the GCS. It still returns the same ObjectRef
, but there is no task behind it anymore, and the overhead is gone.
Here is what all of that bought, the same placement group creation, from 200 up to 10,000 nodes, one group per rack of 18 nodes, one bundle per node, on last year's Ray and on today's nightly.
LinkActor Scheduling: Minutes Behind the Truth
With placement groups down to seconds, actor startup inside the topology aware placement groups was still slow at the 10,000 node level. We are pushing two paths in parallel, one for each of the two problems below.
The first problem is that the syncer itself is slow. Ray assumes every node can make scheduling decisions, so as Figure 9 shows, changes fan in to the GCS from every node, and each one fans out to all 10,000 nodes. This makes the syncer slow, so nodes often hold stale resource views and make poor decisions.
The stale view hurts actors in two ways, especially with placement groups. Actor scheduling normally runs through the head node. After the topology aware placement groups have been placed, the head node needs the placement group resource info before it can schedule actors into them. The placement group resources originally live only on the nodes that hold them, and rely on the syncer to propagate to the head node. The syncer is slow. At the 10,000 node level we have seen it take around 200 seconds to propagate at peak. So the head node cannot schedule at all until the placement group info arrives.
And once it arrives, the head node is still scheduling on a stale view, since the syncer often cannot deliver fast, and the three second reset makes it even staler. Both push the head node into bad decisions, sending actors to nodes that are already full. That is the reject and reschedule, and it gets even worse as the cluster fills up, with more and more rescheduling at the end. We saw that long tail directly.
To solve the problem, we are currently working toward a more centralized scheduling mechanism. Most Ray workloads only need a few nodes to make scheduling decisions, so we are walking back the assumption this whole design started from, that every node can schedule. Once decisions are concentrated, only a few nodes need the whole cluster's resource view at all, so the syncer has less to carry, the view is more accurate, and thus most of the rejects that stale views cause today disappear.
The second problem is that an actor has two brains today. The GCS decides when it is created, restarted, and destroyed, while the owner, usually the driver, holds the references and knows who is still using it. Every actor startup pays for the conversation between the two, and Figure 10 walks through it, at least four messages between the owner and the GCS, from registering the actor, through a standing subscription on the actor's ref count, to getting its address back. And restart is also affected. Remember, for training workloads one dead actor means every actor dies and restarts. Tens of thousands of standing subscriptions fire back at the GCS at the same moment, and the restart replays the whole conversation from the top, once per actor, again all at once.
And to solve the above problem, what we are currently working towards is moving the actor lifecycle out of the GCS, over to the owner. The reason actor lifecycle management lives in the GCS in the first place is detached actors. Their lifecycle is independent of the driver, so they must survive after the driver dies. For non detached actors, once the lifecycle lives with the owner, the whole conversation delay disappears.
The gains below come from the GCS and placement group work above, plus early work for the above two paths, including a community contribution that makes the syncer send its sync messages in batches opportunistically. Here is what that adds up to, creating the topology aware placement groups and then scheduling actors into them, on last year's Ray and on today's nightly.
LinkWhere This Leaves Us #
Every fix described in this post is already in Ray nightly today, and we are continuing to invest in actor and driver scalability, including the two improvements mentioned above. If you are pushing Ray toward either edge, we would love to hear where it bends! You can find us on GitHub or in
Ray Slack