[core] [serve] ray.cancel on async actor task not cancellation the ray task

[abrarsheikh]

What happened + What you expected to happen

this issue is created specifically for ray core but the broader context is provided in #52533 (comment)

tldr is that in ray serve when a batch of request is made and immediately cancelled, handful of request run to completion because their underlying ray actor task that run those request don't actually cancel even though serve internally calls ray.cancel.

The point in serve code that is responsible for calling cancel on the ray actor task is this

ray/python/ray/serve/_private/replica_scheduler/replica_wrapper.py

Line 108 in 7faa62a

ray.cancel(obj_ref_gen)

Versions / Dependencies

2.44.1

Reproduction script

1. checkout debug branch that contain the necessary logging to observe this issue

git fetch --all
git checkout CI-965-abrar-cancel_3_debug

2. build ray from source (maybe necessary)

3. run the following repro script

This script triggers a burst of concurrent HTTP requests to a Ray Serve deployment, waits for the requests to be processed or cancelled, then scans the Ray Serve logs to extract all successfully served request IDs (GET / 200) and their corresponding Ray task IDs from cancellation events. It prints a table showing which requests were assigned a task that was supposed to be cancelled but still ran to completion.

import re
import time
import asyncio
import requests
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
from ray import serve
import shutil
import glob
import os
import ray

ray.init(logging_level="debug", dashboard_host="0.0.0.0")

LOG_DIR = "/tmp/ray/session_latest/logs"

# Matches: ... 30cdafed-5816-4a34-8994-424b4da6c9f2 -- GET / 200 ...
GET_LINE_RE = re.compile(r"\s([a-f0-9\-]{36}) -- GET / 200")

