Backpressure

What is Backpressure

Backpressure is a feedback mechanism when producer is faster than consumer, signaling to slow down.

The Problem

Without Backpressure:

Producer: 10,000 msgs/sec
Consumer: 1,000 msgs/sec
Queue: Fills up → OOM → Crash!

Producer: ──────→ [Queue ████████████] → Consumer: ─→
                   (overflow!)

With Backpressure

With Backpressure:

Producer: ────(slow down!)────→ Consumer: ─→
                    ↓
              Queue: manageable size
              System: stable

Backpressure Signals

1. Queue Depth:
   - Queue length exceeds threshold
   - Signal producer to slow down

2. Latency Increase:
   - Processing time increasing
   - Signal to reduce rate

3. Resource Limits:
   - Memory/CPU near limit
   - Signal to shed load

4. Consumer Lag:
   - Consumer falling behind
   - Signal to reduce production rate

When Backpressure is Needed

Scenario	Need Backpressure?
Producer faster than consumer	Yes
Burst traffic	Yes
Resource constrained	Yes
Equal rates	No
Consumer faster	No

Flow Control

Flow Control Mechanisms

1. Pull-based:
   Consumer pulls when ready
   Natural backpressure
   Example: Kafka

2. Push-based with Credit:
   Producer needs credit to send
   Credit consumed on send
   Example: AMQP

3. Rate Limiting:
   Limit producer rate
   Based on consumer capacity
   Example: Token bucket

Pull-Based Flow Control

// Kafka consumer-controlled
while (true) {
    // Consumer controls when to poll
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    // Process at consumer's pace
    for (ConsumerRecord<String, String> record : records) {
        process(record.value());
    }
}
// Natural backpressure - consumer pulls when ready

Credit-Based Flow Control

class CreditBasedFlowControl:
    def __init__(self, initial_credits=100):
        self.credits = initial_credits
        self.lock = threading.Lock()
    
    def request_credits(self, num_messages):
        """Consumer grants credits to producer"""
        with self.lock:
            self.credits += num_messages
    
    def can_send(self):
        """Check if producer can send"""
        with self.lock:
            return self.credits > 0
    
    def send(self, message):
        """Send message, consume credit"""
        with self.lock:
            if self.credits <= 0:
                raise BackpressureError('No credits')
            self.credits -= 1
        
        # Actually send
        self.producer.send(message)

Rate Limiting

class RateLimiter:
    def __init__(self, rate):
        self.rate = rate  # messages per second
        self.tokens = rate
        self.last_refill = time.time()
    
    def allow(self):
        """Check if message is allowed"""
        self._refill()
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False
    
    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        self.tokens = min(self.rate, self.tokens + elapsed * self.rate)
        self.last_refill = now

Flow Control Patterns

Pattern	Description	Latency	Throughput
Pull-based	Consumer pulls	Higher	Controlled
Credit-based	Credit system	Lower	Variable
Rate limiting	Fixed rate	Constant	Capped
Adaptive	Adjusts based on load	Variable	Optimized

Buffering Strategies

Buffer Types

1. In-Memory Buffer:
   Fast, limited size
   Volatile (lost on crash)
   Example: Ring buffer

2. Persistent Buffer:
   Durable, slower
   Survives crash
   Example: Disk-backed queue

3. Hybrid Buffer:
   Memory + disk overflow
   Best of both
   Example: Memory-mapped files

Ring Buffer

class RingBuffer:
    def __init__(self, capacity):
        self.capacity = capacity
        self.buffer = [None] * capacity
        self.head = 0
        self.tail = 0
        self.count = 0
    
    def put(self, item):
        """Add item, overwrite oldest if full"""
        self.buffer[self.head] = item
        self.head = (self.head + 1) % self.capacity
        
        if self.count < self.capacity:
            self.count += 1
        else:
            self.tail = (self.tail + 1) % self.capacity  # Overwrite
    
    def get(self):
        """Get oldest item"""
        if self.count == 0:
            return None
        
        item = self.buffer[self.tail]
        self.tail = (self.tail + 1) % self.capacity
        self.count -= 1
        return item

Buffer Sizing

Buffer Size = Rate × Latency

Example:
- Producer rate: 1000 msgs/sec
- Consumer latency: 100ms
- Buffer needed: 1000 × 0.1 = 100 messages

For burst handling:
- Burst size: 5000 messages
- Buffer needed: 5000 messages

Buffer Monitoring

def monitor_buffer(buffer):
    metrics = {
        'utilization': buffer.count / buffer.capacity,
        'fill_rate': get_fill_rate(),
        'drain_rate': get_drain_rate(),
        'overflow_count': get_overflow_count()
    }
    
    # Alert on high utilization
    if metrics['utilization'] > 0.8:
        alert('Buffer utilization high')
    
    # Alert on overflow
    if metrics['overflow_count'] > 0:
        alert('Buffer overflow detected')

Best Practices

1. Size buffers appropriately for expected load
2. Monitor buffer utilization and overflow
3. Implement overflow handling (drop, block, spill to disk)
4. Use persistent buffers for critical data
5. Test under peak load to validate sizing