RTO Calculation: The Complete RFC 6298 Algorithm

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

class TCPRTO:
    """TCP RTO Calculator following RFC 6298."""
    
    # Constants from RFC 6298
    ALPHA = 1/8           # SRTT smoothing factor
    BETA = 1/4            # RTTVAR smoothing factor
    K = 4                 # Variance multiplier
    MIN_RTO = 1000        # Minimum RTO in milliseconds (1 second)
    INITIAL_RTO = 1000    # Initial RTO before any measurements
    MAX_RTO = 60000       # Maximum RTO in milliseconds (60 seconds)
    CLOCK_GRANULARITY = 1 # G: Assume 1ms granularity on modern systems
    
    def __init__(self):
        # State: None indicates no measurements yet
        self.srtt = None
        self.rttvar = None
        self.rto = self.INITIAL_RTO  # Start with 1 second
        self._first_measurement = True
    
    def on_first_measurement(self, R: float):
        """
        Handle the first RTT measurement.
        
        Per RFC 6298 Section 2.2:
        - SRTT <- R
        - RTTVAR <- R/2
        - RTO <- SRTT + max(G, K*RTTVAR)
        """
        self.srtt = R
        self.rttvar = R / 2
        
        # Calculate RTO with granularity consideration
        variance_term = max(self.CLOCK_GRANULARITY, self.K * self.rttvar)
        self.rto = self.srtt + variance_term
        
        # Apply minimum bound
        self.rto = max(self.rto, self.MIN_RTO)
        
        self._first_measurement = False
        
        print(f"First measurement: R={R}ms")
        print(f"  SRTT={self.srtt}ms, RTTVAR={self.rttvar}ms")
        print(f"  RTO={self.rto}ms (= {R} + 4×{R/2} = 3×{R}ms)")
 
 
# Example: Connection to a server with 100ms RTT
rto_calc = TCPRTO()
print(f"Initial RTO (no measurements): {rto_calc.rto}ms")
 
# First RTT measurement from SYN-ACK
rto_calc.on_first_measurement(100)
# Output: RTO = 300ms (3 × first measurement)

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

def on_subsequent_measurement(self, R_prime: float):
    """
    Handle subsequent RTT measurements.
    
    Per RFC 6298 Section 2.3, the update order matters:
    1. Update RTTVAR using OLD SRTT
    2. Update SRTT
    3. Recompute RTO
    """
    # Step 1: Compute error using OLD SRTT
    err = R_prime - self.srtt
    
    # Step 2: Update RTTVAR (using OLD SRTT for error calculation)
    # RTTVAR = (1 - β) * RTTVAR + β * |err|
    # With β = 1/4: RTTVAR = 3/4 * RTTVAR + 1/4 * |err|
    abs_err = abs(err)
    self.rttvar = (1 - self.BETA) * self.rttvar + self.BETA * abs_err
    
    # Alternative formulation:
    # self.rttvar = self.rttvar + self.BETA * (abs_err - self.rttvar)
    
    # Step 3: Update SRTT
    # SRTT = (1 - α) * SRTT + α * R'
    # With α = 1/8: SRTT = 7/8 * SRTT + 1/8 * R'
    self.srtt = (1 - self.ALPHA) * self.srtt + self.ALPHA * R_prime
    
    # Alternative formulation:
    # self.srtt = self.srtt + self.ALPHA * err  # Uses err from step 1
    
    # Step 4: Compute RTO
    variance_term = max(self.CLOCK_GRANULARITY, self.K * self.rttvar)
    self.rto = self.srtt + variance_term
    
    # Step 5: Apply bounds
    self.rto = max(self.rto, self.MIN_RTO)
    self.rto = min(self.rto, self.MAX_RTO)
    
    return self.rto
 
 
# Example trace:
rto_calc = TCPRTO()
rto_calc.on_first_measurement(100)  # First: RTO = 300ms
 
measurements = [105, 95, 102, 98, 100]  # Stable network
for r in measurements:
    rto = rto_calc.on_subsequent_measurement(r)
    print(f"R={r}ms -> SRTT={rto_calc.srtt:.1f}, RTTVAR={rto_calc.rttvar:.1f}, RTO={rto:.1f}")
    
# Output shows RTTVAR decreasing as network proves stable

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

/* RFC 6298 RTO calculation with granularity */
 
/* Clock granularity in same units as SRTT/RTTVAR (e.g., milliseconds) */
#define CLOCK_G  1  /* 1ms granularity on modern systems */
 