# Matches: ... UUID -- ... Ray task ID: TaskID(390a4a4a...)
CANCEL_LINE_RE = re.compile(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d+ .*? ([a-f0-9\-]{36}) -- .*?TaskID\(([a-f0-9]+)\)")

NUM_PARALLEL_REQUESTS = 250

def my_func():
    @serve.deployment(max_ongoing_requests=10)
    class Parent:
        async def __call__(self) -> int:
            return await asyncio.sleep(5)

    app = Parent.bind()
    serve.run(app)
    time.sleep(1.0)

    def get_and_cancel(x: int) -> None:
        try:
            _ = requests.get("http://localhost:8000/", timeout=2)
        except Exception:
            pass

    with ThreadPoolExecutor(max_workers=NUM_PARALLEL_REQUESTS) as exc:
        list(exc.map(get_and_cancel, range(NUM_PARALLEL_REQUESTS)))

    # Let Serve finish processing and flush logs
    time.sleep(10)

def logs_have_request_ids():
    for path in glob.glob(f"{LOG_DIR}/**/*", recursive=True):
        if os.path.isfile(path):
            with open(path, 'r', errors='ignore') as f:
                for line in f:
                    if GET_LINE_RE.search(line):
                        return True
    return False

def extract_request_ids():
    request_ids = set()
    for path in glob.glob(f"{LOG_DIR}/**/*", recursive=True):
        if os.path.isfile(path):
            with open(path, 'r', errors='ignore') as f:
                for line in f:
                    match = GET_LINE_RE.search(line)
                    if match:
                        request_ids.add(match.group(1))
    return request_ids

def find_task_ids(request_ids):
    mapping = defaultdict(list)
    for path in glob.glob(f"{LOG_DIR}/**/*", recursive=True):
        if os.path.isfile(path):
            with open(path, 'r', errors='ignore') as f:
                for line in f:
                    match = CANCEL_LINE_RE.search(line)
                    if match:
                        req_id, task_id = match.groups()
                        if req_id in request_ids:
                            mapping[req_id].append(task_id)
    return mapping

def main():
    print("Cleaning logs directory...")
    shutil.rmtree(LOG_DIR, ignore_errors=True)

    print("Waiting for logs with 'GET / 200'...")
    attempts = 0

    while not logs_have_request_ids():
        print(f"Waiting for logs with 'GET / 200'... (attempt {attempts})")
        my_func()
        time.sleep(1)
        attempts += 1

    print("Found logs, extracting request and task IDs...")
    request_ids = extract_request_ids()
    task_id_mapping = find_task_ids(request_ids)

    print("\n\nFollowing are the request IDs and their corresponding Ray task IDs that were supposed to be cancelled but ran to completion\n")
    print(f"{'Request ID':<40} | Ray Task ID(s)")
    print("-" * 80)
    for req_id, task_ids in task_id_mapping.items():
        print(f"{req_id:<40} | {', '.join(task_ids)}")

if __name__ == "__main__":
    main()

this will output something like

Following are the request IDs and their corresponding Ray task IDs that were supposed to be cancelled but ran to completion

Request ID                               | Ray Task ID(s)
--------------------------------------------------------------------------------
4426d915-ed42-45e7-b373-ca8899d19a1b     | 3c239ef5941e72a09642004102f5eaaf3928e1e701000000

expect it to take about 10 mins

Use the task id to search the core-worker logs to confirm that cancel is in fact being called

[2025-04-26 06:27:16,570 I 426616 426637] core_worker.cc:4377: Cancel an actor task task_id=3c239ef5941e72a09642004102f5eaaf3928e1e701000000 actor_id=9642004102f5eaaf3928e1e701000000

Issue Severity

Low: It annoys or frustrates me.

[edoakes]

@abrarsheikh have you confirmed that on the receiving side, the actor task is not being canceled?

From the writeup it seems that the level of diagnosis is: we call ray.cancel, but the task continues to run to completion.

This could be an issue in the core asyncio code that doesn't properly send the cancellation signal, or it could be something in the serve replica code that doesn't properly handle it. Have you narrowed it down further?

[abrarsheikh]

I tried adding logs at multiple places in the serve replica receiver code, but I am not seeing cancellation show up anywhere. One other observation I had was that the cancellation request (ray.cancel) happens before any code in the task has run, not sure what the implication of that is, but I thought I would mention it.

Here is a small repro script that maybe isolates the problem to the core

import asyncio
import concurrent.futures

import ray


@ray.remote(num_cpus=.1)
class Worker:
    async def long_running_task(self, task_id):
        await asyncio.sleep(15)
        return task_id

async def main():
    worker = Worker.remote()

    NUM_TASKS = 5000

    # Launch more tasks than available CPUs to cause queuing
    # Assume num_cpus = 2, launch 5 tasks
    with concurrent.futures.ThreadPoolExecutor(max_workers=NUM_TASKS) as executor:
        futures = [executor.submit(worker.long_running_task.remote, i) for i in range(NUM_TASKS)]
        refs = [future.result() for future in concurrent.futures.as_completed(futures)]

    # Immediately cancel one of the later tasks before it has a chance to run
    tasks_to_cancel = []
    for i in range(NUM_TASKS - 1, -1, -1):
        tasks_to_cancel.append(i)
        ray.cancel(refs[i])

    tasks_cancelled = []
    tasks_completed = []
    # Gather results, catching any cancellation
    for task_to_cancel in range(NUM_TASKS-1, -1, -1):
        try:
            result = await refs[task_to_cancel]
            tasks_completed.append(task_to_cancel)
            # print(f"Result: {result}")
        except ray.exceptions.TaskCancelledError:
            tasks_cancelled.append(task_to_cancel)

    print(f"Total tasks enqueued: {NUM_TASKS}")
    print(f"Total tasks completed: {len(tasks_completed)}")
    print(f"Total tasks cancelled: {len(tasks_cancelled)}")
    assert set(tasks_cancelled) == set(tasks_to_cancel), f"Task that were not cancelled: {set(tasks_to_cancel) - set(tasks_cancelled)}"

asyncio.run(main())

I had to run it a few times and increase NUM_TASKS to get it to fail. It doesn't matter how large a value we set for sleep(), some tasks run to completion despite immediate cancellation.

I suspect this is similar to what's happening in serve, but not 100% sure.

[edoakes]

Looking at the core cancellation path for asyncio actor tasks, my guess is the problem lies here:

ray/python/ray/_raylet.pyx

Line 2550 in afe51d8

future = worker.core_worker.get_queued_future(task_id)

It could be that in rare cases, cancel_async_task is executed before the future is actually submitted and populated here:

ray/python/ray/_raylet.pyx

Line 4592 in afe51d8

with self._task_id_to_future_lock:

Need to study the concurrency model here a bit more, because it looks naiively like there are clear race conditions related to the fine-grained lock that gates access to the futures.

[edoakes]

I don't see any mutex being held on the cancellation path in the core worker:

ray/src/ray/core_worker/core_worker.cc

Line 4343 in afe51d8

void CoreWorker::HandleCancelTask(rpc::CancelTaskRequest request,

[edoakes]

I confirmed the above:

• Bumped NUM_TASKS to 5000 in the script
• Added print on this line in the cancellation callback if the future is not found
• For each task that isn't cancelled appropriately, the print statement is hit

This happens reliably.

[edoakes]

I implemented a naiive fix that loops forever until the future exists in cancel_async_task, it fixes the issue.

-        future = worker.core_worker.get_queued_future(task_id)
-        if future is not None:
-            future.cancel()
+        while True:
+            future = worker.core_worker.get_queued_future(task_id)
+            if future is not None:
+                future.cancel()
+                break
+            else:
+                print("FUTURE DID NOT EXIST, WILL RETRY:", task_id)
+            time.sleep(0.1)

Obviously not the right fix, but confirms the timing of the race condition: cancel_async_task happens before the future is submitted and added to the map.