Occasional Reference Counting Error under a high load

Ray Core
1

1. Severity of the issue: (select one)
Medium: Significantly affects my productivity but can find a workaround.

2. Environment:

  • Ray version: 2.49.1
  • Python version: 3.10.18
  • OS: Ubuntu 24.04.3 LTS
  • Cloud/Infrastructure: Docker cluster
  • Other libs/tools (if relevant):

3. What happened vs. what you expected:

  • Expected: Ray keeps reference counts greater than 0 for all objects if they are required in later stages until all of them were utilized (e.g. an object created in one deployment should be kept in object store until another deployment successfully retrieves and uses it)
  • Actual: Certain objects reference count goes to zero and the object is being deleted, while still being requested later

I am working with a production system that heavily relies on Ray’s remote tasks and Ray’s Deployments. A majority of those deployments contains an ML/CV logic that produces somewhat large numpy arrays (hundreds, sometimes tens of MB for each array, usually tens of those are required for a successful run). We use the persistent Actor as an owner for all those arrays and we set the ownership with the NumpyStore wrapper like this:

@ray.remote
class PlasmaStoreActor:
    def __init__(self):
        self.alive = True

    def ensure_alive(self) -> bool:
        return self.alive


class NumpyStore:
    def __init__(self, data: np.ndarray):
        if not isinstance(data, np.ndarray):
            raise TypeError(type(data))
        self.shape = data.shape
        self.size = data.size
        self.ndim = data.ndim
        self.dtype = data.dtype
        self.length = len(data)
        actor = ray.get_actor(name="PlasmaStoreActor", namespace="PlasmaStore")
        self.data_ref = [ray.put(data, _owner=actor)]

    def retrieve(self) -> np.ndarray:
        return ray.get(self.data_ref[0])

    def __len__(self) -> int:
        return self.length

Under normal conditions this ownership model works well, but under heavier load it occasionally fails at two specific spots, when calling retrieve method (so, essentially ray.get). I can simulate that heavy load by setting max_ongoing_requests at least to 3 and sending at least 2 requests to the system. Under those conditions I might get ReferenceCountingAssertionError, saying that a requested numpy array was deleted from memory while still being referenced. It is also worth noting that this error doesn’t appear consistently on the same data - even when running the same data over and over only a fraction of it will fail.

For know, unfortunately, I cannot create a short reproducible script. Inspecting ray memory output with RAY_record_ref_creation_sites=1doesn’t tell much; Ray logs with DEBUG verbosity only demonstrate that Ray honestly counts object references and deletes it when all possible counters drop to 0.

Here what I’ve already tried:

  1. Getting rid of NumpyStore for those masks and returning/passing them as regular numpy arrays and corresponding references results in a severe memory leak. Main hypothesis here is that Ray workers become owners for those arrays and never die - therefore it keeps reference count above 0.
  2. Moving to newer Ray (2.54.0) - didn’t help
  3. Lowering and increasing kill_idle_workers_interval_ms and idle_worker_killing_time_threshold_ms parameters - didn’t correlate with error frequency
  4. Lowering max_ongoing_requests - that lowers a simultaneous load and the error no longer reproduces.

My request for help sums up into 2 questions - Is our ownership model wrong and fragile, which leads to this error appearing? And how should one come up to debugging such errors? I know exactly where a faulty array is being created, what deployments and remote functions it passes through and where it might fail (same place, but unpredictable); it’s id in object store; even a worker/Deployment in which it was dereferenced.

2

Your ownership model—using a persistent actor as the owner via ray.put(data, _owner=actor)—is generally correct for keeping objects alive across deployments, but Ray’s distributed reference counting can still be fragile under high concurrency or complex object reference flows. The ReferenceCountingAssertionError indicates that, despite your efforts, the reference count for some objects is dropping to zero prematurely, causing Ray to delete the object before all consumers have finished using it. This can happen if references are not properly propagated or if intermediate tasks/actors holding references die or lose track of them before all downstream consumers have acquired their own references, especially under heavy load or parallel requests. This is a known challenge in Ray’s reference counting protocol, particularly when objects are passed through multiple tasks or actors and one of the intermediaries exits unexpectedly or fails to notify the owner about new borrowers (see discussion, reference counting doc).

