[data][llm] LLM Batch Inference Resiliency Thread

[jiangwu300]

Description

Hi team! I'd like to list a few of the things I'm seeing when scaling up Ray Data LLM to batch sizes of 10 million + prompts (each prompt itself is also quite long, often times over 10k tokens). Thought this would serve as a helpful place for discussion on the resiliency of Ray Data LLM at a large scale.

1. When using Ray Data LLM with the vLLM processor, there seems to be no actor level resiliency during vllm engine failures. When running at a high concurrency (for example 4 replicas of Qwen3-32B-FP8), if one of the vllm engine dies, I expect that the job will continue on the 3 remaining replicas while the 4th one tries to restart itself.
   As an example, I'm processing millions of prompts in a single rayjob via kuberay, this rayjob asks for 4 GPU instances (p5.48xlarge). Sometimes the engine dies randomly (see logs below) out of my control and shuts the whole job down. It would be very cool if it was possible to add resiliency at the engine failure level to restart a failed engine while continuing the batch on the existent vllm engines.

2. Another prevalent case here is especially for spot instances, it seems that if the spot node goes down, we lose a lot of objects in the object store memory and the job then also fails (will also attach logs). Is there a recommended way to deal with the storage of objects? I'm playing around with a very powerful CPU headnode and trying to allocate storage there but sometimes the spread of objects just distributes across all the nodes in the cluster. Also why would this kill a job?

Logs:

1. Random vllm engine failure logs:

