# 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:
```python
# Fast producer + slow consumer = memory explosion! 💥
fast_source → [queue grows infinitely] → slow_processor
```

With backpressure:
```python
# Producer slows down to match consumer ✅
fast_source → [bounded queue] → slow_processor
```

## Queue Configuration

### Setting Queue Size

```python
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:

```python
(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:

```python
(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:

```python
(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:

```python
(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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
# Start small and increase if needed
queue_size=10  # Default is usually fine
```

### 2. Monitor Queue Metrics

```python
# 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

```python
# Batch small items to reduce overhead
.to(batch_accumulator)
.to(bulk_processor)
```

### 5. Plan for Bursts

```python
# Size queues for expected bursts
queue_size=peak_burst_size * safety_factor
```

## Debugging Backpressure Issues

### Symptoms of Problems

1. **Memory Growth**: Queues growing unbounded
2. **High Latency**: Items spending too long in queues  
3. **Data Loss**: Items being dropped unexpectedly
4. **Deadlocks**: Pipeline stops processing

### Diagnostic Tools

```python
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](lifecycle.md)
- Explore [Async Programming](async-programming.md) patterns
- See [Common Patterns](patterns.md) for more examples