/* Calculate RTO from SRTT and RTTVAR */
unsigned int calculate_rto(unsigned int srtt, unsigned int rttvar) {
    unsigned int variance_term;
    unsigned int rto;
    
    /* RFC 6298: RTO = SRTT + max(G, K*RTTVAR) */
    variance_term = 4 * rttvar;  /* K = 4 */
    
    if (variance_term < CLOCK_G) {
        variance_term = CLOCK_G;  /* max(G, K*RTTVAR) */
    }
    
    rto = srtt + variance_term;
    
    /* Apply minimum bound from RFC 6298 */
    if (rto < MIN_RTO) {
        rto = MIN_RTO;
    }
    
    /* Apply maximum bound (implementation-specific) */
    if (rto > MAX_RTO) {
        rto = MAX_RTO;
    }
    
    return rto;
}
 
/* 
 * Note: In practice, the MIN_RTO bound (typically 1 second)
 * is usually larger than any value that would result from
 * the G term dominating, so G is often irrelevant.
 */

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

class TCPRetransmissionTimer:
    """Manages the retransmission timer per RFC 6298 Section 5."""
    
    def __init__(self):
        self.timer_running = False
        self.timer_expiry = None
        self.rto_calculator = TCPRTO()
        
        # Track unacknowledged data
        self.snd_una = 0      # Oldest unacknowledged byte
        self.snd_nxt = 0      # Next byte to send
    
    def send_data(self, segment):
        """Called when data is transmitted."""
        segment_end = segment.sequence + len(segment.data)
        self.snd_nxt = max(self.snd_nxt, segment_end)
        
        # Rule 5.1: Start timer if not already running
        if not self.timer_running and self.snd_una < self.snd_nxt:
            self._start_timer()
    
    def receive_ack(self, ack_num, segment):
        """Called when ACK is received."""
        
        # Check if ACK acknowledges new data
        if ack_num > self.snd_una:
            # New data acknowledged
            self.snd_una = ack_num
            
            # Update RTO estimate if applicable (Karn's algorithm handled elsewhere)
            if not segment.was_retransmitted:
                sample_rtt = self._get_rtt_for_segment(segment)
                self.rto_calculator.on_measurement(sample_rtt)
            
            # Rule 5.2: All data acknowledged?
            if self.snd_una >= self.snd_nxt:
                self._stop_timer()
            
            # Rule 5.3: Still have outstanding data? Restart timer
            elif self.snd_una < self.snd_nxt:
                self._restart_timer()
    
    def handle_timeout(self):
        """Called when the retransmission timer expires."""
        # This is a genuine timeout (not a spurious one)
        
        # RFC 6298 Section 5.5: Backoff RTO
        self.rto_calculator.backoff()
        
        # RFC 6298 Section 5.4: Retransmit earliest unacknowledged segment
        self._retransmit_segment(self.snd_una)
        
        # RFC 6298 Section 5.6: Restart timer with backed-off RTO
        self._start_timer()
    
    def _start_timer(self):
        """Start the retransmission timer."""
        self.timer_running = True
        rto = self.rto_calculator.get_rto()
        self.timer_expiry = current_time() + rto
        print(f"Timer started: expires in {rto}ms")
    
    def _stop_timer(self):
        """Stop the retransmission timer."""
        self.timer_running = False
        self.timer_expiry = None
        print("Timer stopped: all data acknowledged")
    
    def _restart_timer(self):
        """Restart the timer with current RTO."""
        rto = self.rto_calculator.get_rto()
        self.timer_expiry = current_time() + rto
        print(f"Timer restarted: expires in {rto}ms")

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

"""
Complete RFC 6298 RTO Implementation
 
This implementation includes:
- Jacobson's algorithm for SRTT/RTTVAR estimation
- Karn's algorithm for retransmission handling
- All bounds and constraints from RFC 6298
- Timer management per RFC 6298 Section 5
"""
 
from dataclasses import dataclass
from typing import Optional
import time
 
