adu2021 Taming the Chaos: Coordinated Autoscaling for Disaggregated LLM Inference
Core Concept
HeteroScale addresses the autoscaling challenge in Prefill-Decode (P/D) disaggregated inference architectures. Traditional autoscaling treats prefill and decode independently, creating bottlenecks. HeteroScale combines topology-aware scheduling (accounting for network and hardware constraints) with a single metric-driven policy that jointly scales both stages, achieving 26.6 percentage point GPU utilization improvement.
Architecture Overview
- Prefill-Decode Disaggregation: Separate resource pools for prompt processing and token generation
- Topology-Aware Scheduler: Routes requests considering hardware heterogeneity and network latency
- Metric-Driven Joint Scaling: Single signal scales both prefill and decode coherently
- Production Infrastructure: Operates on tens of thousands of GPUs
- SLO Preservation: Maintains latency targets while improving utilization
Implementation Steps
Stage 1: Model Prefill-Decode Infrastructure
Understand P/D disaggregated serving architecture.
# Prefill-Decode disaggregated serving
from dataclasses import dataclass
from typing import List, Dict, Tuple
import time
@dataclass
class Request:
"""LLM serving request"""
request_id: str
prompt_tokens: int
output_length: int
arrival_time: float
deadline: float # Latency SLO
priority: int = 0
@dataclass
class PrefillWorker:
"""Worker specialized for prompt processing"""
worker_id: str
gpu_memory: int # GB
available_memory: int
throughput: float # prompts/sec
@dataclass
class DecodeWorker:
"""Worker specialized for token generation"""
worker_id: str
gpu_memory: int
available_memory: int
throughput: float # tokens/sec
batch_size_limit: int
class DisaggregatedServingCluster:
"""Manage P/D separated serving cluster"""
def __init__(
self,
num_prefill_gpus: int = 1000,
num_decode_gpus: int = 2000
):
self.prefill_workers = self._init_workers(num_prefill_gpus, PrefillWorker)
self.decode_workers = self._init_workers(num_decode_gpus, DecodeWorker)
self.request_queue = []
self.prefill_queue = []
self.decode_queue = []
def _init_workers(self, num_gpus: int, worker_class):
"""Initialize worker pool"""
workers = []
for i in range(num_gpus):
worker = worker_class(
worker_id=f"{worker_class.__name__}_{i}",
gpu_memory=40, # 40GB per GPU
available_memory=40,
throughput=1000.0 # prompts/sec or tokens/sec
)
if worker_class == DecodeWorker:
worker.batch_size_limit = 2048
workers.append(worker)
return workers
def schedule_request(self, request: Request) -> Dict:
"""
Schedule request through P/D pipeline.
1. Route to prefill worker
2. Route to decode worker(s)
3. Track end-to-end latency
"""
start_time = time.time()
# Stage 1: Prefill (process prompt)
prefill_worker = self.select_prefill_worker(request)
if not prefill_worker:
return {"status": "queued", "reason": "no_prefill_capacity"}
prefill_latency = self.prefill_latency(request.prompt_tokens)
# Stage 2: Decode (generate tokens)
decode_workers = self.select_decode_workers(request.output_length)
if not decode_workers:
return {"status": "queued", "reason": "no_decode_capacity"}
decode_latency = self.decode_latency(request.output_length, len(decode_workers))
total_latency = prefill_latency + decode_latency
return {
"status": "scheduled",
"prefill_worker": prefill_worker.worker_id,
"decode_workers": [w.worker_id for w in decode_workers],
"estimated_latency": total_latency,
"meets_slo": total_latency <= request.deadline
}
def select_prefill_worker(self, request: Request) -> PrefillWorker:
"""Select least-loaded prefill worker"""
eligible = [w for w in self.prefill_workers if w.available_memory >= 4]
if not eligible:
return None
return min(eligible, key=lambda w: len(self.prefill_queue))
def select_decode_workers(self, output_length: int) -> List[DecodeWorker]:
"""Select decode workers for output generation"""
eligible = [w for w in self.decode_workers if w.available_memory >= 2]
if not eligible:
return []
# Sort by utilization
sorted_workers = sorted(eligible, key=lambda w: len(self.decode_queue))
return sorted_workers[:max(1, output_length // 256)]
def prefill_latency(self, num_tokens: int) -> float:
"""Estimate prefill latency"""
return num_tokens / 10000 # Simple model: 10k tokens/sec
def decode_latency(self, output_length: int, num_workers: int) -> float:
"""Estimate decode latency with batching"""
return output_length / (100 * num_workers) # 100 tokens/sec per worker
Stage 2: Topology-Aware Scheduler
Route requests considering hardware heterogeneity and network constraints.
# Topology-aware scheduling
import networkx as nx
class ClusterTopology:
"""Model cluster network and hardware topology"""
def __init__(self):
# Build network graph
self.topology = nx.DiGraph()
self.gpu_network_latency = {} # (gpu1, gpu2) -> latency_ms
def add_gpu(self, gpu_id: str, rack: str, zone: str):
"""Add GPU to topology"""
self.topology.add_node(
gpu_id,
rack=rack,
zone=zone,
latency_to_network=1.0 # ms
)
def estimate_latency(self, gpu1: str, gpu2: str) -> float:
"""Estimate network latency between GPUs"""
node1 = self.topology.nodes[gpu1]
node2 = self.topology.nodes[gpu2]
# Same GPU
if gpu1 == gpu2:
return 0.0
# Same rack: low latency
if node1["rack"] == node2["rack"]:
return 0.2 # ms
# Same zone: medium latency
elif node1["zone"] == node2["zone"]:
return 1.0 # ms
# Different zone: high latency
else:
return 5.0 # ms
class TopologyAwareScheduler:
"""Schedule requests respecting topology constraints"""
def __init__(self, topology: ClusterTopology):
self.topology = topology
def place_prefill_and_decode(
self,
request: Request,
prefill_workers: List[PrefillWorker],
decode_workers: List[DecodeWorker]
) -> Dict:
"""
Co-locate prefill and decode workers considering:
- Network latency
- GPU availability
- Load balancing
"""
# Step 1: Find best prefill worker
best_prefill_score = float('inf')
best_prefill = None
for pf_worker in prefill_workers:
# Cost: network utilization + queue depth
queue_cost = len([r for r in request_queue if r.assigned_prefill == pf_worker.worker_id])
mem_cost = pf_worker.gpu_memory - pf_worker.available_memory
score = 0.3 * queue_cost + 0.7 * mem_cost
if score < best_prefill_score:
best_prefill_score = score
best_prefill = pf_worker
# Step 2: Find decode workers co-located with prefill
decode_candidates = []
for d_worker in decode_workers:
# Prefer workers in same rack/zone
network_cost = self.topology.estimate_latency(
best_prefill.worker_id,
d_worker.worker_id
)
queue_cost = len([r for r in request_queue if d_worker in r.assigned_decode])
total_cost = 0.4 * network_cost + 0.6 * queue_cost
decode_candidates.append((d_worker, total_cost))
# Sort by cost and select top workers
decode_candidates.sort(key=lambda x: x[1])
selected_decode = [w for w, _ in decode_candidates[:max(1, request.output_length // 256)]]
return {
"prefill_worker": best_prefill,
"decode_workers": selected_decode,
"network_latency": sum(
self.topology.estimate_latency(best_prefill.worker_id, d.worker_id)
for d in selected_decode
) / len(selected_decode)
}
Stage 3: Metric-Driven Autoscaling Policy
Single metric scales both prefill and decode stages coherently.
# Metric-driven autoscaling
import math
class AutoscalingMetric:
"""Unified metric for P/D scaling decisions"""
def __init__(self):
self.prefill_queue_depth = 0
self.decode_queue_depth = 0
self.prefill_utilization = 0.0
self.decode_utilization = 0.0
def compute_scaling_signal(self) -> float:
"""
Compute single metric to drive scaling.
Signal = f(queue_depth, utilization, latency_headroom)
Higher signal = scale up
"""
# Queue depth contribution (0-1)
max_queue = 1000
queue_signal = min(
(self.prefill_queue_depth + self.decode_queue_depth) / max_queue,
1.0
)
# Utilization contribution (0-1)
avg_utilization = (self.prefill_utilization + self.decode_utilization) / 2
utilization_signal = avg_utilization
# Combined signal
scaling_signal = 0.4 * queue_signal + 0.6 * utilization_signal
return scaling_signal
class MetricDrivenAutoscaler:
"""Scale P/D resources based on unified metric"""
def __init__(self, cluster: DisaggregatedServingCluster):
self.cluster = cluster
self.metric = AutoscalingMetric()
self.target_utilization = 0.75 # Target GPU utilization
def update_metrics(self, cluster_state: Dict):
"""Update autoscaling metrics from cluster state"""
self.metric.prefill_queue_depth = cluster_state.get("prefill_queue_len", 0)
self.metric.decode_queue_depth = cluster_state.get("decode_queue_len", 0)
self.metric.prefill_utilization = cluster_state.get("prefill_util", 0.0)
self.metric.decode_utilization = cluster_state.get("decode_util", 0.0)
def make_scaling_decision(self) -> Dict:
"""
Decide whether to scale prefill and decode.
Key insight: scale both together to maintain balance.
"""
signal = self.metric.compute_scaling_signal()
# Scaling thresholds
scale_up_threshold = 0.75
scale_down_threshold = 0.25
decision = {
"scale_up": False,
"scale_down": False,
"prefill_delta": 0,
"decode_delta": 0,
"signal": signal
}
if signal > scale_up_threshold:
# Scale up: maintain 2:1 decode to prefill ratio
decision["scale_up"] = True
decision["prefill_delta"] = 100 # Add 100 prefill GPUs
decision["decode_delta"] = 200 # Add 200 decode GPUs
elif signal < scale_down_threshold and self.can_scale_down():
# Scale down
decision["scale_down"] = True
decision["prefill_delta"] = -50
decision["decode_delta"] = -100
return decision
def can_scale_down(self) -> bool:
"""Check if safe to scale down (maintain SLOs)"""
# Never scale below minimum
if (len(self.cluster.prefill_workers) <= 100 or
len(self.cluster.decode_workers) <= 200):
return False
# Check latency headroom
return self.metric.compute_scaling_signal() < 0.25
def apply_scaling(self, decision: Dict):
"""Apply scaling decision to cluster"""
if decision["prefill_delta"] > 0:
for _ in range(decision["prefill_delta"]):
self.cluster.prefill_workers.append(
self._create_prefill_worker()
)
elif decision["prefill_delta"] < 0:
for _ in range(-decision["prefill_delta"]):
if self.cluster.prefill_workers:
self.cluster.prefill_workers.pop()
if decision["decode_delta"] > 0:
for _ in range(decision["decode_delta"]):
self.cluster.decode_workers.append(
self._create_decode_worker()
)
elif decision["decode_delta"] < 0:
for _ in range(-decision["decode_delta"]):
if self.cluster.decode_workers:
self.cluster.decode_workers.pop()
def _create_prefill_worker(self) -> PrefillWorker:
worker_id = f"prefill_{len(self.cluster.prefill_workers)}"
return PrefillWorker(
worker_id=worker_id,
gpu_memory=40,
available_memory=40,
throughput=1000.0
)
def _create_decode_worker(self) -> DecodeWorker:
worker_id = f"decode_{len(self.cluster.decode_workers)}"
return DecodeWorker(
worker_id=worker_id,
gpu_memory=40,
available_memory=40,
throughput=100.0,
batch_size_limit=2048
)
Stage 4: Production Autoscaling Loop
Implement the complete serving loop with autoscaling.
# Production autoscaling loop
import time
class ProductionServing:
"""Production LLM serving with autoscaling"""
def __init__(self):
self.cluster = DisaggregatedServingCluster()
self.topology = ClusterTopology()
self.scheduler = TopologyAwareScheduler(self.topology)
self.autoscaler = MetricDrivenAutoscaler(self.cluster)
self.metrics_history = {
"gpu_utilization": [],
"gpu_hours_saved": [],
"request_latency": []
}
def serving_loop(self, duration_seconds: int = 86400):
"""Main serving loop"""
start_time = time.time()
check_interval = 60 # Check autoscaling every minute
while time.time() - start_time < duration_seconds:
# Process incoming requests
incoming_requests = self.get_incoming_requests()
for request in incoming_requests:
self.process_request(request)
# Update metrics
cluster_state = self.measure_cluster_state()
self.autoscaler.update_metrics(cluster_state)
# Make scaling decision
decision = self.autoscaler.make_scaling_decision()
# Apply scaling
if decision["scale_up"] or decision["scale_down"]:
self.autoscaler.apply_scaling(decision)
print(f"Scaling decision: {decision}")
# Log metrics
self.metrics_history["gpu_utilization"].append(
cluster_state["overall_util"]
)
# Sleep until next check
time.sleep(check_interval)
# Report results
return self.generate_report()
def get_incoming_requests(self) -> List[Request]:
"""Simulate incoming requests"""
return [] # Placeholder
def process_request(self, request: Request):
"""Process single request through P/D pipeline"""
schedule = self.scheduler.place_prefill_and_decode(
request,
self.cluster.prefill_workers,
self.cluster.decode_workers
)
def measure_cluster_state(self) -> Dict:
"""Measure current cluster state"""
prefill_util = sum(
(w.gpu_memory - w.available_memory) / w.gpu_memory
for w in self.cluster.prefill_workers
) / len(self.cluster.prefill_workers) if self.cluster.prefill_workers else 0
decode_util = sum(
(w.gpu_memory - w.available_memory) / w.gpu_memory
for w in self.cluster.decode_workers
) / len(self.cluster.decode_workers) if self.cluster.decode_workers else 0
return {
"prefill_util": prefill_util,
"decode_util": decode_util,
"overall_util": (prefill_util + decode_util) / 2,
"prefill_queue_len": len(self.cluster.prefill_queue),
"decode_queue_len": len(self.cluster.decode_queue)
}
def generate_report(self) -> Dict:
"""Generate serving report"""
avg_util = sum(self.metrics_history["gpu_utilization"]) / len(
self.metrics_history["gpu_utilization"]
) if self.metrics_history["gpu_utilization"] else 0
return {
"avg_gpu_utilization": avg_util,
"utilization_improvement": 0.266, # 26.6% from paper
"gpu_hours_saved_daily": 500000, # Hundreds of thousands from paper
"maintained_slos": True
}
Practical Guidance
Deployment Configuration
- Prefill-Decode Ratio: Maintain 1:2 ratio (1 prefill per 2 decode GPUs)
- Scaling Cooldown: 5-10 minutes between scaling operations to avoid flapping
- Metric Update Frequency: Every 1-2 minutes
- Queue Depth Threshold: Scale up if queue exceeds 800 requests
Network Topology
- Co-locate prefill and decode workers in same racks when possible
- Use high-bandwidth inter-rack links for P/D communication
- Monitor network latency; add capacity if > 5ms between critical paths
When to Use HeteroScale
- Large-scale LLM serving (1000+ GPUs)
- Variable request patterns (scheduling can adapt)
- Cost-sensitive deployments (maximizes GPU utilization)
- Multi-tenant clusters with diverse workloads
When NOT to Use
- Small clusters (<100 GPUs)
- Monolithic LLM serving (not disaggregated)
- Ultra-low latency requirements (autoscaling adds complexity)
Performance Expectations
- GPU Utilization Improvement: +26.6 percentage points
- GPU-Hours Saved: Hundreds of thousands daily at scale
- Latency Impact: Minimal if topology well-designed
- Scaling Overhead: <2% reduction in throughput during scale operations
Reference
Taming the Chaos: Coordinated Autoscaling for Disaggregated LLM Inference. arXiv:2508.19559