For debugging, you are already using RAY_record_ref_creation_sites=1 and ray memory, which are the recommended tools. However, these tools may not always reveal subtle reference propagation issues, especially if the problem is due to lost notifications between actors or tasks. To further debug, ensure that:

  • All references to the object are explicitly passed and not just stored in local variables that may go out of scope.
  • No intermediate actor or task holding a reference is dying or being killed before all downstream consumers have acquired their references.
  • Avoid passing object references through multiple layers of tasks/actors if possible, or ensure that the owner is always notified of all borrowers (see this issue for a similar scenario).
  • Consider using named actors or a more centralized reference management pattern if you need strong guarantees.

Would you like a step-by-step breakdown of how Ray’s reference counting can fail in these scenarios, or more concrete suggestions for restructuring your workflow?

Sources:

Hint: Mention @Herald in the post for followups.

3

@Herald

Could you give me more detailed answer on the last suggestion with named actors and centralized reference management pattern?

4

Certainly! Using named actors as a centralized reference manager means you create a long-lived, named actor (e.g., “PlasmaStoreActor”) whose sole responsibility is to store and manage all ObjectRefs that need to persist across tasks, actors, or deployments. Instead of passing ObjectRefs directly between tasks/actors (which can lead to lost references if an intermediate dies), you store all ObjectRefs in this actor and have all consumers retrieve the data via remote calls to the actor. This ensures the actor always holds a strong reference, preventing premature deletion, and provides a single place to manage object lifetimes and cleanup (see discussion, reference counting test).

Example pattern:

@ray.remote
class PlasmaStoreActor:
    def __init__(self):
        self.refs = {}

    def put(self, key, data):
        self.refs[key] = ray.put(data)

    def get(self, key):
        return ray.get(self.refs[key])

    def delete(self, key):
        del self.refs[key]

# Usage:
store = PlasmaStoreActor.options(name="PlasmaStoreActor", lifetime="detached").remote()
store.put.remote("my_array", np_array)
result = store.get.remote("my_array")

This approach avoids passing ObjectRefs through multiple layers and ensures the actor’s reference keeps the object alive until you explicitly delete it. It also makes debugging and memory management easier, as all references are tracked in one place (see also).

Would you like a step-by-step explanation of how this pattern prevents reference counting errors?

Sources:

Hint: Mention @Herald in the post for followups.

5

@Herald

Is there another way? Introducing centralized reference manager would significantly complicated our code - not only we would have to manually put and delete new objects (numpy arrays), but also we would have to track which objects relates to which currently processed requests.

Another related question here - how does Ray internally understand when objects are no longer necessary and can be safely deleted? I understand how this works in simple setup (for example, a couple of tasks running one after another), but in our case we have a hierarchy of Ray Deployments, and created numpy arrays might be used really late in the code.

6

Ray uses distributed reference counting to track when objects can be safely deleted: an object remains in the object store as long as there is at least one live ObjectRef anywhere in the cluster, including references held by drivers, actors, tasks, or as arguments to pending tasks (Ray memory management docs). When all references go out of scope (i.e., are deleted or garbage collected), Ray decrements the count, and once it reaches zero, the object is eligible for eviction from the object store (reference counting doc).

In complex workflows with multiple deployments and late usage, Ray’s reference counting can become fragile—especially if ObjectRefs are passed through intermediate actors/tasks that may die or lose references before downstream consumers acquire their own. If you want to avoid a centralized manager, you must ensure that every consumer obtains its own ObjectRef directly from the original owner or that all intermediate holders remain alive until all consumers are done. Otherwise, you risk premature deletion. There is no built-in way to “pin” objects globally without manual management; Ray’s design relies on the lifetime of ObjectRefs in the cluster (see discussion, memory management FAQ).

Would you like a step-by-step example of how reference counting works in a multi-deployment scenario, or suggestions for minimizing manual tracking?

