TCP Timers: The Temporal Guardians of Reliable Communication

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

# Original TCP SRTT calculation (RFC 793 style)
def update_srtt_original(srtt: float, rtt_sample: float, alpha: float = 0.875) -> float:
    """
    Update the Smoothed Round-Trip Time using EWMA.
    
    Args:
        srtt: Current smoothed RTT estimate (in ms)
        rtt_sample: New RTT measurement (in ms)
        alpha: Smoothing factor (typically 0.875 = 7/8)
    
    Returns:
        Updated SRTT estimate
    
    Example:
        >>> srtt = 100  # Initial estimate: 100ms
        >>> srtt = update_srtt_original(srtt, 80)   # New sample: 80ms
        >>> print(f"New SRTT: {srtt:.2f}ms")  # SRTT moves slightly toward 80
        New SRTT: 97.50ms
        
        >>> srtt = update_srtt_original(srtt, 120)  # New sample: 120ms  
        >>> print(f"New SRTT: {srtt:.2f}ms")  # SRTT moves toward 120
        New SRTT: 100.31ms
    """
    return alpha * srtt + (1 - alpha) * rtt_sample
 
 
# Simulation: How SRTT evolves over time
def simulate_srtt_evolution():
    """Demonstrate SRTT smoothing over varying RTT samples."""
    import random
    
    # Initial SRTT estimate
    srtt = 100.0  # 100ms initial estimate
    
    # Simulate RTT samples with some variation and a spike
    samples = [95, 102, 98, 105, 97, 150, 103, 99, 101, 98]
    
    print("RTT Measurement Smoothing Simulation")
    print("=" * 50)
    print(f"Initial SRTT estimate: {srtt:.2f}ms")
    print(f"Smoothing factor (α): 0.875 (7/8)")
    print()
    
    for i, sample in enumerate(samples):
        old_srtt = srtt
        srtt = update_srtt_original(srtt, sample)
        delta = srtt - old_srtt
        print(f"Sample {i+1}: RTT={sample:3d}ms → SRTT: {old_srtt:.2f} → {srtt:.2f}ms (Δ={delta:+.2f})")
    
    print()
    print("Notice how the spike (150ms) affects SRTT only gradually,")
    print("preventing overreaction to temporary conditions.")
 
 
if __name__ == "__main__":
    simulate_srtt_evolution()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143

class JacobsonRTOEstimator:
    """
    Implements Jacobson's algorithm for RTO estimation (RFC 6298).
    
    This is the algorithm used by modern TCP implementations to calculate
    the retransmission timeout based on RTT measurements.
    """
    
    def __init__(self, clock_granularity: float = 1.0, min_rto: float = 1000.0):
        """
        Initialize the RTO estimator.
        
        Args:
            clock_granularity: Timer granularity in ms (G in RFC 6298)
            min_rto: Minimum RTO value in ms (RFC 6298 recommends 1 second)
        """
        self.srtt: float | None = None      # Smoothed RTT
        self.rttvar: float | None = None    # RTT variance
        self.rto: float = 1000.0            # Current RTO (default 1 second)
        self.clock_granularity = clock_granularity
        self.min_rto = min_rto
        
        # Constants from RFC 6298
        self.alpha = 1/8    # SRTT smoothing factor
        self.beta = 1/4     # RTTVAR smoothing factor
        self.K = 4          # Variance multiplier
    
    def update(self, rtt_sample: float) -> float:
        """
        Update RTO estimate with a new RTT measurement.
        
        Args:
            rtt_sample: A new RTT measurement in milliseconds
            
        Returns:
            The updated RTO value in milliseconds
        """
        if self.srtt is None:
            # First measurement - initialize state
            self.srtt = rtt_sample
            self.rttvar = rtt_sample / 2
        else:
            # Subsequent measurements - apply Jacobson's algorithm
            # Note: RTTVAR must be updated BEFORE SRTT (uses old SRTT value)
            
            # Calculate absolute deviation from current estimate
            deviation = abs(self.srtt - rtt_sample)
            
            # Update RTTVAR: exponentially weighted deviation
            self.rttvar = (1 - self.beta) * self.rttvar + self.beta * deviation
            
            # Update SRTT: exponentially weighted average
            self.srtt = (1 - self.alpha) * self.srtt + self.alpha * rtt_sample
        
        # Calculate new RTO
        # RTO = SRTT + max(G, K * RTTVAR)
        self.rto = self.srtt + max(self.clock_granularity, self.K * self.rttvar)
        
        # Apply minimum RTO (RFC 6298: SHOULD be 1 second)
        self.rto = max(self.rto, self.min_rto)
        
        return self.rto
    
    def get_state(self) -> dict:
        """Return current estimator state for debugging/monitoring."""
        return {
            "srtt": self.srtt,
            "rttvar": self.rttvar,
            "rto": self.rto,
        }
 
 
