Backpressure and Flow Control
Backpressure is a critical concept in stream processing that prevents fast producers from overwhelming slow consumers. Flokkit Flow provides several mechanisms to handle backpressure effectively.
Understanding Backpressure
The Problem
Without backpressure:
# Fast producer + slow consumer = memory explosion! 💥
fast_source → [queue grows infinitely] → slow_processor
With backpressure:
# Producer slows down to match consumer ✅
fast_source → [bounded queue] → slow_processor
Queue Configuration
Setting Queue Size
from flow import flow, QueueFullStrategy
builder = flow("Backpressure Example")
# Default queue size (10)
builder.source(fast_producer).to(slow_consumer)
# Custom queue size
builder.source(fast_producer).to(
slow_consumer,
queue_size=100 # Larger buffer
)
# Very small queue for tight coupling
builder.source(producer).to(
consumer,
queue_size=1 # Minimal buffering
)
Queue Full Strategies
Flokkit Flow provides four strategies when queues are full:
1. BLOCK (Default)
Producer waits until space is available:
(builder
.source(fast_producer)
.to(slow_consumer,
queue_size=10,
full_strategy=QueueFullStrategy.BLOCK))
Use when: You want natural flow control and can afford to slow down producers.
2. DROP_NEW
New items are discarded when queue is full:
(builder
.source(sensor_stream) # OK to lose some readings
.to(process_reading,
queue_size=100,
full_strategy=QueueFullStrategy.DROP_NEW))
Use when: Latest data is less important than keeping up (e.g., sensor readings, metrics).
3. DROP_OLD
Oldest items are removed to make room:
(builder
.source(stock_prices) # Want most recent prices
.to(calculate_average,
queue_size=50,
full_strategy=QueueFullStrategy.DROP_OLD))
Use when: Recent data is more valuable than old data.
4. ERROR
Raises an exception when queue is full:
(builder
.source(critical_events) # Must not lose any events
.to(event_processor,
queue_size=1000,
full_strategy=QueueFullStrategy.ERROR))
Use when: Data loss is unacceptable and you need to handle overload explicitly.
Real-World Examples
Rate-Limited API Calls
import asyncio
from datetime import datetime
class RateLimiter:
def __init__(self, calls_per_second: float):
self.calls_per_second = calls_per_second
self.interval = 1.0 / calls_per_second
self.last_call = 0
async def acquire(self):
now = asyncio.get_event_loop().time()
time_passed = now - self.last_call
if time_passed < self.interval:
await asyncio.sleep(self.interval - time_passed)
self.last_call = asyncio.get_event_loop().time()
async def rate_limited_pipeline():
builder = flow("API Pipeline")
rate_limiter = RateLimiter(10) # 10 calls/second
async def fetch_from_api(url: str) -> dict:
await rate_limiter.acquire()
# Make API call
return {"url": url, "data": "..."}
(builder
.source(generate_urls)
.to(fetch_from_api, queue_size=5) # Small queue
.to(process_response))
await builder.run()
Batch Processing with Backpressure
class BatchAccumulator:
def __init__(self, batch_size: int, timeout: float):
self.batch_size = batch_size
self.timeout = timeout
self.batch = []
self.last_emit = asyncio.get_event_loop().time()
async def accumulate(self, item: Any) -> Optional[List[Any]]:
self.batch.append(item)
now = asyncio.get_event_loop().time()
# Emit if batch is full or timeout reached
if (len(self.batch) >= self.batch_size or
now - self.last_emit >= self.timeout):
result = self.batch[:]
self.batch = []
self.last_emit = now
return result
return None
async def batch_pipeline():
builder = flow("Batch Processing")
accumulator = BatchAccumulator(batch_size=100, timeout=5.0)
async def write_batch(batch: List[dict]) -> None:
if batch:
# Slow batch write
await database.write_many(batch)
(builder
.source(stream_source)
.to(accumulator.accumulate)
.filter(lambda x: x is not None) # Only emit full batches
.to(write_batch, queue_size=5)) # Limited batch queue
Multi-Speed Pipeline
async def multi_speed_pipeline():
builder = flow("Multi-Speed")
# Fast source
def fast_source():
for i in range(1000):
yield i
# No delay - produces as fast as possible
# Slow processor
async def slow_processor(x: int) -> int:
await asyncio.sleep(0.1) # 10 items/second
return x * 2
# Fast sink
def fast_sink(x: int) -> None:
print(f"Result: {x}")
(builder
.source(fast_source)
.to(slow_processor,
queue_size=20, # Buffer up to 20 items
full_strategy=QueueFullStrategy.BLOCK) # Natural backpressure
.to(fast_sink))
# The fast source will be throttled by the slow processor
await builder.run()
Monitoring Queue Health
Queue Metrics
class QueueMonitor:
def __init__(self):
self.queues = {}
def register_queue(self, name: str, queue: asyncio.Queue):
self.queues[name] = queue
async def monitor(self):
while True:
for name, queue in self.queues.items():
size = queue.qsize()
capacity = queue.maxsize
usage = size / capacity if capacity > 0 else 0
print(f"Queue {name}: {size}/{capacity} ({usage:.1%})")
if usage > 0.8:
logger.warning(f"Queue {name} is {usage:.1%} full!")
await asyncio.sleep(5)
# Use with tap to monitor
def monitor_queue_size(item: Any) -> None:
# Log queue metrics
metrics.gauge("queue.size", queue.qsize())
metrics.gauge("queue.usage", queue.qsize() / queue.maxsize)
Adaptive Queue Sizing
class AdaptiveQueue:
def __init__(self, initial_size: int = 10):
self.size = initial_size
self.high_water_mark = 0
self.resize_interval = 60 # seconds
self.last_resize = time.time()
def should_resize(self, current_size: int) -> Optional[int]:
self.high_water_mark = max(self.high_water_mark, current_size)
now = time.time()
if now - self.last_resize < self.resize_interval:
return None
self.last_resize = now
# Resize based on usage patterns
usage = self.high_water_mark / self.size
if usage > 0.9:
# Increase size
return int(self.size * 1.5)
elif usage < 0.3 and self.size > 10:
# Decrease size
return int(self.size * 0.7)
self.high_water_mark = 0
return None
Advanced Patterns
Backpressure with Multiple Consumers
async def load_balanced_pipeline():
builder = flow("Load Balanced")
source = builder.source(task_generator)
# Create multiple parallel workers
num_workers = 4
workers = []
for i in range(num_workers):
worker = source.to(
process_task,
name=f"worker_{i}",
queue_size=10
)
workers.append(worker)
# Merge results
merged = workers[0]
for worker in workers[1:]:
merged = merged.merge_with(worker)
merged.to(collect_results)
await builder.run()
Dynamic Backpressure
class DynamicBackpressure:
def __init__(self):
self.processing_times = []
self.window_size = 100
async def adaptive_processor(self, item: Any) -> Any:
start = time.time()
# Adjust processing based on queue pressure
queue_size = self.get_input_queue_size()
if queue_size > 50:
# Speed up processing
result = await self.fast_process(item)
else:
# Can afford slower, better processing
result = await self.quality_process(item)
# Track processing time
elapsed = time.time() - start
self.processing_times.append(elapsed)
if len(self.processing_times) > self.window_size:
self.processing_times.pop(0)
return result
Circuit Breaker Pattern
from enum import Enum
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject requests
HALF_OPEN = "half_open" # Testing recovery
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, timeout: float = 60):
self.failure_threshold = failure_threshold
self.timeout = timeout
self.failures = 0
self.last_failure = None
self.state = CircuitState.CLOSED
async def process(self, item: Any) -> Optional[Any]:
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure > self.timeout:
self.state = CircuitState.HALF_OPEN
else:
return None # Reject while open
try:
result = await self.risky_operation(item)
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.CLOSED
self.failures = 0
return result
except Exception as e:
self.failures += 1
self.last_failure = time.time()
if self.failures >= self.failure_threshold:
self.state = CircuitState.OPEN
logger.error(f"Circuit breaker opened: {e}")
return None
Best Practices
1. Start with Small Queues
# Start small and increase if needed
queue_size=10 # Default is usually fine
2. Monitor Queue Metrics
# Add monitoring to detect bottlenecks
.tap(lambda x: metrics.gauge("queue.size", queue.qsize()))
3. Use Appropriate Strategies
BLOCK: Default choice for most pipelines
DROP_NEW: Sampling, metrics, non-critical data
DROP_OLD: Real-time data, latest values matter
ERROR: Critical data, must handle explicitly
4. Consider Batch Processing
# Batch small items to reduce overhead
.to(batch_accumulator)
.to(bulk_processor)
5. Plan for Bursts
# Size queues for expected bursts
queue_size=peak_burst_size * safety_factor
Debugging Backpressure Issues
Symptoms of Problems
Memory Growth: Queues growing unbounded
High Latency: Items spending too long in queues
Data Loss: Items being dropped unexpectedly
Deadlocks: Pipeline stops processing
Diagnostic Tools
async def diagnose_pipeline():
# Log queue sizes
for name, node in graph.nodes.items():
for port in node._input_ports.values():
size = port.queue.qsize()
logger.info(f"{name}: {size} items queued")
# Measure throughput
start_count = processed_items
await asyncio.sleep(10)
rate = (processed_items - start_count) / 10
logger.info(f"Processing rate: {rate} items/sec")
# Find bottlenecks
slowest_node = max(
nodes,
key=lambda n: n.average_processing_time
)
logger.info(f"Bottleneck: {slowest_node.name}")
Next Steps
Learn about Lifecycle Management
Explore Async Programming patterns
See Common Patterns for more examples