Sources:

Hint: Mention @Herald in the post for followups.

7

I’ve reconstructed a minimalistic setup to reproduce it: a server.py deploys several chained Deployments and load.py sends requests to process.

Here is an exact environment:

ray[serve]==2.49.1
numpy>=1.24.0,<3
fastapi>=0.100.0,<0.139.2
httpx>=0.24.0
uvicorn>=0.23.0
starlette>=0.27.0,<1.0
pydantic>=2.0.0
protobuf>=3.20.3,<7

At this point I’ve also tried running it with Ray 2.56.1 and removing `_owner` - runs completely the same; the majority of requests are being processed successfully, while some catch error in the same spot. In the attached code it is this line - _ = result["a"]["mask"].retrieve().

Here is a server.py:

#!/usr/bin/env python3
"""Deploy the minimal ref-count repro Serve app. Press Ctrl+C to stop."""

from __future__ import annotations

import os
import time

import ray
from ray import serve

from deployments import bind_app, create_object_owner_actor

_ROOT = os.path.dirname(os.path.abspath(__file__))

os.environ.setdefault("RAY_record_ref_creation_sites", "1")
os.environ["PYTHONPATH"] = _ROOT + (
    os.pathsep + os.environ["PYTHONPATH"] if os.environ.get("PYTHONPATH") else ""
)

ray.init(
    num_cpus=12,
    object_store_memory=800_000_000,
    runtime_env={"working_dir": _ROOT, "env_vars": {"PYTHONPATH": _ROOT}},
    _system_config={"local_fs_capacity_threshold": 0.99},
)

create_object_owner_actor()

serve.start(http_options={"host": "0.0.0.0", "port": 8000})
serve.run(bind_app(), name="ref_count_repro", route_prefix="/")

print("Server running at http://0.0.0.0:8000  POST /process")
try:
    while True:
        time.sleep(3600)
except KeyboardInterrupt:
    serve.shutdown()

load.py:

#!/usr/bin/env python3
"""Concurrent load against the minimal ref-count repro server."""

from __future__ import annotations

import asyncio

import httpx

URL = "http://127.0.0.1:8000/process"
CONCURRENCY = 10
TOTAL = 500
TIMEOUT_S = 300.0


async def main() -> None:
    sem = asyncio.Semaphore(CONCURRENCY)

    async def one(client: httpx.AsyncClient, i: int) -> None:
        async with sem:
            r = await client.post(URL, timeout=TIMEOUT_S)
            if "ReferenceCountingAssertionError" in r.text or r.status_code >= 500:
                print(f"request={i} status={r.status_code} body={r.text}")

    async with httpx.AsyncClient() as client:
        await asyncio.gather(*(one(client, i) for i in range(TOTAL)))


if __name__ == "__main__":
    asyncio.run(main())

and deployments.py

from __future__ import annotations

import asyncio
import time
from typing import Any

import numpy as np
import ray
from fastapi import FastAPI
from ray import serve
from ray.serve.handle import DeploymentHandle

_PATCHED = False

def _apply_serve_lock_patch() -> None:
    global _PATCHED
    if _PATCHED:
        return
    from ray.serve._private.replica_result import ActorReplicaResult

    async def to_object_ref_async(self) -> object:
        assert not self._is_streaming
        while True:
            if self._obj_ref is not None:
                return self._obj_ref
            acquired = self._object_ref_or_gen_sync_lock.acquire(blocking=False)
            if acquired:
                try:
                    if self._obj_ref is None:
                        self._obj_ref = await self._obj_ref_gen.__anext__()
                    return self._obj_ref
                finally:
                    self._object_ref_or_gen_sync_lock.release()
            else:
                await asyncio.sleep(0)

    ActorReplicaResult.to_object_ref_async = to_object_ref_async  # type: ignore[method-assign]
    _PATCHED = True


_apply_serve_lock_patch()