def demonstrate_jacobson():
    """Demonstrate Jacobson's algorithm with real-world scenarios."""
    
    print("=" * 70)
    print("Jacobson's Algorithm Demonstration")
    print("=" * 70)
    
    # Scenario 1: Stable network
    print("
📊 SCENARIO 1: Stable Low-Latency Network")
    print("-" * 50)
    estimator = JacobsonRTOEstimator(min_rto=200)  # Lower min for demo
    
    stable_samples = [100, 102, 99, 101, 98, 103, 100, 97, 102, 99]
    
    for i, sample in enumerate(stable_samples):
        rto = estimator.update(sample)
        state = estimator.get_state()
        print(f"Sample {i+1:2d}: RTT={sample:3d}ms | "
              f"SRTT={state['srtt']:.1f}ms | "
              f"RTTVAR={state['rttvar']:.1f}ms | "
              f"RTO={rto:.1f}ms")
    
    print(f"
✓ Final RTO for stable network: {estimator.rto:.1f}ms")
    print("  Notice: Low RTTVAR leads to tight RTO close to SRTT")
    
    # Scenario 2: Variable network
    print("
📊 SCENARIO 2: High-Variance Network")
    print("-" * 50)
    estimator = JacobsonRTOEstimator(min_rto=200)
    
    variable_samples = [100, 150, 80, 130, 60, 140, 90, 120, 70, 110]
    
    for i, sample in enumerate(variable_samples):
        rto = estimator.update(sample)
        state = estimator.get_state()
        print(f"Sample {i+1:2d}: RTT={sample:3d}ms | "
              f"SRTT={state['srtt']:.1f}ms | "
              f"RTTVAR={state['rttvar']:.1f}ms | "
              f"RTO={rto:.1f}ms")
    
    print(f"
✓ Final RTO for variable network: {estimator.rto:.1f}ms")
    print("  Notice: High RTTVAR leads to more conservative RTO")
    
    # Scenario 3: Network with sudden congestion
    print("
📊 SCENARIO 3: Sudden Congestion Event")
    print("-" * 50)
    estimator = JacobsonRTOEstimator(min_rto=200)
    
    # Start stable, then congestion spike, then recovery
    congestion_samples = [100, 98, 102, 99, 250, 300, 280, 150, 110, 102]
    
    for i, sample in enumerate(congestion_samples):
        rto = estimator.update(sample)
        state = estimator.get_state()
        marker = " ← congestion!" if 250 <= sample <= 300 else ""
        print(f"Sample {i+1:2d}: RTT={sample:3d}ms | "
              f"SRTT={state['srtt']:.1f}ms | "
              f"RTTVAR={state['rttvar']:.1f}ms | "
              f"RTO={rto:.1f}ms{marker}")
    
    print(f"
✓ Algorithm adapts to congestion and recovers gradually")
 
 
if __name__ == "__main__":
    demonstrate_jacobson()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

def calculate_backoff_schedule(initial_rto: float, max_retries: int = 6, 
                                  max_rto: float = 120000.0) -> list[dict]:
    """
    Calculate the exponential backoff schedule for TCP retransmissions.
    
    Args:
        initial_rto: Initial RTO value in milliseconds
        max_retries: Maximum number of retransmission attempts
        max_rto: Maximum RTO cap in milliseconds (default 120 seconds)
    
    Returns:
        List of dictionaries with retry info
    """
    schedule = []
    current_rto = initial_rto
    cumulative_time = 0
    
    for retry in range(max_retries + 1):
        schedule.append({
            "attempt": retry,
            "rto_ms": current_rto,
            "rto_seconds": current_rto / 1000,
            "cumulative_ms": cumulative_time,
            "cumulative_seconds": cumulative_time / 1000,
        })
        
        cumulative_time += current_rto
        current_rto = min(current_rto * 2, max_rto)  # Double with cap
    
    return schedule
 
 
def demonstrate_backoff():
    """Show TCP exponential backoff in action."""
    
    print("TCP Retransmission Exponential Backoff")
    print("=" * 70)
    print()
    
    # Typical internet RTO scenario
    initial_rto = 1000  # 1 second (RFC 6298 minimum)
    schedule = calculate_backoff_schedule(initial_rto, max_retries=6)
    
    print("Scenario: Typical Internet connection (RTO starts at 1 second)")
    print("-" * 70)
    print(f"{'Attempt':<10} {'RTO':<15} {'Cumulative Wait':<20} {'Status'}")
    print("-" * 70)
    
    for entry in schedule:
        attempt = entry['attempt']
        rto = entry['rto_seconds']
        cumulative = entry['cumulative_seconds']
        
        if attempt == 0:
            status = "Initial transmission"
        elif attempt <= 2:
            status = "Probably recoverable"
        elif attempt <= 4:
            status = "Significant delay"
        else:
            status = "Connection likely dead"
        
        print(f"{attempt:<10} {rto:>10.1f}s    {cumulative:>10.1f}s           {status}")
    
    total_time = schedule[-1]['cumulative_seconds'] + schedule[-1]['rto_seconds']
    print("-" * 70)
    print(f"Total time before giving up: {total_time:.0f} seconds ({total_time/60:.1f} minutes)")
    
    print()
    print("Key Observations:")
    print("• Each retry doubles the timeout (exponential growth)")
    print("• After 6 retries, total wait time exceeds 2 minutes")
    print("• This prevents TCP from hammering a failing path")
    print("• Applications should implement their own timeouts for responsiveness")
 
 
if __name__ == "__main__":
    demonstrate_backoff()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171

from dataclasses import dataclass
from enum import Enum
from typing import Optional
 
class SegmentState(Enum):
    SENT = "sent"
    RETRANSMITTED = "retransmitted"
    ACKNOWLEDGED = "acknowledged"
 
@dataclass
class TrackedSegment:
    """Track state of a sent segment for RTT measurement."""
    sequence_number: int
    first_send_time: float
    retransmit_count: int = 0
    state: SegmentState = SegmentState.SENT
 
 
class TCPRTOManagerWithKarn:
    """
    TCP RTO manager implementing Karn's algorithm.
    
    Demonstrates how RTT measurements are handled with
    proper ambiguity resolution.
    """
    
    def __init__(self):
        self.srtt: Optional[float] = None
        self.rttvar: Optional[float] = None
        self.rto: float = 1000.0  # Initial 1 second
        self.backed_off_rto: float = 1000.0  # Current operational RTO
        
        # Track outstanding segments
        self.segments: dict[int, TrackedSegment] = {}
        
        # Statistics
        self.rtt_samples_used = 0
        self.rtt_samples_discarded = 0
    
    def segment_sent(self, seq: int, send_time: float):
        """Record a segment being sent for the first time."""
        self.segments[seq] = TrackedSegment(
            sequence_number=seq,
            first_send_time=send_time
        )
    
    def segment_retransmitted(self, seq: int, retransmit_time: float):
        """Record a segment being retransmitted (timeout occurred)."""
        if seq in self.segments:
            seg = self.segments[seq]
            seg.retransmit_count += 1
            seg.state = SegmentState.RETRANSMITTED
            
            # Apply exponential backoff to operational RTO
            self.backed_off_rto = min(self.backed_off_rto * 2, 120000)
            
            print(f"⚠️  Segment {seq} retransmitted (attempt #{seg.retransmit_count})")
            print(f"    Backed-off RTO: {self.backed_off_rto:.0f}ms")
    
    def ack_received(self, ack: int, recv_time: float) -> Optional[float]:
        """
        Process acknowledgment. Returns RTT if measurement was valid.
        Implements Karn's algorithm for ambiguity handling.
        """
        # Find the acknowledged segment
        acked_seqs = [s for s in self.segments if s < ack]
        
        if not acked_seqs:
            return None
        
        measured_rtt = None
        
        for seq in acked_seqs:
            seg = self.segments[seq]
            
            if seg.retransmit_count > 0:
                # KARN'S ALGORITHM: Do not use this RTT sample
                self.rtt_samples_discarded += 1
                print(f"📊 Segment {seq}: ACK received but RTT DISCARDED (ambiguous)")
                print(f"   Segment was retransmitted - cannot determine which copy was ACKed")
            else:
                # First transmission was acknowledged - safe to measure
                rtt = recv_time - seg.first_send_time
                measured_rtt = rtt
                self._update_rtt_estimates(rtt)
                self.rtt_samples_used += 1
                
                # KARN's ALGORITHM: Reset backed-off RTO only on clean measurement
                self.backed_off_rto = self.rto
                
                print(f"✓ Segment {seq}: RTT measured = {rtt:.1f}ms (used for estimate)")
            
            del self.segments[seq]
        
        return measured_rtt
    
    def _update_rtt_estimates(self, rtt_sample: float):
        """Update SRTT/RTTVAR using Jacobson's algorithm."""
        if self.srtt is None:
            self.srtt = rtt_sample
            self.rttvar = rtt_sample / 2
        else:
            deviation = abs(self.srtt - rtt_sample)
            self.rttvar = 0.75 * self.rttvar + 0.25 * deviation
            self.srtt = 0.875 * self.srtt + 0.125 * rtt_sample
        
        self.rto = self.srtt + max(1, 4 * self.rttvar)
        self.rto = max(self.rto, 200)  # Minimum 200ms for demo
    
    def get_current_rto(self) -> float:
        """Get the current operational RTO (includes backoff)."""
        return self.backed_off_rto
    
    def print_statistics(self):
        """Print measurement statistics."""
        print(f"
📈 RTT Measurement Statistics:")
        print(f"   Samples used: {self.rtt_samples_used}")
        print(f"   Samples discarded (Karn): {self.rtt_samples_discarded}")
        print(f"   Current SRTT: {self.srtt:.1f}ms" if self.srtt else "   SRTT: Not yet measured")
        print(f"   Current RTO: {self.rto:.1f}ms")
 
 
def demonstrate_karns_algorithm():
    """Demonstrate Karn's algorithm with a realistic scenario."""
    
    print("=" * 70)
    print("Karn's Algorithm Demonstration")
    print("=" * 70)
    print()
    
    manager = TCPRTOManagerWithKarn()
    
    # Scenario: Mix of successful and retransmitted segments
    print("Timeline of events:")
    print("-" * 50)
    
    # Segment 1: Successful first transmission
    print("
T=0ms:    Send Segment 1000")
    manager.segment_sent(1000, 0)
    
    # Segment 2: Will need retransmission
    print("T=50ms:   Send Segment 1500")
    manager.segment_sent(1500, 50)
    
    # ACK for segment 1 arrives
    print("T=120ms:  ACK 1500 received")
    manager.ack_received(1500, 120)
    
    # Segment 2 times out
    print("T=1100ms: Timeout for Segment 1500")
    manager.segment_retransmitted(1500, 1100)
    
    # Segment 3 sent during backoff period
    print("T=1200ms: Send Segment 2000")
    manager.segment_sent(2000, 1200)
    
    # ACK for retransmitted segment 2
    print("T=1400ms: ACK 2000 received")
    manager.ack_received(2000, 1400)
    
    # ACK for segment 3 (clean measurement)
    print("T=1350ms: ACK 2500 received")
    manager.ack_received(2500, 1350)
    
    manager.print_statistics()
 
 
if __name__ == "__main__":
    demonstrate_karns_algorithm()