[Data] Actor pools cannot scale down to zero in multi-step pipelines with downstream bottleneck

Problem

In Ray Data pipelines with multiple steps where earlier steps use expensive resources (e.g., GPUs) and later steps become a bottleneck, actors from earlier steps cannot be released to free up resources.

Specific Issue

The ActorPoolStrategy enforces min_size >= 1 (line 152 in python/ray/data/_internal/compute.py), which prevents actor pools from scaling down to zero even when they are completely idle.

if min_size is not None and min_size < 1:
    raise ValueError("min_size must be >= 1", min_size)

This causes resource wastage in pipelines with the following pattern:

• Steps 1-N: Expensive operations (e.g., GPU-accelerated) that complete quickly
• Step N+1: CPU-bound bottleneck that processes slowly

Concrete Example

# 8-step pipeline: first 7 steps are GPU-heavy, step 8 is slow CPU processing
ds = ray.data.range(10000, parallelism=100)

# Steps 1-7: GPU operations
for i in range(7):
    ds = ds.map_batches(
        gpu_transform_fn,
        compute=ray.data.ActorPoolStrategy(
            min_size=1,  # Cannot be 0!
            max_size=4,
            initial_size=2
        ),
        num_cpus=0, num_gpus=1
    )

# Step 8: Slow CPU bottleneck (takes 100x longer)
ds = ds.map_batches(
    slow_cpu_fn,
    compute=ray.data.ActorPoolStrategy(min_size=1, max_size=2),
    num_cpus=2
)

result = ds.collect()  # GPU actors from steps 1-7 remain allocated but idle

Expected Behavior

When Step 8 becomes a bottleneck:

1. Steps 1-7 complete their input processing
2. Their outputs are queued waiting for Step 8 to process
3. The GPU actor pools should automatically scale down to 0 (releasing all GPU resources)
4. As Step 8 processes and requests more data, Steps 1-7 should scale up actors on demand

Actual Behavior

• GPU actors from Steps 1-7 remain allocated (at least 1 per operator)
• These idle actors occupy GPU memory/compute that could be used elsewhere
• The min_size >= 1 constraint prevents releasing all actors to the cluster

Root Cause

Two related issues in ActorPoolMapOperator and default_actor_autoscaler.py:

1. Hard constraint in ActorPoolStrategy.__init__ (compute.py:151-152):

   • Rejects min_size=0 at configuration time
   • No way to allow pools to scale to zero

2. Autoscaler protects min_size (default_actor_autoscaler.py:157-159):

   python复制

   if actor_pool.current_size() <= actor_pool.min_size():
       return ActorPoolScalingRequest.no_op(reason="reached min size")

   • Even with utilization at 0%, cannot scale below min_size

3. No pipeline-aware idle detection:

   • Autoscaler only forces full scale-down when op.has_completed()
   • It doesn't detect: "inputs complete + outputs enqueued + utilization=0"

Impact

• GPU Waste: In GPU-heavy pipelines, this can waste 10-50+ GPUs per bottleneck
• Memory Waste: GPU memory stays allocated even when unused
• User Confusion: Users expect autoscaling to free unused resources

Proposed Solutions

Solution 1: Allow min_size=0 (Recommended)

Remove the min_size >= 1 constraint and implement lazy actor initialization:

# compute.py
if min_size is not None and min_size < 0:  # Changed from < 1
    raise ValueError("min_size must be >= 0", min_size)

# autoscaling_actor_pool.py
assert self.min_size >= 0  # Changed from >= 1

Tradeoff: First task will have startup latency. Mitigation: Implement actor warm-up.

Solution 2: Introduce idle_timeout_s parameter

compute=ActorPoolStrategy(
    min_size=1,
    max_size=4,
    idle_timeout_s=30,  # Release all actors after 30s idle
)

Solution 3: Pipeline-aware idle detection

Enhance default_actor_autoscaler.py to detect:

• op._inputs_complete == True
• op_state.total_enqueued_input_blocks() == 0
• actor_pool.current_size() > 0
• No tasks running

Then force scale-down to 0 even if min_size > 0.

Additional Context

• Affected Files:

  • python/ray/data/_internal/compute.py (line 152)
  • python/ray/data/_internal/actor_autoscaler/autoscaling_actor_pool.py (line 62)
  • python/ray/data/_internal/actor_autoscaler/default_actor_autoscaler.py (lines 157-159)
  • python/ray/data/_internal/execution/operators/actor_pool_map_operator.py (line 544)

• Related Issues: Pipeline autoscaling, resource utilization, GPU memory management

Workarounds

Until this is fixed:

1. Use .persist() to separate pipeline stages:

   python复制

   ds = (ds
         .map_batches(gpu_fn_1)
         .map_batches(gpu_fn_2)
         .persist())  # Force execution and release GPU actors
   
   ds = ds.map_batches(slow_fn)
   result = ds.collect()

2. Minimize min_size and initial_size:

   python复制

   ActorPoolStrategy(min_size=1, max_size=4, initial_size=1)

   (Still wastes 1 GPU per operator, but minimizes waste)

3. Use task pool for non-stateful operations:

   python复制

   ds = ds.map_batches(stateless_fn, compute="tasks")  # No actor overhead