MAX_ONGOING = 10
INFERENCE_SLEEP_S = 0.05
ARRAY_SHAPE = (64, 128, 128)
_OPTS = {"max_ongoing_requests": MAX_ONGOING, "ray_actor_options": {"num_cpus": 0}}

_OWNER_NAME = "ObjectOwnerActor"
_OWNER_NAMESPACE = "ObjectOwner"


@ray.remote
class ObjectOwnerActor:
    """Detached actor that owns object refs via ray.put(..., _owner=actor)."""

    def __init__(self) -> None:
        self.alive = True

    def ensure_alive(self) -> bool:
        return self.alive


def create_object_owner_actor() -> None:
    actor = ObjectOwnerActor.options(
        name=_OWNER_NAME,
        namespace=_OWNER_NAMESPACE,
        lifetime="detached",
        get_if_exists=True,
        num_cpus=0,
    ).remote()
    assert ray.get(actor.ensure_alive.remote())


class ArrayRef:
    """Wraps a numpy array owned by ObjectOwnerActor."""

    def __init__(self, data: np.ndarray) -> None:
        self.shape = data.shape
        self.dtype = data.dtype
        owner = ray.get_actor(name=_OWNER_NAME, namespace=_OWNER_NAMESPACE)
        self.data_ref = [ray.put(data, _owner=owner)]

    def retrieve(self) -> np.ndarray:
        return ray.get(self.data_ref[0])


# ---------------------------------------------------------------------------
# Deployments
# ---------------------------------------------------------------------------

@serve.deployment
class Worker:
    async def __call__(self, image: ArrayRef) -> ArrayRef:
        return await worker_infer.remote(image)


@ray.remote
def worker_infer(image: ArrayRef) -> ArrayRef:
    arr = image.retrieve()
    time.sleep(INFERENCE_SLEEP_S)
    return ArrayRef(arr.copy())


@serve.deployment
class BranchB:
    def __init__(self, worker: DeploymentHandle) -> None:
        self.worker = worker

    async def __call__(self, image: ArrayRef) -> dict[str, Any]:
        refs = [self.worker.remote(image=image) for _ in range(3)]
        masks = await asyncio.gather(*refs)
        return {"masks": list(masks)}

    async def localize(self, result: dict[str, Any], shared: ArrayRef) -> dict[str, Any]:
        _ = await localize_b.remote(shared)
        return result


@ray.remote
def localize_b(shared: ArrayRef) -> ArrayRef:
    arr = shared.retrieve()
    return ArrayRef(arr.copy())


@serve.deployment
class BranchA:
    async def __call__(self, image: ArrayRef) -> dict[str, Any]:
        time.sleep(INFERENCE_SLEEP_S)
        arr = image.retrieve()
        return {"mask": ArrayRef(arr.copy())}

    async def merge(
        self,
        a_result: dict[str, Any],
        b_result: dict[str, Any],
    ) -> dict[str, Any]:
        return await merge_ab.remote(a_result["mask"], b_result["masks"])

    async def localize(self, result: dict[str, Any], shared: ArrayRef) -> dict[str, Any]:
        _ = await localize_a.remote(result["mask"], shared)
        return result


@ray.remote
def merge_ab(mask_a: ArrayRef, masks_b: list[ArrayRef]) -> dict[str, Any]:
    arr = mask_a.retrieve()
    for m in masks_b:
        _ = m.retrieve()
    return {"mask": ArrayRef(arr.copy())}


@ray.remote
def localize_a(mask: ArrayRef, shared: ArrayRef) -> ArrayRef:
    arr = mask.retrieve()
    _ = shared.retrieve()
    return ArrayRef(arr.copy())