- Map(_postprocess)->Write: Tasks: 0; Actors: 0; Queued blocks: 0; Resources: 0.0 CPU, 0.0B object store: 100%|██████████| 252/252 [59:42<00:00, 12.0s/ row]�[A�[A�[A�[A�[A�[A
- Map(_postprocess)->Write: Tasks: 0; Actors: 0; Queued blocks: 0; Resources: 0.0 CPU, 0.0B object store: 100%|██████████| 252/252 [59:42<00:00, 14.2s/ row]
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #4: c10d::ProcessGroupNCCL::Watchdog::runLoop() + 0x978 (0x7f6c58695f48 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #5: c10d::ProcessGroupNCCL::Watchdog::run() + 0xd2 (0x7f6c58697ec2 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #6: <unknown function> + 0xd8198 (0x7f6cb7589198 in /home/ray/anaconda3/bin/../lib/libstdc++.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #7: <unknown function> + 0x94ac3 (0x7f6cb99d2ac3 in /usr/lib/x86_64-linux-gnu/libc.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #8: clone + 0x44 (0x7f6cb9a63a04 in /usr/lib/x86_64-linux-gnu/libc.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m 
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m Exception raised from run at /pytorch/torch/csrc/distributed/c10d/ProcessGroupNCCL.cpp:2074 (most recent call first):
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #0: c10::Error::Error(c10::SourceLocation, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >) + 0x80 (0x7f6cb517eeb0 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libc10.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #1: <unknown function> + 0xe1c1a1 (0x7f6c5866f1a1 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #2: <unknown function> + 0x9468e6 (0x7f6c581998e6 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #3: <unknown function> + 0xd8198 (0x7f6cb7589198 in /home/ray/anaconda3/bin/../lib/libstdc++.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #4: <unknown function> + 0x94ac3 (0x7f6cb99d2ac3 in /usr/lib/x86_64-linux-gnu/libc.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #5: clone + 0x44 (0x7f6cb9a63a04 in /usr/lib/x86_64-linux-gnu/libc.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m 
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m [rank4]:[E1215 14:21:16.386873484 ProcessGroupNCCL.cpp:2068] [PG ID 2 PG GUID 3 Rank 4] Process group watchdog thread terminated with exception: CUDA error: an illegal memory access was encountered
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m CUDA kernel errors might be asynchronously reported at some other API call, so the stacktrace below might be incorrect.
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m For debugging consider passing CUDA_LAUNCH_BLOCKING=1
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m Compile with `TORCH_USE_CUDA_DSA` to enable device-side assertions.
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m 
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m Exception raised from c10_cuda_check_implementation at /pytorch/c10/cuda/CUDAException.cpp:42 (most recent call first):
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #0: c10::Error::Error(c10::SourceLocation, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >) + 0x80 (0x7fd66cad9eb0 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libc10.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #1: <unknown function> + 0x111c7 (0x7fd66cb6c1c7 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libc10_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #2: c10d::ProcessGroupNCCL::WorkNCCL::finishedGPUExecutionInternal() const + 0x50 (0x7fd610083640 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #3: c10d::ProcessGroupNCCL::WorkNCCL::isCompleted() + 0x68 (0x7fd610092e28 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #4: c10d::ProcessGroupNCCL::Watchdog::runLoop() + 0x978 (0x7fd610095f48 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #5: c10d::ProcessGroupNCCL::Watchdog::run() + 0xd2 (0x7fd610097ec2 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #6: <unknown function> + 0xd8198 (0x7fd66ef3d198 in /home/ray/anaconda3/bin/../lib/libstdc++.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #7: <unknown function> + 0x94ac3 (0x7fd671386ac3 in /usr/lib/x86_64-linux-gnu/libc.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #8: clone + 0x44 (0x7fd671417a04 in /usr/lib/x86_64-linux-gnu/libc.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m 
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m terminate called after throwing an instance of 'c10::DistBackendError'
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m   what():  [PG ID 2 PG GUID 3 Rank 4] Process group watchdog thread terminated with exception: CUDA error: an illegal memory access was encountered
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m CUDA kernel errors might be asynchronously reported at some other API call, so the stacktrace below might be incorrect.
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m For debugging consider passing CUDA_LAUNCH_BLOCKING=1
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m Compile with `TORCH_USE_CUDA_DSA` to enable device-side assertions.
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m 
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m Exception raised from c10_cuda_check_implementation at /pytorch/c10/cuda/CUDAException.cpp:42 (most recent call first):
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #0: c10::Error::Error(c10::SourceLocation, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >) + 0x80 (0x7fd66cad9eb0 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libc10.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #1: <unknown function> + 0x111c7 (0x7fd66cb6c1c7 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libc10_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #2: c10d::ProcessGroupNCCL::WorkNCCL::finishedGPUExecutionInternal() const + 0x50 (0x7fd610083640 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #3: c10d::ProcessGroupNCCL::WorkNCCL::isCompleted() + 0x68 (0x7fd610092e28 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #4: c10d::ProcessGroupNCCL::Watchdog::runLoop() + 0x978 (0x7fd610095f48 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #5: c10d::ProcessGroupNCCL::Watchdog::run() + 0xd2 (0x7fd610097ec2 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #6: <unknown function> + 0xd8198 (0x7fd66ef3d198 in /home/ray/anaconda3/bin/../lib/libstdc++.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #7: <unknown function> + 0x94ac3 (0x7fd671386ac3 in /usr/lib/x86_64-linux-gnu/libc.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #8: clone + 0x44 (0x7fd671417a04 in /usr/lib/x86_64-linux-gnu/libc.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m 
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m Exception raised from run at /pytorch/torch/csrc/distributed/c10d/ProcessGroupNCCL.cpp:2074 (most recent call first):
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #0: c10::Error::Error(c10::SourceLocation, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >) + 0x80 (0x7fd66cad9eb0 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libc10.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #1: <unknown function> + 0xe1c1a1 (0x7fd61006f1a1 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #2: <unknown function> + 0x9468e6 (0x7fd60fb998e6 in /home/ray/anaconda3/lib/python3.11/site-packages/torch/lib/libtorch_cuda.so)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #3: <unknown function> + 0xd8198 (0x7fd66ef3d198 in /home/ray/anaconda3/bin/../lib/libstdc++.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #4: <unknown function> + 0x94ac3 (0x7fd671386ac3 in /usr/lib/x86_64-linux-gnu/libc.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m frame #5: clone + 0x44 (0x7fd671417a04 in /usr/lib/x86_64-linux-gnu/libc.so.6)
�[36m(MapWorker(MapBatches(vLLMEngineStageUDF)) pid=74562, ip=192.168.92.185)�[0m 
2025-12-15 14:21:17,179	ERROR exceptions.py:63 -- Exception occurred in user code, with the abbreviated stack trace below. By default, the Ray Data internal stack trace is omitted from stdout, and only written to the Ray Data log files at `/tmp/ray/session_2025-12-15_13-18-54_437845_1/logs/ray-data`. To output the full stack trace to stdout, set `DataContext.log_internal_stack_trace_to_stdout` to True.
Batch inference failed on attempt 1/3: �[36mray::MapWorker(MapBatches(vLLMEngineStageUDF)).submit()�[39m (pid=74562, ip=192.168.92.185, actor_id=6f12dc41d214b60dc4fa2cae02000000, repr=MapWorker(MapBatches(vLLMEngineStageUDF)))
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/llm/_internal/batch/stages/base.py", line 166, in __call__
    async for output in self.udf(inputs):
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/llm/_internal/batch/stages/vllm_engine_stage.py", line 569, in udf
    request, output, time_taken_llm = await resp
                                      ^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/asyncio/tasks.py", line 615, in _wait_for_one
    return f.result()  # May raise f.exception().
           ^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/llm/_internal/batch/stages/vllm_engine_stage.py", line 325, in generate_async
    output = await self._generate_async(request)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/llm/_internal/batch/stages/vllm_engine_stage.py", line 409, in generate_async_v1
    async for request_output in stream:
  File "/home/ray/anaconda3/lib/python3.11/site-packages/vllm/v1/engine/async_llm.py", line 370, in generate
    q = await self.add_request(
        ^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/vllm/v1/engine/async_llm.py", line 276, in add_request
    raise EngineDeadError()
vllm.v1.engine.exceptions.EngineDeadError: EngineCore encountered an issue. See stack trace (above) for the root cause.

The above exception was the direct cause of the following exception:

�[36mray::MapWorker(MapBatches(vLLMEngineStageUDF)).submit()�[39m (pid=74562, ip=192.168.92.185, actor_id=6f12dc41d214b60dc4fa2cae02000000, repr=MapWorker(MapBatches(vLLMEngineStageUDF)))
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/operators/actor_pool_map_operator.py", line 552, in submit
    yield from _map_task(
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/operators/map_operator.py", line 564, in _map_task
    for b_out in map_transformer.apply_transform(iter(blocks), ctx):
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/operators/map_transformer.py", line 201, in _udf_timed_iter
    output = next(input)
             ^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/operators/map_transformer.py", line 102, in __call__
    yield from self._post_process(results)
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/operators/map_transformer.py", line 84, in _shape_blocks
    for result in results:
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/operators/map_transformer.py", line 361, in _apply_transform
    yield from self._batch_fn(batches, ctx)
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/planner/plan_udf_map_op.py", line 810, in _transform
    raise items
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/planner/plan_udf_map_op.py", line 731, in _reorder
    output_queue.put(await next_task)
                     ^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/planner/plan_udf_map_op.py", line 651, in _apply_udf
    return [out async for out in gen]
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/planner/plan_udf_map_op.py", line 651, in <listcomp>
    return [out async for out in gen]
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/planner/plan_udf_map_op.py", line 402, in _wrapped_udf_map_fn
    _try_wrap_udf_exception(e, item)
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/planner/plan_udf_map_op.py", line 443, in _try_wrap_udf_exception
    raise UserCodeException("UDF failed to process a data block.") from e
ray.exceptions.UserCodeException: UDF failed to process a data block.

Object lost logs:

2025-12-12 12:16:09,975	ERROR exceptions.py:73 -- Exception occurred in Ray Data or Ray Core internal code. If you continue to see this error, please open an issue on the Ray project GitHub page with the full stack trace below: https://github.com/ray-project/ray/issues/new/choose
2025-12-12 12:16:09,975	ERROR exceptions.py:81 -- Full stack trace:
Traceback (most recent call last):
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/exceptions.py", line 49, in handle_trace
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/plan.py", line 533, in execute
    blocks = execute_to_legacy_block_list(
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/legacy_compat.py", line 127, in execute_to_legacy_block_list
    block_list = _bundles_to_block_list(bundles)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/legacy_compat.py", line 175, in _bundles_to_block_list
    bundle_list = list(bundles)
                  ^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/interfaces/executor.py", line 34, in __next__
    return self.get_next()
           ^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/streaming_executor.py", line 680, in get_next
    bundle = state.get_output_blocking(output_split_idx)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/streaming_executor_state.py", line 375, in get_output_blocking
    raise self._exception
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/streaming_executor.py", line 331, in run
    continue_sched = self._scheduling_loop_step(self._topology)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/streaming_executor.py", line 436, in _scheduling_loop_step
    num_errored_blocks = process_completed_tasks(
                         ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/streaming_executor_state.py", line 557, in process_completed_tasks
    raise e from None
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/streaming_executor_state.py", line 524, in process_completed_tasks
    bytes_read = task.on_data_ready(
                 ^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/interfaces/physical_operator.py", line 157, in on_data_ready
    raise ex from None
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/data/_internal/execution/interfaces/physical_operator.py", line 153, in on_data_ready
    ray.get(block_ref)
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/_private/auto_init_hook.py", line 22, in auto_init_wrapper
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/_private/client_mode_hook.py", line 104, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/_private/worker.py", line 2962, in get
    values, debugger_breakpoint = worker.get_objects(
                                  ^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.11/site-packages/ray/_private/worker.py", line 1026, in get_objects
    raise value.as_instanceof_cause()
ray.exceptions.RayTaskError(ObjectLostError): [36mray::MapWorker(MapBatches(vLLMEngineStageUDF)).submit()[39m (pid=370, ip=192.168.6.4, actor_id=ca2ddefc2b91b390984d161002000000, repr=MapWorker(MapBatches(vLLMEngineStageUDF)))
  At least one of the input arguments for this task could not be computed:
ray.exceptions.RayTaskError: [36mray::MapWorker(MapBatches(TokenizeUDF)).submit()[39m (pid=255452, ip=192.168.105.197, actor_id=66ee436a0aedc57d7ee2d33d02000000, repr=MapWorker(MapBatches(TokenizeUDF)))
  At least one of the input arguments for this task could not be computed:
ray.exceptions.RayTaskError: [36mray::MapWorker(MapBatches(ChatTemplateUDF)).submit()[39m (pid=313288, ip=192.168.1.227, actor_id=4b90c5debffc67f074ade23402000000, repr=MapWorker(MapBatches(ChatTemplateUDF)))
  At least one of the input arguments for this task could not be computed:
ray.exceptions.RayTaskError: [36mray::Map(_preprocess)()[39m (pid=167973, ip=192.168.1.227)
  At least one of the input arguments for this task could not be computed:
ray.exceptions.ObjectLostError: Failed to retrieve object 07cd5c469a796c96e2b5c0d2a97a2be95099ccfa0200000002000000. To see information about where this ObjectRef was created in Python, set the environment variable RAY_record_ref_creation_sites=1 during `ray start` and `ray.init()`.

All copies of 07cd5c469a796c96e2b5c0d2a97a2be95099ccfa0200000002000000 have been lost due to node failure. Check cluster logs (`/tmp/ray/session_latest/logs`) for more information about the failure.

Use case

No response

[tgowda1996]

+1

Thanks for creating this jiang! Its been a major issue with my production runs as well. We use 100+ nodes and these cuda failures lead to random job failures.

one way that I have added some resiliency is by adding a checkpointing step. As there is a 1:1 mapping between input and output, I join on some keys and get the rows that are not processed
auto retry is handled using kuberay using backoffLimit

In order to add some more fault tolerance I had considered just writing my own llm framework directly using ray data instead of using ray.data.llm. But if there is official support for this, it will be awesome!!

[richardliaw]

Hey @tgowda1996, @jiangwu300 thanks for raising this!

We recently merged #59105 which should help a lot with 1.
We also are in progress of adding checkpointing (#59409) which should can provide more resilience as well.

[richardliaw]

Another prevalent case here is especially for spot instances, it seems that if the spot node goes down, we lose a lot of objects in the object store memory and the job then also fails (will also attach logs).

I'm not sure why this would be the case. If you need a workaround for this, you could write out the object to object store? it's odd because the lineage should be available all the way back to the driver.

[iamjustinhsu]

@jiangwu300 can you share a bit more of the code that your running? I'm wondering why

ray.data.DataContext.get_current().actor_task_retry_on_errors = True
ray.data.DataContext.get_current().actor_init_max_retries = 3
ray.data.DataContext.get_current().actor_init_retry_on_errors = True
...map_batches(VLM_actor, max_task_retries=3)...

doesn't work? Is it because we don't expose those parameters in ray data llms?

[jiangwu300]

@iamjustinhsu Here's my full code so far:

from __future__ import annotations
import json
import os
import time
from typing import Any, Dict
import boto3
import ray
from ray.data.llm import vLLMEngineProcessorConfig, build_llm_processor
from urllib.parse import urlparse
import argparse
import pandas as pd
from ray.data.expressions import col

def load_json_from_s3(s3_uri: str) -> dict:
    """
    Load a JSON file from an S3 URI into a Python dictionary.
    
    Args:
        s3_uri (str): S3 URI like 's3://bucket-name/path/to/file.json'
        
    Returns:
        dict: Parsed JSON content as a dictionary
    """
    # Parse the S3 URI
    parsed = urlparse(s3_uri)
    bucket = parsed.netloc
    key = parsed.path.lstrip('/')

    # Initialize the S3 client
    s3 = boto3.client('s3')

    # Get the object
    response = s3.get_object(Bucket=bucket, Key=key)
    content = response['Body'].read().decode('utf-8')

    return json.loads(content)

def _load_cfg(arg: str) -> Dict[str, Any]:
    """Interpret *arg* as JSON text, local file, or S3 URI and return a dict."""
    if arg.strip().startswith("{"):
        return json.loads(arg)

    if arg.startswith("s3://"):
        return load_json_from_s3(arg)

    with open(arg, "rt") as f:
        return json.load(f)


def _normalise_cfg(cfg: Dict[str, Any]) -> Dict[str, Any]:
    """Unify synonyms and fill defaults so downstream code is stable."""
    if "s3_prompts_path" not in cfg:
        raise KeyError("config must include 's3_prompts_path'")

    # derive default output path if needed
    if "s3_output_path" not in cfg:
        in_uri = cfg["s3_prompts_path"]
        base, ext = os.path.splitext(in_uri)
        cfg["s3_output_path"] = f"{base}_results"

    # sane defaults
    cfg.setdefault("instances", 1)
    cfg.setdefault("apply_chat_template", True)
    cfg.setdefault("batch_size", 128)
    cfg.setdefault("runtime_env", {})
    cfg.setdefault("engine_kwargs", {})
    cfg.setdefault("sampling_params", {})
    cfg.setdefault("task", "generate")
    return cfg


def run_batch_inference(cfg: Dict[str, Any]) -> None:
    start = time.time()

    # 1) Resolve I/O paths
    in_s3 = cfg["s3_prompts_path"]
    out_s3 = cfg["s3_output_path"]
    # logs_path = cfg["s3_logs_path"]
    instances = int(cfg["instances"])
    batch_size = int(cfg.get("batch_size", 128))
    task_type = cfg.get("task_type", "generate")
    detokenize = True if task_type in ("generate", "chat") else False
    apply_chat_template = True if task_type in ("generate", "chat") else False
    max_job_instances = 16

    id_col = cfg.get("id_col", "prompt_id") # change to default later
    ds = ray.data.read_parquet(in_s3)

    try:
        filter_ds = ray.data.read_parquet(out_s3, columns=[id_col])

        # Only do join if filter_ds is non-empty (optional, but often useful)
        if filter_ds.count() > 0:
            ds = ds.join(
                filter_ds,
                join_type="left_anti",
                num_partitions=8,
                on=(id_col,),
            )
            
        else:
            print("Filter dataset is empty; skipping filtering.")

    except Exception as e:
        print(f"Could not read filter dataset from {out_s3}: {e}")
        print("Proceeding without filtering.")
    
    prompts_ds = ds.materialize()

    model_source = cfg["model_source"]
    engine_kwargs = cfg["engine_kwargs"]
    if model_source.startswith("s3://"):
        engine_kwargs["load_format"] = "runai_streamer"

    runtime_env = cfg.get("runtime_env", {})

    engine_cfg = vLLMEngineProcessorConfig(
        model_source=model_source,
        runtime_env=runtime_env,
        task_type=task_type,
        engine_kwargs=engine_kwargs,
        concurrency=(instances, max_job_instances),
        batch_size=batch_size,
        max_concurrent_batches=8,
        detokenize=detokenize,
        apply_chat_template=apply_chat_template,
    )

    if task_type == "embed":
        print("Runing embedding batch inference …")
        outputs_to_exclude = set([
            'prompt_token_ids', 'generated_tokens', 'prompt',
            'generated_text', 'num_generated_tokens', 'metrics',
            'request_id', 'batch_uuid', 'time_taken_llm', 'params'
            ]
        )
        processor = build_llm_processor(
            engine_cfg,
            preprocess=lambda row: dict(prompt=row["prompt"]),
            postprocess=lambda row: dict(**{k: v for k, v in row.items() if k not in outputs_to_exclude})
        )
    
    else:
        print("Runing generation/chat batch inference …")
        outputs_to_exclude = set(["batch_uuid", "embeddings", "messages", "metrics", "generated_tokens", 
                            "params", "prompt_token_ids", "request_id", "time_taken_llm", "prompt", 
                            "system"])
        processor = build_llm_processor(
            engine_cfg,
            preprocess=lambda row: dict(
                messages=[
                    {"role": "system", "content": row["system"]},
                    {"role": "user", "content": row["prompt"]},
                ],
                sampling_params=cfg["sampling_params"],
            ),
            postprocess=lambda row: dict(**{k: v for k, v in row.items() if k not in outputs_to_exclude}),
        )

    # 4) Execute & write results
    print("Running batch inference …")
    processed = processor(prompts_ds)
    processed.write_parquet(out_s3)
    duration = time.time() - start
    print(f"Completed in {duration:,.1f} s → {out_s3}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Run Ray LLM batch inference job")
    parser.add_argument(
        "--config",
        required=True,
        help="Path or S3 URI to JSON config file",
    )
    args = parser.parse_args()
    retries = 3

    for attempt in range(1, retries + 1):
        print(f"On retry {attempt}")
        try:
            ray.init(address="auto")
            cfg_raw = _load_cfg(args.config)
            cfg = _normalise_cfg(cfg_raw)
            run_batch_inference(cfg)
            # success: break out of the loop
            break
        except Exception as e:
            print(f"Batch inference failed on attempt {attempt}/{retries}: {e}")
            if attempt == retries:
                # after last attempt, re-raise or wrap
                raise ValueError(f"Batch inference failed after {retries} attempts") from e
        finally:
            # cleanup ray resources
            ray.shutdown()

[tgowda1996]

@jiangwu300 can you share a bit more of the code that your running? I'm wondering why

ray.data.DataContext.get_current().actor_task_retry_on_errors = True
ray.data.DataContext.get_current().actor_init_max_retries = 3
ray.data.DataContext.get_current().actor_init_retry_on_errors = True
...map_batches(VLM_actor, max_task_retries=3)...
doesn't work? Is it because we don't expose those parameters in ray data llms?

ya we do not expose the map batch kwargs in the data llm wrapper -

ray/python/ray/llm/_internal/batch/processor/vllm_engine_proc.py

Line 253 in 0d21f57

map_batches_kwargs=dict(

I feel like we can probably add map batches kwargs as an attribute to the new nested stage configs that were added recently. It will give power users more control.

[jeffreywang-anyscale]

If I understand correctly, DataContext.actor_init_max_retries seems to be orthogonal to restarting a died actor in an actor pool. It controls Ray Data's level retry behavior for actor initialization failures and does not apply when an actor crashes or exits unexpectedly in the middle of execution.

In Ray Data LLM, ray.data.DataContext.get_current() has both actor_init_retry_on_errors and actor_task_retry_on_errors set to False. We can set them to True with the following code as DataContext is implemented as a singleton.

ctx = ray.data.DataContext.get_current()
ctx.actor_init_retry_on_errors = True
ctx.actor_task_retry_on_errors = True

max_restarts and max_task_retries are the knobs in Ray Core's remote args that ensure fault tolerance when an actor dies unexpectedly in the middle of an execution. These are set to -1 by default in the ActorPoolMapOperator which vLLMEngineStage builds on. I also confirmed that these values are -1 by stepping through a debugger.

[wsluma]

how would the ctx.actor_init_retry_on_errors behave when for vllm autoscaling is default to false?

ray/python/ray/llm/_internal/batch/processor/vllm_engine_proc.py

Line 287 in ac15dba

**config.get_concurrency(autoscaling_enabled=False),

iiuc even if we have actor init retry, the processing won't start until we reach the concurrency number set at initialization?
if an actor fail to initialize, will the processing continue ignoring the original concurrency number set?