gRPC Flow Control — Backpressure, Window Sizing, and Stream Management

package main

import (
    "google.golang.org/grpc"
)

// Server-side flow control
func startServer() {
    s := grpc.NewServer(
        grpc.InitialWindowSize(65535),       // 64KB stream window
        grpc.InitialConnWindowSize(1048576), // 1MB connection window
        
        // For streaming services with large messages
        // grpc.InitialWindowSize(524288),     // 512KB stream window
        // grpc.InitialConnWindowSize(4194304), // 4MB connection window
        
        // For real-time services with small messages
        // grpc.InitialWindowSize(16384),      // 16KB stream window
        // grpc.InitialConnWindowSize(262144),  // 256KB connection window
    )
    
    pb.RegisterThreatServiceServer(s, &server{})
}

// Client-side flow control
func createClient() {
    conn, _ := grpc.Dial("localhost:50051",
        grpc.WithInsecure(),
        grpc.WithInitialWindowSize(65535),
        grpc.WithInitialConnWindowSize(1048576),
    )
}

import grpc
import asyncio

class BackpressureClient:
    """Client that respects flow control backpressure."""
    
    def __init__(self, channel):
        self.stub = pb.ThreatServiceStub(channel)
    
    async def stream_threats(self, callback):
        """Stream threats with backpressure awareness."""
        call = self.stub.StreamThreats(
            pb.StreamRequest(device_id="dev-001"),
        )
        
        async for threat in call:
            # Process threat
            await callback(threat)
            
            # The async for loop naturally paces reads
            # based on flow control window updates
            # If callback is slow, gRPC stops reading
            # from the network, applying backpressure
    
    async def upload_reports_with_backpressure(self, reports):
        """Client-streaming with manual flow control."""
        async def generate():
            for report in reports:
                yield report
                # Simulate slow upload by yielding control
                await asyncio.sleep(0.01)
        
        # Server's flow control automatically paces
        # how fast the generator runs
        response = await self.stub.UploadReports(generate())
        return response
    
    async def monitor_flow_control(self, call):
        """Monitor effective flow control."""
        while True:
            # Check if we're being flow-controlled
            if call.done():
                break
            
            # Log pending bytes in the flow control window
            # (platform-dependent, shown for illustration)
            pending = call.get_pending_data_amount()
            if pending > 100000:
                print(f"Backpressure: {pending} bytes queued")
            
            await asyncio.sleep(1)

# Slow consumer simulation
async def slow_callback(threat):
    print(f"Processing threat: {threat.name}")
    await asyncio.sleep(0.5)  # Slow processing

async def main():
    channel = grpc.aio.insecure_channel("localhost:50051")
    client = BackpressureClient(channel)
    await client.stream_threats(slow_callback)

const grpc = require('@grpc/grpc-js');

// Create a client with custom flow control
const client = new ThreatServiceClient(
  'localhost:50051',
  grpc.credentials.createInsecure(),
  {
    'grpc.initial_reconnect_backoff_ms': 100,
    'grpc.max_reconnect_backoff_ms': 5000,
    // These control HTTP/2 flow control
    'grpc.http2.min_time_between_pings_ms': 10000,
    'grpc.http2.max_pings_without_data': 0,
  },
);

// Backpressure-aware streaming
function streamWithBackpressure() {
  const call = client.StreamThreats({ deviceId: 'dev-001' });
  
  let isPaused = false;
  const buffer = [];
  
  call.on('data', (threat) => {
    if (buffer.length > 10) {
      // Apply backpressure: pause reading
      call.pause();
      isPaused = true;
      console.log('Backpressure applied, buffer full');
    }
    
    buffer.push(threat);
    processNext();
  });
  
  function processNext() {
    if (buffer.length === 0) {
      if (isPaused) {
        call.resume();
        isPaused = false;
        console.log('Resumed reading');
      }
      return;
    }
    
    const threat = buffer.shift();
    processThreat(threat, () => {
      setImmediate(processNext);
    });
  }
  
  function processThreat(threat, callback) {
    // Simulate async processing
    setTimeout(() => {
      console.log(`Processed: ${threat.name}`);
      callback();
    }, 100);
  }
  
  call.on('end', () => console.log('Stream ended'));
  call.on('error', (err) => console.error('Stream error:', err));
}