@dataclass
class RTOState:
    """Complete RTO calculator state."""
    srtt: Optional[float] = None      # Smoothed RTT (ms)
    rttvar: Optional[float] = None    # RTT Variance (ms)
    rto: float = 1000                 # Retransmission timeout (ms)
    
    # Constants (RFC 6298)
    ALPHA: float = 1/8
    BETA: float = 1/4
    K: float = 4
    G: float = 1                      # Clock granularity (ms)
    MIN_RTO: float = 1000             # 1 second
    MAX_RTO: float = 60000            # 60 seconds
    
    def on_first_rtt(self, R: float) -> float:
        """Handle first RTT measurement. Section 2.2."""
        self.srtt = R
        self.rttvar = R / 2
        self._update_rto()
        return self.rto
    
    def on_rtt_measurement(self, R: float) -> float:
        """Handle subsequent RTT measurement. Section 2.3."""
        if self.srtt is None:
            return self.on_first_rtt(R)
        
        # IMPORTANT: Calculate err using OLD SRTT
        err = R - self.srtt
        
        # Update RTTVAR first (uses old SRTT)
        self.rttvar = (1 - self.BETA) * self.rttvar + self.BETA * abs(err)
        
        # Update SRTT
        self.srtt = (1 - self.ALPHA) * self.srtt + self.ALPHA * R
        
        self._update_rto()
        return self.rto
    
    def on_timeout(self) -> float:
        """Handle retransmission timeout. Section 5.5."""
        # Exponential backoff
        self.rto = min(self.rto * 2, self.MAX_RTO)
        return self.rto
    
    def _update_rto(self):
        """Compute RTO from current SRTT and RTTVAR."""
        # RTO = SRTT + max(G, K * RTTVAR)
        variance_term = max(self.G, self.K * self.rttvar)
        self.rto = self.srtt + variance_term
        
        # Apply bounds
        self.rto = max(self.rto, self.MIN_RTO)
        self.rto = min(self.rto, self.MAX_RTO)
 
 
class TCPConnectionRTO:
    """
    Complete TCP connection RTO management.
    Includes segment tracking and Karn's algorithm.
    """
    
    def __init__(self):
        self.state = RTOState()
        
        # Segment tracking: seq_num -> (send_time, was_retransmitted)
        self.pending = {}
        
        # Timer state
        self.timer_expiry: Optional[float] = None
    
    def segment_sent(self, seq_num: int, is_retransmit: bool = False):
        """Record segment transmission."""
        now = time.time() * 1000  # Current time in ms
        
        if seq_num in self.pending:
            # Mark as retransmitted (for Karn's algorithm)
            self.pending[seq_num] = (self.pending[seq_num][0], True)
        else:
            self.pending[seq_num] = (now, is_retransmit)
        
        # Start/restart timer
        if self.timer_expiry is None:
            self.timer_expiry = now + self.state.rto
    
    def ack_received(self, ack_num: int):
        """Process acknowledgment."""
        now = time.time() * 1000
        
        # Find acknowledged segments
        acked_seqs = [s for s in self.pending.keys() if s < ack_num]
        
        for seq in acked_seqs:
            send_time, was_retransmitted = self.pending.pop(seq)
            
            # Karn's Rule 1: Only use clean samples
            if not was_retransmitted:
                sample_rtt = now - send_time
                self.state.on_rtt_measurement(sample_rtt)
        
        # Timer management
        if not self.pending:
            # All acknowledged: stop timer
            self.timer_expiry = None
        else:
            # Restart timer for remaining data
            self.timer_expiry = now + self.state.rto
    
    def check_timeout(self) -> bool:
        """Check if timeout has occurred."""
        if self.timer_expiry is None:
            return False
        
        now = time.time() * 1000
        if now >= self.timer_expiry:
            # Timeout occurred
            # Karn's Rule 2: Back off
            self.state.on_timeout()
            return True
        
        return False
    
    def get_current_rto(self) -> float:
        return self.state.rto
 
 
# === Demonstration ===
if __name__ == "__main__":
    conn = TCPConnectionRTO()
    
    print("=== RFC 6298 RTO Calculation Demo ===
")
    
    # Simulate connection establishment
    print("Connection established, first RTT measurement: 100ms")
    conn.state.on_first_rtt(100)
    print(f"  SRTT={conn.state.srtt}ms, RTTVAR={conn.state.rttvar}ms, RTO={conn.state.rto}ms")
    print(f"  (Initial RTO = 3 × first RTT = 300ms, bounded to min 1000ms)
")
    
    # Simulate some data transfer
    measurements = [95, 105, 98, 102, 100, 97, 103]
    print("Subsequent RTT measurements:")
    for i, rtt in enumerate(measurements, 2):
        conn.state.on_rtt_measurement(rtt)
        print(f"  Sample {i}: R={rtt}ms -> SRTT={conn.state.srtt:.1f}, RTTVAR={conn.state.rttvar:.1f}, RTO={conn.state.rto:.1f}")
    
    print("
(Note: RTO stays at minimum 1000ms even though calculated value is lower)")