Strategy Routing Pattern: ComputationCoordinator's 7-Level Priority Dispatch
In a complex ontology-driven decision platform, computation requests come in every variety:
CoomiaPublished on December 16, 202512 min read
Share this articleTwitter / X
Strategy Routing Pattern: ComputationCoordinator's 7-Level Priority Dispatch
“Series: S10 Design Patterns · Article 2 | Level: Advanced | Reading Time: 18 min
#TL;DR
- The Strategy Routing Pattern elevates request dispatch from hardcoded if-else blocks into a configurable, extensible priority routing chain.
- coomia-dip's ComputationCoordinator implements 7-level priority scheduling: from emergency real-time decisions to low-priority batch analytics, each level with independent resource quotas, timeout policies, and degradation strategies.
- This pattern ensures high-priority computation SLAs are met even under extreme load, while idle resources are fully utilized for lower-priority tasks.
#Introduction: When a Request Doesn't Know Where to Go
In a complex ontology-driven decision platform, computation requests come in every variety:
Code
Real-time risk assessment → Must return within 50ms
Dashboard refresh → 500ms acceptable
Reasoning chain execution → May need 10 seconds
Batch property derivation → Can queue for minutes
Data backfill → Run overnight
The traditional approach is a wall of if-else in the API gateway:
Python
if request.type == "realtime_risk":
route_to(fast_pool)
elif request.type == "dashboard":
route_to(medium_pool)
elif request.type == "batch":
route_to(slow_pool)
# ... add a branch for every new type
The problems are obvious: routing logic is tightly coupled to business types, every new computation type requires code changes and redeployment. Worse, when the system overloads, all requests are rejected equally — critical decisions and optional analytics perish together.
The core idea of the Strategy Routing Pattern is: model routing decisions as an independent, pluggable strategy chain where priority, resource state, and business context collectively determine request destination.
#Part 1: Defining the Strategy Routing Pattern
#1.1 Pattern Structure
The Strategy Routing Pattern comprises four core components:
Code
┌──────────────────────────────────────────────────────┐
│ ComputationCoordinator │
│ │
│ ┌─────────────┐ ┌──────────────┐ ┌───────────┐ │
│ │ Priority │ │ Resource │ │ Strategy │ │
│ │ Classifier │──▶│ Allocator │──▶│ Router │ │
│ └─────────────┘ └──────────────┘ └───────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────┐ ┌──────────────┐ ┌───────────┐ │
│ │ Context │ │ Quota │ │ Fallback │ │
│ │ Evaluator │ │ Manager │ │ Chain │ │
│ └─────────────┘ └──────────────┘ └───────────┘ │
└──────────────────────────────────────────────────────┘
- Priority Classifier: Determines request priority based on metadata, source, and business context.
- Resource Allocator: Allocates compute resources based on current utilization and quota policies.
- Strategy Router: Selects the optimal execution path based on priority and available resources.
- Context Evaluator: Evaluates business context (tenant SLA, operational urgency) to adjust priority.
- Quota Manager: Manages resource quotas and token buckets per priority tier.
- Fallback Chain: Degrades gracefully through alternatives when the preferred path is unavailable.
#1.2 Differences from the Classic Strategy Pattern
The classic GoF Strategy Pattern focuses on "interchangeable algorithm implementations behind a common interface." The Strategy Routing Pattern goes further:
| Dimension | Classic Strategy | Strategy Routing |
|---|---|---|
| Focus | Algorithm swap | Request dispatch |
| Selection basis | Static config | Dynamic multi-factor |
| Degradation | None | Built-in fallback chain |
| Resource awareness | None | Quota-aware |
| Priority | None | Multi-level priority |
#Part 2: ComputationCoordinator's 7-Level Priority System
#2.1 Priority Definitions
coomia-dip's ComputationCoordinator defines seven priority levels, each with explicit SLAs and resource guarantees:
Python
class ComputationPriority(IntEnum):
"""7-level priority for computation requests."""
P0_EMERGENCY = 0 # Emergency decisions: real-time risk, alert response
P1_INTERACTIVE = 1 # Interactive ops: user-triggered Actions
P2_REALTIME = 2 # Real-time analytics: dashboards, live aggregation
P3_NEAR_REALTIME = 3 # Near real-time: reasoning chains, rule engine
P4_STANDARD = 4 # Standard: derived property calc, materialized view refresh
P5_BACKGROUND = 5 # Background: data quality checks, index rebuilds
P6_BATCH = 6 # Batch: data backfill, historical analysis
#2.2 SLA Contracts Per Level
YAML
priority_sla:
P0_EMERGENCY:
max_latency: 50ms
resource_guarantee: 20% # Always reserve 20% of resources
preemptible: false
retry_policy: immediate_3x
fallback: in_memory_cache
P1_INTERACTIVE:
max_latency: 200ms
resource_guarantee: 15%
preemptible: false
retry_policy: exponential_3x
fallback: degraded_result
P2_REALTIME:
max_latency: 500ms
resource_guarantee: 15%
preemptible: false
retry_policy: exponential_3x
fallback: stale_cache
P3_NEAR_REALTIME:
max_latency: 5s
resource_guarantee: 15%
preemptible: true # Can be preempted by P0-P2
retry_policy: exponential_5x
fallback: async_queue
P4_STANDARD:
max_latency: 30s
resource_guarantee: 10%
preemptible: true
retry_policy: queue_retry
fallback: delayed_execution
P5_BACKGROUND:
max_latency: 5min
resource_guarantee: 5%
preemptible: true
retry_policy: scheduled_retry
fallback: next_window
P6_BATCH:
max_latency: best_effort
resource_guarantee: 0% # Uses only idle resources
preemptible: true
retry_policy: daily_retry
fallback: skip_and_log
#2.3 Priority Classification Algorithm
Request priority is not fixed but dynamically computed from multiple factors:
Python
class PriorityClassifier:
"""Classify computation requests into priority levels."""
def classify(self, request: ComputationRequest) -> ComputationPriority:
base_priority = self._get_base_priority(request.computation_type)
# Factor 1: Tenant SLA tier adjustment
tenant_adjustment = self._tenant_sla_adjustment(request.tenant_id)
# Factor 2: Operational context adjustment
context_adjustment = self._context_adjustment(request.context)
# Factor 3: System load adjustment
load_adjustment = self._load_adjustment()
# Factor 4: Time window adjustment (e.g., boost during trading hours)
time_adjustment = self._time_window_adjustment(request.timestamp)
final_score = (
base_priority.value
+ tenant_adjustment
+ context_adjustment
+ load_adjustment
+ time_adjustment
)
return ComputationPriority(
max(0, min(6, round(final_score)))
)
def _tenant_sla_adjustment(self, tenant_id: str) -> float:
sla = self.tenant_registry.get_sla(tenant_id)
return {
"platinum": -1.0, # Platinum tenants get one-level boost
"gold": -0.5,
"silver": 0.0,
"bronze": 0.5,
}.get(sla.tier, 0.0)
def _context_adjustment(self, context: dict) -> float:
adjustment = 0.0
if context.get("triggered_by") == "alert":
adjustment -= 2.0 # Alert-triggered computations get major boost
if context.get("is_retry"):
adjustment -= 0.5 # Retries get moderate boost
if context.get("user_waiting"):
adjustment -= 1.0 # User is waiting for result
return adjustment
#Part 3: Resource Allocation and Quota Management
#3.1 Token Bucket Algorithm
Each priority level uses an independent token bucket to control concurrency:
Python
class PriorityQuotaManager:
"""Manage resource quotas per priority level using token buckets."""
def __init__(self, config: QuotaConfig):
self.buckets: dict[ComputationPriority, TokenBucket] = {}
for priority in ComputationPriority:
bucket_config = config.get_bucket_config(priority)
self.buckets[priority] = TokenBucket(
capacity=bucket_config.max_concurrent,
refill_rate=bucket_config.refill_per_second,
burst_allowance=bucket_config.burst_factor,
)
def try_acquire(self, priority: ComputationPriority) -> bool:
"""Try to acquire a token for the given priority."""
if self.buckets[priority].try_consume(1):
return True
# High priority can preempt lower priority resources
if priority.value <= 2: # P0-P2 can preempt
return self._try_preempt(priority)
return False
def _try_preempt(self, priority: ComputationPriority) -> bool:
"""Preempt lower priority tasks to free resources."""
for lower in reversed(ComputationPriority):
if lower.value <= priority.value:
continue
preempted = self.preemption_manager.preempt_lowest(
pool=lower, count=1
)
if preempted:
return True
return False
#3.2 Dynamic Quota Adjustment
The system dynamically adjusts resource proportions per tier based on actual load:
Python
class DynamicQuotaAdjuster:
"""Adjust quotas based on real-time system load."""
def adjust(self, metrics: SystemMetrics) -> dict[ComputationPriority, float]:
"""Return adjusted quota percentages."""
if metrics.cpu_usage > 0.8:
# High load: shrink low priority, protect high priority
return {
ComputationPriority.P0_EMERGENCY: 0.30,
ComputationPriority.P1_INTERACTIVE: 0.25,
ComputationPriority.P2_REALTIME: 0.20,
ComputationPriority.P3_NEAR_REALTIME: 0.15,
ComputationPriority.P4_STANDARD: 0.07,
ComputationPriority.P5_BACKGROUND: 0.03,
ComputationPriority.P6_BATCH: 0.00,
}
elif metrics.cpu_usage < 0.3:
# Low load: relax low priority quotas
return {
ComputationPriority.P0_EMERGENCY: 0.15,
ComputationPriority.P1_INTERACTIVE: 0.10,
ComputationPriority.P2_REALTIME: 0.10,
ComputationPriority.P3_NEAR_REALTIME: 0.15,
ComputationPriority.P4_STANDARD: 0.15,
ComputationPriority.P5_BACKGROUND: 0.15,
ComputationPriority.P6_BATCH: 0.20,
}
else:
return self._default_quotas()
#Part 4: Strategy Router Implementation
#4.1 Routing Chain Architecture
The strategy router uses the Chain of Responsibility pattern, evaluating each routing strategy in sequence:
Python
class StrategyRouter:
"""Route computation requests through a chain of strategies."""
def __init__(self):
self.strategies: list[RoutingStrategy] = [
LocalCacheStrategy(), # Prefer local cache hits
DedicatedPoolStrategy(), # Dedicated compute pools
SharedPoolStrategy(), # Shared compute pools
RemoteNodeStrategy(), # Remote nodes
DegradedModeStrategy(), # Degraded mode
]
async def route(
self,
request: ComputationRequest,
priority: ComputationPriority
) -> ComputationResult:
context = RoutingContext(request=request, priority=priority)
for strategy in self.strategies:
if strategy.can_handle(context):
try:
return await strategy.execute(context)
except StrategyFailure as e:
context.add_failure(strategy.name, e)
continue # Try next strategy
raise AllStrategiesExhausted(
f"No strategy could handle request {request.id}",
failures=context.failures,
)
#4.2 Dedicated Pool Strategy
For high-priority requests, use reserved dedicated compute pools:
Python
class DedicatedPoolStrategy(RoutingStrategy):
"""Route to dedicated compute pools for high-priority requests."""
def can_handle(self, context: RoutingContext) -> bool:
return context.priority.value <= 2 # P0-P2 use dedicated pools
async def execute(self, context: RoutingContext) -> ComputationResult:
pool = self.pool_registry.get_dedicated_pool(context.priority)
worker = await pool.acquire(
timeout=context.priority_sla.max_wait_for_worker
)
try:
result = await worker.compute(
request=context.request,
deadline=context.priority_sla.max_latency,
)
self.metrics.record_latency(
priority=context.priority,
latency=result.elapsed,
)
return result
finally:
pool.release(worker)
#Part 5: Degradation Strategies
#5.1 Multi-Level Fallback Chain
When the primary path fails, the system selects degradation strategies by priority:
Python
class FallbackChain:
"""Execute fallback strategies in priority order."""
FALLBACK_MAP = {
ComputationPriority.P0_EMERGENCY: [
InMemoryCacheFallback(),
LastKnownGoodFallback(),
ManualOverrideFallback(),
],
ComputationPriority.P1_INTERACTIVE: [
StaleCacheFallback(max_age=timedelta(seconds=30)),
DegradedResultFallback(),
RetryQueueFallback(),
],
ComputationPriority.P2_REALTIME: [
StaleCacheFallback(max_age=timedelta(minutes=1)),
ApproximateResultFallback(),
AsyncNotificationFallback(),
],
# ... lower priorities have more lenient fallback strategies
}
async def execute(
self,
priority: ComputationPriority,
request: ComputationRequest
) -> ComputationResult:
fallbacks = self.FALLBACK_MAP[priority]
for fb in fallbacks:
try:
result = await fb.attempt(request)
result.metadata["degraded"] = True
result.metadata["fallback_strategy"] = fb.name
return result
except FallbackFailure:
continue
raise CriticalFailure(
f"All fallbacks exhausted for P{priority.value} request"
)
#5.2 Cache Degradation
Cache fallback strategy for emergency requests:
Python
class InMemoryCacheFallback(FallbackStrategy):
"""Return cached result when computation fails."""
async def attempt(self, request: ComputationRequest) -> ComputationResult:
cache_key = self._compute_cache_key(request)
cached = self.cache.get(cache_key)
if cached is None:
raise FallbackFailure("No cached result available")
if cached.age > timedelta(minutes=5):
self.alerter.warn(
f"Serving stale cache (age={cached.age}) for {request.id}"
)
return ComputationResult(
value=cached.value,
source="cache",
freshness=cached.timestamp,
confidence=max(0.5, 1.0 - cached.age.total_seconds() / 300),
)
#Part 6: Preemptive Scheduling
#6.1 Preemption Decisions
When a high-priority request arrives but resources are insufficient, the system must decide whether to preempt lower-priority tasks:
Python
class PreemptionManager:
"""Manage preemption of lower-priority tasks."""
def should_preempt(
self,
incoming: ComputationPriority,
running: list[RunningTask],
) -> list[RunningTask]:
"""Determine which running tasks to preempt."""
# Rule 1: P0 can preempt any non-P0 task
# Rule 2: P1 can only preempt P4 and below
# Rule 3: P2 can only preempt P5 and below
# Rule 4: P3 and below cannot preempt
preemption_threshold = {
ComputationPriority.P0_EMERGENCY: 1, # Can preempt P1+
ComputationPriority.P1_INTERACTIVE: 4, # Can preempt P4+
ComputationPriority.P2_REALTIME: 5, # Can preempt P5+
}
threshold = preemption_threshold.get(incoming)
if threshold is None:
return []
candidates = [
task for task in running
if task.priority.value >= threshold
]
# Prefer preempting longest-running low-priority tasks
candidates.sort(
key=lambda t: (-t.priority.value, -t.elapsed.total_seconds())
)
return candidates[:1] # Preempt at most one at a time
async def preempt(self, task: RunningTask) -> None:
"""Preempt a running task, saving its checkpoint."""
checkpoint = await task.save_checkpoint()
await self.checkpoint_store.save(task.id, checkpoint)
await task.cancel(reason="preempted")
# Re-enqueue the preempted task
await self.requeue(task, checkpoint)
#6.2 Checkpointing and Recovery
Preempted tasks resume via a checkpointing mechanism:
Python
class CheckpointManager:
"""Save and restore computation checkpoints."""
async def save_checkpoint(self, task: RunningTask) -> Checkpoint:
return Checkpoint(
task_id=task.id,
priority=task.priority,
progress=task.progress_percentage,
state=await task.serialize_state(),
saved_at=datetime.utcnow(),
preempted_by=task.preempted_by,
)
async def restore_and_resume(self, checkpoint: Checkpoint) -> None:
task = await self.task_factory.create_from_checkpoint(checkpoint)
# Boost priority by 0.5 after preemption (avoid repeated preemption)
adjusted_priority = max(
0, checkpoint.priority.value - 0.5
)
task.priority = ComputationPriority(round(adjusted_priority))
await self.scheduler.enqueue(task)
#Part 7: Monitoring and Observability
#7.1 Key Metrics
Python
class RoutingMetrics:
"""Metrics for the strategy routing system."""
def __init__(self):
self.request_count = Counter(
"computation_requests_total",
"Total computation requests",
["priority", "strategy", "status"],
)
self.latency = Histogram(
"computation_latency_seconds",
"Computation latency",
["priority"],
buckets=[0.01, 0.05, 0.1, 0.5, 1, 5, 30, 300],
)
self.preemption_count = Counter(
"computation_preemptions_total",
"Total preemptions",
["preemptor_priority", "victim_priority"],
)
self.fallback_count = Counter(
"computation_fallbacks_total",
"Total fallback activations",
["priority", "fallback_strategy"],
)
self.queue_depth = Gauge(
"computation_queue_depth",
"Current queue depth",
["priority"],
)
#7.2 Alert Rules
YAML
alerts:
- name: P0LatencyExceeded
condition: computation_latency_seconds{priority="P0"} > 0.05
severity: critical
action: page_oncall
- name: PreemptionStorm
condition: rate(computation_preemptions_total[5m]) > 10
severity: warning
action: scale_up_pool
- name: FallbackActivated
condition: increase(computation_fallbacks_total{priority="P0"}[1m]) > 0
severity: critical
action: page_oncall_and_incident
- name: QueueBacklog
condition: computation_queue_depth{priority="P4"} > 1000
severity: warning
action: notify_channel
#Part 8: Integration with coomia-dip Architecture
#8.1 Cross-Layer Routing
ComputationCoordinator is a core component of Reasoning & Decision Layer (Reasoning & Decision), communicating with other Layers exclusively via gRPC:
PROTOBUF
service ComputationCoordinator {
rpc SubmitComputation(ComputationRequest) returns (ComputationResponse);
rpc GetComputationStatus(StatusRequest) returns (StatusResponse);
rpc CancelComputation(CancelRequest) returns (CancelResponse);
rpc StreamResults(StreamRequest) returns (stream ComputationResult);
}
message ComputationRequest {
string request_id = 1;
string tenant_id = 2;
string computation_type = 3;
bytes payload = 4;
map<string, string> context = 5;
int32 priority_hint = 6; // Caller's priority suggestion
}
#8.2 Ontology Layer Integration
Routing strategies can be dynamically configured through the Ontology model:
Python
# Define routing rules through Ontology
routing_rule = platform.objects.RoutingRule.create(
name="high_value_customer_boost",
condition="request.context.customer_tier == 'enterprise'",
priority_adjustment=-1, # Boost by one level
effective_from=datetime(2026, 1, 1),
effective_to=datetime(2026, 12, 31),
)
#Part 9: Real-World Case Study — Financial Risk Control
#9.1 Scenario Description
A bank's coomia-dip deployment must simultaneously handle:
- Real-time transaction risk (P0): Every transaction must be risk-assessed within 50ms
- Customer profile updates (P3): Near real-time behavioral profile updates
- Anti-money laundering analysis (P5): Background complex graph analytics
- Monthly compliance reports (P6): End-of-month batch generation
#9.2 System Behavior
Code
09:30 Market opens: transaction flood arrives
→ P6 batch tasks paused
→ P5 background tasks throttled
→ P0/P1 receive 60% of resources
11:00 Trading stabilizes
→ Dynamic quotas revert to normal allocation
→ P5 resumes normal speed
→ P6 starts processing backlog
14:55 Pre-close volatility
→ P0 latency alert fires
→ Two P3 reasoning tasks preempted
→ P0 latency returns to normal after resource release
22:00 Night maintenance window
→ P6 batch tasks run at full speed
→ P5 index rebuild kicks off
#Part 10: Summary
#Key Takeaways
- The 7 priority levels are not arbitrary — each corresponds to a real business scenario class with explicit SLAs and degradation strategies.
- Dynamic classification beats static configuration — request priority is determined jointly by tenant SLA, operational context, system load, and time windows.
- Preemptive scheduling is key to P0 SLA guarantees — without preemption, high priority is just a label with no real enforcement.
- Fallback chains are friendlier than hard failures — even under extreme load, users see degraded results rather than error pages.
- Observability determines whether the pattern succeeds or fails — without comprehensive monitoring and alerting, priority scheduling is a black box.
#References
- Linux CFS Scheduler↗
- Palantir Foundry Computation Service↗
- coomia-dip Architecture Overview
- Gamma et al., Design Patterns, Addison-Wesley, 1994
- Google Borg: Large-Scale Cluster Management↗
tags: strategy-pattern routing priority-scheduling preemption computation coomia-dip
Next article: S10-03 Event Sourcing