@serve.deployment
class Orchestrator:
    def __init__(self, branch_a: DeploymentHandle, branch_b: DeploymentHandle) -> None:
        self.branch_a = branch_a
        self.branch_b = branch_b

    async def __call__(self, image: ArrayRef) -> dict[str, Any]:
        # Fire both branches; keep responses unresolved.
        a_ref = self.branch_a.remote(image=image)
        b_ref = self.branch_b.remote(image=image)

        # Chain unresolved responses into merge.
        a_ref = self.branch_a.merge.remote(a_result=a_ref, b_result=b_ref)

        shared = await make_shared.remote(image=image)

        # Fan out shared ArrayRef + still-unresolved branch refs into localize.
        b_ref = self.branch_b.localize.remote(result=b_ref, shared=shared)
        a_ref = self.branch_a.localize.remote(result=a_ref, shared=shared)

        a_result = await a_ref
        b_result = await b_ref
        return {"a": a_result, "b": b_result}


@ray.remote
def make_shared(image: ArrayRef) -> ArrayRef:
    arr = image.retrieve()
    return ArrayRef(arr.copy())


app = FastAPI()


@serve.deployment
@serve.ingress(app)
class Ingress:
    def __init__(self, orch: DeploymentHandle) -> None:
        create_object_owner_actor()
        self.orch = orch

    @app.post("/process")
    async def process(self) -> dict[str, str]:
        data = np.random.randint(-1000, 400, size=ARRAY_SHAPE, dtype=np.int16)
        image = ArrayRef(data)
        result = await self.orch.remote(image=image)

        # Extra retrieve at the ingress boundary.
        _ = result["a"]["mask"].retrieve()
        for mask in result["b"]["masks"]:
            _ = mask.retrieve()

        return {"status": "ok"}


def bind_app() -> Any:
    worker = Worker.options(**_OPTS).bind()
    branch_b = BranchB.options(**_OPTS).bind(worker=worker)
    branch_a = BranchA.options(**_OPTS).bind()
    orch = Orchestrator.options(**_OPTS).bind(branch_a=branch_a, branch_b=branch_b)
    return Ingress.options(**_OPTS).bind(orch=orch)

How to run: run server.py and wait until all Deployments had been deployed. Then in a separate shell run load.py. It should return something like request=28 status=500 body=Internal Server Error in case of ReferenceCountingAssertionError

What I’ve also noticed - if I await both b_ref and a_ref before calling .localize methods, then this error doesn’t appear, like that:

        shared = await make_shared.remote(image=image)
        b_ref = await b_ref
        a_ref = await a_ref
        # Fan out shared ArrayRef + still-unresolved branch refs into localize.
        b_ref = self.branch_b.localize.remote(result=b_ref, shared=shared)
        a_ref = self.branch_a.localize.remote(result=a_ref, shared=shared)

Unfortunately, it only solves one case in my production code since this error appears in other Deployment as well, and I cannot figure out what and how to await there.

@Herald

8

Your minimal reproduction demonstrates a classic Ray distributed reference counting pitfall: when you chain unresolved ObjectRefs (e.g., passing unresolved deployment responses or remote results as arguments to further remote calls), Ray may drop references too early if no strong reference is held anywhere in the cluster. This is especially problematic when you pass ObjectRefs through multiple layers of tasks/actors/deployments without awaiting them, as Ray’s reference counting protocol relies on each consumer or intermediate explicitly holding a reference until all downstream users have acquired their own (see reference counting doc, issue 18456).

When you await the intermediate results before passing them to the next stage (as in your workaround), you ensure that the returned objects are materialized and their references are held by the driver or deployment, preventing premature deletion. If you do not await, and the only reference is in a pending task or deployment, Ray may drop the object if the intermediate task/actor finishes or dies before the next consumer acquires its reference. This is why the error disappears when you await both a_ref and b_ref before calling .localize, but appears otherwise.

There is no built-in way to “pin” objects globally except by holding a reference in a persistent actor or by always awaiting and storing intermediate results before passing them further. If you cannot always await, you must ensure that at least one live reference exists (e.g., by storing the ObjectRef in a persistent actor or in the driver) until all consumers are done. This is a limitation of Ray’s current reference counting protocol, especially in complex, asynchronous, or highly parallel workflows (see also, test_reference_counting.py).

Would you like a step-by-step breakdown of how reference loss occurs in your specific chain, or suggestions for refactoring to minimize manual reference management?

Sources:

Hint: Mention @Herald in the post for followups.