Understanding the TCP/IP Protocol Suite
The TCP/IP protocol suite is the fundamental communication language of the internet. Every web request, email transmission, file download, and streaming video relies on this protocol stack to move data reliably across networks. While most developers interact with TCP/IP through high-level abstractions, understanding how to implement and work directly with these protocols unlocks powerful capabilities for building robust networked applications.
The Layered Architecture
TCP/IP is organized into four conceptual layers, each responsible for a specific aspect of data transmission:
- Link Layer (Network Interface) — Handles physical transmission of data frames over Ethernet, Wi-Fi, or other media. Deals with MAC addresses and frame encapsulation.
- Internet Layer (IP) — Routes packets across networks using IP addresses. Provides best-effort, connectionless delivery. Fragments and reassembles datagrams as needed.
- Transport Layer (TCP/UDP) — TCP provides reliable, ordered, connection-oriented data streams with flow control and congestion management. UDP offers lightweight, connectionless datagram service.
- Application Layer — Protocols like HTTP, FTP, DNS, and SMTP that use the transport layer to exchange application-specific messages.
Each layer encapsulates the data from the layer above, adding its own header. When a TCP segment travels from source to destination, it's wrapped in an IP datagram, which is wrapped in a link-layer frame. This encapsulation enables modularity — each layer can evolve independently without breaking the others.
Why Implementing TCP/IP Matters
Most developers consume TCP/IP through socket libraries without ever touching the protocol internals. However, building your own implementation — even a simplified one — provides critical benefits:
- Deep debugging skills — When production systems exhibit strange network behavior, understanding the wire-level protocol helps you interpret packet captures and diagnose root causes.
- Performance optimization — Knowing how TCP congestion control and window scaling work lets you tune buffer sizes, adjust Nagle's algorithm behavior, and avoid common throughput pitfalls.
- Custom protocol design — IoT devices, gaming engines, and distributed systems often need lightweight custom transports. Understanding TCP's design tradeoffs informs your own protocol decisions.
- Security awareness — SYN floods, sequence number attacks, and TCP reset vulnerabilities become concrete threats you can reason about, not abstract CVEs.
- Embedded systems development — Resource-constrained devices may need minimal TCP stacks. Knowing the protocol lets you strip away unnecessary features confidently.
Practical Implementation: Socket Programming Fundamentals
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →The most common way to implement TCP/IP communication is through the Berkeley sockets API, available on virtually every operating system. Let's build complete, working examples that demonstrate the full lifecycle of a TCP connection.
TCP Client Implementation
Below is a complete Python implementation of a TCP client that connects to a server, sends a message, and receives a response. This demonstrates the canonical sequence of socket operations: create, connect, send, receive, close.
#!/usr/bin/env python3
"""
tcp_client.py — A complete TCP client implementation demonstrating
the socket lifecycle: socket() → connect() → send() → recv() → close()
"""
import socket
import sys
def run_tcp_client(host: str, port: int, message: str) -> None:
"""
Establish a TCP connection to a remote host, transmit a message,
and print the server's response.
"""
# Step 1: Create a TCP socket
# AF_INET = IPv4 address family
# SOCK_STREAM = TCP transport protocol
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Set a timeout to avoid hanging indefinitely
client_socket.settimeout(10.0)
try:
# Step 2: Initiate the three-way handshake with connect()
# The operating system automatically performs:
# - SYN sent to server
# - SYN-ACK received from server
# - ACK sent to complete handshake
print(f"[*] Connecting to {host}:{port}...")
client_socket.connect((host, port))
print(f"[+] Connection established. Local endpoint: "
f"{client_socket.getsockname()}")
# Step 3: Send application data
# The OS handles segmentation, sequencing, and retransmission
payload = message.encode('utf-8')
bytes_sent = client_socket.send(payload)
print(f"[*] Sent {bytes_sent} bytes: {message}")
# Step 4: Receive response data
# recv() returns up to the buffer size; loop until complete
response_chunks = []
while True:
chunk = client_socket.recv(4096)
if not chunk:
# Connection closed by server (FIN received)
break
response_chunks.append(chunk)
# A simple heuristic: stop if we've received a complete line
if b'\n' in chunk:
break
full_response = b''.join(response_chunks).decode('utf-8')
print(f"[+] Received response: {full_response}")
except socket.timeout:
print("[!] Connection timed out")
except ConnectionRefusedError:
print(f"[!] Connection refused — is the server running on {host}:{port}?")
except Exception as exc:
print(f"[!] Error: {exc}")
finally:
# Step 5: Graceful close — sends FIN, waits for ACK
client_socket.close()
print("[*] Connection closed")
if __name__ == "__main__":
# Usage: python tcp_client.py
HOST = sys.argv[1] if len(sys.argv) > 1 else "localhost"
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 8080
MESSAGE = sys.argv[3] if len(sys.argv) > 3 else "Hello, TCP!"
run_tcp_client(HOST, PORT, MESSAGE)
This client demonstrates several critical concepts. The socket.socket() call with SOCK_STREAM selects TCP. The connect() call triggers the kernel's TCP stack to perform the three-way handshake transparently. The send() and recv() calls operate on the byte stream abstraction — TCP's segmentation, sequencing, and acknowledgment happen invisibly inside the kernel.
TCP Server Implementation
A TCP server must bind to a port, listen for incoming connections, accept each client, and then exchange data. Here's a complete multi-client server using Python's select module for non-blocking I/O:
#!/usr/bin/env python3
"""
tcp_server.py — A production-style TCP echo server with concurrent
client handling using select-based I/O multiplexing.
"""
import socket
import select
import signal
import sys
from dataclasses import dataclass, field
from typing import Dict, List
@dataclass
class ClientState:
"""Tracks per-connection state including buffered data."""
sock: socket.socket
address: tuple
recv_buffer: bytearray = field(default_factory=bytearray)
send_buffer: bytearray = field(default_factory=bytearray)
pending_close: bool = False
class TCPServer:
"""
A non-blocking TCP server that handles multiple concurrent clients
using select() for I/O multiplexing.
"""
def __init__(self, host: str, port: int):
self.host = host
self.port = port
self.clients: Dict[int, ClientState] = {}
# Create the listening socket
self.listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Allow rapid restart by disabling TIME_WAIT
self.listen_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# Set non-blocking mode for select-based multiplexing
self.listen_sock.setblocking(False)
# Bind to all interfaces on the specified port
self.listen_sock.bind((host, port))
# Start listening with a backlog of 128 pending connections
self.listen_sock.listen(128)
print(f"[*] TCP echo server listening on {host}:{port}")
def run(self) -> None:
"""Main event loop using select() for I/O readiness notification."""
running = True
while running:
# Build the file descriptor lists for select()
read_socks = [self.listen_sock] + [c.sock for c in self.clients.values()]
write_socks = [
c.sock for c in self.clients.values()
if len(c.send_buffer) > 0 or c.pending_close
]
# select() blocks until at least one socket is ready
readable, writable, exceptional = select.select(
read_socks, write_socks, read_socks, 1.0
)
# Handle exceptional conditions (OOB data, errors)
for sock in exceptional:
self._handle_disconnect(sock)
# Accept new connections
for sock in readable:
if sock == self.listen_sock:
self._accept_client()
# Read from ready clients
for sock in readable:
if sock != self.listen_sock and sock in [
c.sock for c in self.clients.values()
]:
self._handle_read(sock)
# Write to ready clients
for sock in writable:
self._handle_write(sock)
def _accept_client(self) -> None:
"""Accept a new client connection and register it."""
try:
client_sock, client_addr = self.listen_sock.accept()
client_sock.setblocking(False)
fd = client_sock.fileno()
self.clients[fd] = ClientState(
sock=client_sock, address=client_addr
)
print(f"[+] New connection from {client_addr}")
except (BlockingIOError, OSError):
pass # No pending connections right now
def _handle_read(self, sock: socket.socket) -> None:
"""Read available data from a client socket."""
fd = sock.fileno()
client = self.clients.get(fd)
if not client:
return
try:
data = sock.recv(4096)
if data:
# Echo the data back: queue it in the send buffer
print(f"[*] Received {len(data)} bytes from {client.address}")
client.send_buffer.extend(data)
else:
# Zero-length read means the client sent FIN
self._handle_disconnect(sock)
except (BlockingIOError, OSError):
pass # No data available right now
def _handle_write(self, sock: socket.socket) -> None:
"""Write buffered data to a client socket."""
fd = sock.fileno()
client = self.clients.get(fd)
if not client:
return
try:
if len(client.send_buffer) > 0:
bytes_sent = sock.send(client.send_buffer)
# Remove sent bytes from the buffer
del client.send_buffer[:bytes_sent]
print(f"[*] Sent {bytes_sent} bytes to {client.address}")
if client.pending_close and len(client.send_buffer) == 0:
self._handle_disconnect(sock)
except (BlockingIOError, BrokenPipeError, OSError):
self._handle_disconnect(sock)
def _handle_disconnect(self, sock: socket.socket) -> None:
"""Clean up a disconnected or errored client."""
fd = sock.fileno()
client = self.clients.pop(fd, None)
if client:
print(f"[-] Disconnected: {client.address}")
try:
client.sock.close()
except OSError:
pass
def shutdown(self) -> None:
"""Gracefully shut down the server."""
print("[*] Shutting down server...")
for client in list(self.clients.values()):
try:
client.sock.close()
except OSError:
pass
self.clients.clear()
try:
self.listen_sock.close()
except OSError:
pass
if __name__ == "__main__":
HOST = sys.argv[1] if len(sys.argv) > 1 else "0.0.0.0"
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 8080
server = TCPServer(HOST, PORT)
# Handle graceful shutdown on SIGINT (Ctrl+C)
def sigint_handler(signum, frame):
server.shutdown()
sys.exit(0)
signal.signal(signal.SIGINT, sigint_handler)
try:
server.run()
except KeyboardInterrupt:
server.shutdown()
This server demonstrates several production-grade patterns. The SO_REUSEADDR socket option prevents "Address already in use" errors when restarting the server quickly. Non-blocking sockets combined with select() allow a single thread to handle thousands of concurrent connections efficiently. The send buffer management ensures data is transmitted completely even when the kernel's send buffer is temporarily full.
Understanding the Three-Way Handshake
When the client calls connect(), the operating system's TCP stack performs this sequence automatically:
- Step 1 — SYN: Client sends a TCP segment with the SYN flag set, containing an initial sequence number (ISN). The client enters SYN-SENT state.
- Step 2 — SYN-ACK: Server responds with a segment that has both SYN and ACK flags set. It acknowledges the client's ISN and provides its own ISN. The server enters SYN-RECEIVED state.
- Step 3 — ACK: Client sends a segment with the ACK flag, acknowledging the server's ISN. Both endpoints transition to ESTABLISHED state.
Here's a raw socket implementation in Python (using socket.SOCK_RAW) that crafts a custom SYN packet, illustrating the handshake at the packet level. Note: this requires root privileges and is for educational purposes.
#!/usr/bin/env python3
"""
syn_sender.py — Craft a raw TCP SYN packet to demonstrate the three-way
handshake initiation at the packet level. Requires root privileges.
"""
import socket
import struct
import random
import sys
def compute_checksum(data: bytes) -> int:
"""
Compute the 16-bit one's complement of the one's complement sum
of the ICMP/TCP pseudo-header and payload.
"""
if len(data) % 2 != 0:
data += b'\x00'
total = 0
for i in range(0, len(data), 2):
word = (data[i] << 8) + data[i + 1]
total += word
# Fold 32-bit sum into 16 bits
total = (total >> 16) + (total & 0xFFFF)
total += (total >> 16)
return (~total) & 0xFFFF
def craft_syn_packet(source_ip: str, dest_ip: str, source_port: int, dest_port: int) -> bytes:
"""
Build a complete IP datagram containing a TCP SYN segment.
"""
# --- IP Header (20 bytes, no options) ---
ip_version_ihl = 0x45 # Version 4, IHL 5 (20 bytes)
ip_dscp_ecn = 0x00 # Default DSCP and ECN
ip_total_length = 40 # 20 bytes IP header + 20 bytes TCP header
ip_identification = random.randint(0, 65535)
ip_flags_fragment = 0x4000 # Don't Fragment flag
ip_ttl = 64
ip_protocol = socket.IPPROTO_TCP # 6
ip_checksum = 0x0000 # Kernel fills this in for raw sockets
# Convert IP addresses to 32-bit integers
source_ip_int = struct.unpack("!I", socket.inet_aton(source_ip))[0]
dest_ip_int = struct.unpack("!I", socket.inet_aton(dest_ip))[0]
ip_header = struct.pack(
"!BBHHHBBHII",
ip_version_ihl, ip_dscp_ecn, ip_total_length,
ip_identification, ip_flags_fragment,
ip_ttl, ip_protocol, ip_checksum,
source_ip_int, dest_ip_int
)
# --- TCP Header (20 bytes) ---
source_port = source_port
dest_port = dest_port
sequence_number = random.randint(0, 0xFFFFFFFF)
acknowledgment_number = 0 # Not used in SYN
data_offset_reserved = 0x50 # Data offset 5 (20 bytes), reserved 0
flags = 0x02 # SYN flag only
window_size = 64240 # Typical default
checksum = 0x0000 # Placeholder, computed below
urgent_pointer = 0x0000
# TCP pseudo-header for checksum computation
pseudo_header = struct.pack(
"!IIBBH",
source_ip_int, dest_ip_int,
0x00, socket.IPPROTO_TCP,
20 # TCP segment length (header only, no data)
)
tcp_header_without_checksum = struct.pack(
"!HHIIBBHHH",
source_port, dest_port,
sequence_number, acknowledgment_number,
data_offset_reserved, flags,
window_size, checksum, urgent_pointer
)
# Compute checksum over pseudo-header + TCP header
checksum = compute_checksum(pseudo_header + tcp_header_without_checksum)
# Re-pack TCP header with correct checksum
tcp_header = struct.pack(
"!HHIIBBHHH",
source_port, dest_port,
sequence_number, acknowledgment_number,
data_offset_reserved, flags,
window_size, checksum, urgent_pointer
)
return ip_header + tcp_header
if __name__ == "__main__":
if len(sys.argv) < 4:
print("Usage: sudo python syn_sender.py ")
sys.exit(1)
DEST_IP = sys.argv[1]
DEST_PORT = int(sys.argv[2])
SOURCE_PORT = int(sys.argv[3])
# Get local source IP
SOURCE_IP = socket.gethostbyname(socket.gethostname())
packet = craft_syn_packet(SOURCE_IP, DEST_IP, SOURCE_PORT, DEST_PORT)
# Create a raw socket (requires root)
raw_sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
raw_sock.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
print(f"[*] Sending SYN: {SOURCE_IP}:{SOURCE_PORT} → {DEST_IP}:{DEST_PORT}")
raw_sock.sendto(packet, (DEST_IP, 0))
print("[+] SYN packet sent. Check wireshark/tcpdump for SYN-ACK response.")
This raw socket example peels back the abstraction. You can see exactly how the IP and TCP headers are constructed byte-by-byte. The checksum computation uses the TCP pseudo-header that includes source and destination IPs — this is why TCP is bound to the IP layer. Running this alongside Wireshark gives you a vivid picture of the handshake.
Implementing Core TCP Mechanisms
Beyond basic socket usage, understanding TCP's internal algorithms lets you implement custom transport behaviors. Let's examine three critical mechanisms: sliding window flow control, congestion control, and reliable retransmission.
Sliding Window and Flow Control
TCP uses a sliding window protocol to prevent a fast sender from overwhelming a slow receiver. The receiver advertises a window size in each ACK, indicating how many bytes it can buffer. The sender must never have more unacknowledged bytes in flight than this window.
Here's a simplified implementation of a sliding window sender in Python that respects a receiver's advertised window:
#!/usr/bin/env python3
"""
sliding_window_sender.py — Demonstrates TCP sliding window logic.
The sender tracks the window size advertised by the receiver and
never exceeds it with in-flight data.
"""
from dataclasses import dataclass
from collections import deque
@dataclass
class Segment:
"""Represents an outstanding segment tracked by the sender."""
seq_start: int
seq_end: int
data: bytes
retransmit_count: int = 0
class SlidingWindowSender:
"""
Simulates a TCP sender that respects the receiver's advertised window.
"""
def __init__(self, initial_seq: int = 0):
self.snd_una = initial_seq # Oldest unacknowledged byte
self.snd_nxt = initial_seq # Next byte to send
self.snd_wnd = 65535 # Initial window (will be updated by receiver)
self.snd_wl1 = 0 # Sequence number of last window update
self.snd_wl2 = 0 # Acknowledgment number of last window update
self.outstanding: deque[Segment] = deque()
self.max_segment_size = 1460 # Typical Ethernet MSS
def update_window(self, ack_number: int, advertised_window: int) -> None:
"""
Process an incoming ACK: advance snd_una, update window, remove acked segments.
"""
# Update the window information
self.snd_wnd = advertised_window
# Remove segments that have been fully acknowledged
while self.outstanding and self.outstanding[0].seq_end <= ack_number:
acked_segment = self.outstanding.popleft()
print(f" [ACK] Segment {acked_segment.seq_start}-{acked_segment.seq_end} fully acked")
# Advance snd_una to the ack number
if ack_number > self.snd_una:
self.snd_una = ack_number
print(f" [WINDOW] Updated: snd_una={self.snd_una}, snd_wnd={self.snd_wnd}")
def can_send(self) -> int:
"""
Return the number of bytes we're allowed to send given current window.
This implements: available = snd_wnd - (snd_nxt - snd_una)
"""
inflight = self.snd_nxt - self.snd_una
available = self.snd_wnd - inflight
return max(0, available)
def send(self, data: bytes) -> int:
"""
Send as much data as the window allows. Returns bytes actually sent.
"""
available = self.can_send()
if available <= 0:
print(" [BLOCKED] Window is full — cannot send")
return 0
send_size = min(available, len(data), self.max_segment_size)
chunk = data[:send_size]
segment = Segment(
seq_start=self.snd_nxt,
seq_end=self.snd_nxt + send_size,
data=chunk
)
self.outstanding.append(segment)
self.snd_nxt += send_size
print(f" [SEND] Segment {segment.seq_start}-{segment.seq_end} "
f"({send_size} bytes), inflight={self.snd_nxt - self.snd_una}")
return send_size
# Demonstration
if __name__ == "__main__":
sender = SlidingWindowSender(initial_seq=1000)
data_to_send = b"A" * 5000 # 5000 bytes of payload
print("=== Initial State ===")
print(f"snd_una={sender.snd_una}, snd_nxt={sender.snd_nxt}, snd_wnd={sender.snd_wnd}")
# Send in chunks, simulating receiver ACKs with varying windows
remaining = data_to_send
while remaining:
sent = sender.send(remaining)
if sent == 0:
print("Waiting for ACKs to free window space...")
break
remaining = remaining[sent:]
# Simulate receiver acknowledging first 1460 bytes with window=4380
print("\n=== Receiver ACKs 1460, advertises window=4380 ===")
sender.update_window(ack_number=2460, advertised_window=4380)
# Send more data
while remaining:
sent = sender.send(remaining)
if sent == 0:
break
remaining = remaining[sent:]
# Simulate receiver acknowledging up to 3920
print("\n=== Receiver ACKs 3920, advertises window=8760 ===")
sender.update_window(ack_number=3920, advertised_window=8760)
while remaining:
sent = sender.send(remaining)
if sent == 0:
break
remaining = remaining[sent:]
print(f"\n=== Final State ===")
print(f"Remaining to send: {len(remaining)} bytes")
print(f"Outstanding segments: {len(sender.outstanding)}")
This simulation demonstrates the core constraint: the sender cannot exceed snd_wnd bytes in flight. When the receiver advertises a small window, the sender blocks. When the window opens, transmission resumes. Real TCP implementations also use the window scale option (RFC 1323) to support windows larger than 65535 bytes for high-bandwidth, high-latency paths.
Implementing Retransmission with Exponential Backoff
TCP reliability relies on retransmission of lost segments. The sender starts a timer for each outstanding segment. If an ACK doesn't arrive within the retransmission timeout (RTO), the segment is resent. The RTO doubles on each retransmission (exponential backoff) to avoid flooding a congested network.
#!/usr/bin/env python3
"""
retransmit_timer.py — Demonstrates TCP retransmission logic with
exponential backoff and RTT estimation (Jacobson/Karels algorithm).
"""
import time
import math
from dataclasses import dataclass
from collections import deque
@dataclass
class TimedSegment:
seq_start: int
seq_end: int
data: bytes
sent_time: float
retransmit_count: int = 0
first_sent_time: float = 0.0
class RetransmitManager:
"""
Manages TCP retransmission timers with Jacobson's RTT estimation.
"""
def __init__(self):
# Smoothed Round-Trip Time (SRTT) and RTTVAR (mean deviation)
self.srtt = 1.0 # Initial estimate: 1 second
self.rttvar = 0.5 # Initial deviation: 0.5 seconds
self.rto = self.srtt + 4 * self.rttvar # Initial RTO
# Constants from RFC 6298
self.ALPHA = 0.125 # 1/8
self.BETA = 0.25 # 1/4
self.K = 4 # RTTVAR multiplier
# Minimum and maximum bounds (RFC 6298)
self.MIN_RTO = 0.2 # 200 ms minimum
self.MAX_RTO = 60.0 # 60 seconds maximum
self.pending: deque[TimedSegment] = deque()
def update_rtt(self, measured_rtt: float) -> None:
"""
Update SRTT and RTTVAR using Jacobson's algorithm.
Call this when a valid ACK (non-retransmitted segment) arrives.
"""
# First measurement: initialize SRTT and RTTVAR
if self.srtt == 1.0 and self.rttvar == 0.5:
self.srtt = measured_rtt
self.rttvar = measured_rtt / 2.0
else:
# rttvar = (1 - BETA) * rttvar + BETA * |srtt - measured|
delta = abs(self.srtt - measured_rtt)
self.rttvar = (1 - self.BETA) * self.rttvar + self.BETA * delta
# srtt = (1 - ALPHA) * srtt + ALPHA * measured
self.srtt = (1 - self.ALPHA) * self.srtt + self.ALPHA * measured_rtt
# RTO = srtt + K * rttvar, clamped to bounds
self.rto = max(self.MIN_RTO, min(
self.MAX_RTO,
self.srtt + self.K * self.rttvar
))
print(f" [RTT UPDATE] measured={measured_rtt:.3f}s, "
f"srtt={self.srtt:.3f}s, rttvar={self.rttvar:.3f}s, "
f"rto={self.rto:.3f}s")
def segment_sent(self, segment: TimedSegment) -> None:
"""Record a newly sent segment with its timestamp."""
segment.sent_time = time.time()
if segment.first_sent_time == 0.0:
segment.first_sent_time = segment.sent_time
self.pending.append(segment)
def check_timeouts(self) -> list:
"""
Check all pending segments for timeout.
Returns list of segments that need retransmission.
"""
now = time.time()
timed_out = []
for segment in list(self.pending):
elapsed = now - segment.sent_time
# The effective RTO doubles with each retransmission
effective_rto = self.rto * (2 ** segment.retransmit_count)
if elapsed >= effective_rto:
timed_out.append(segment)
segment.retransmit_count += 1
segment.sent_time = now # Reset timer for this retransmission
print(f" [TIMEOUT] Segment {segment.seq_start}-{segment.seq_end} "
f"retransmit #{segment.retransmit_count}, "
f"RTO={effective_rto:.3f}s")
return timed_out
def ack_received(self, ack_number: int) -> None:
"""Remove fully acknowledged segments from pending queue."""
while self.pending and self.pending[0].seq_end <= ack_number:
segment = self.pending.popleft()
if segment.retransmit_count == 0:
# Only use non-retransmitted segments for RTT estimation
# (Karn's algorithm — don't use retransmitted segments)
measured_rtt = time.time() - segment.first_sent_time
self.update_rtt(measured_rtt)
# Demonstration
if __name__ == "__main__":
mgr = RetransmitManager()
print("=== Initial RTO ===")
print(f"SRTT={mgr.srtt:.3f}s, RTTVAR={mgr.rttvar:.3f}s, RTO={mgr.rto:.3f}s")
# Simulate sending segments
seg1 = TimedSegment(seq_start=1000, seq_end=2460, data=b"A"*1460)
seg2 = TimedSegment(seq_start=2460, seq_end=3920, data=b"B"*1460)
mgr.segment_sent(seg1)
mgr.segment_sent(seg2)
print("\n=== Simulating normal ACK for seg1 ===")
time.sleep(0.05) # 50ms RTT simulation
mgr.ack_received(2460) # ACKs seg1
# Now simulate a timeout for seg2
print("\n=== Waiting for seg2 timeout (artificial delay) ===")
time.sleep(0.3) # Exceed initial RTO
timed_out = mgr.check_timeouts()
for seg in timed_out:
print(f"Retransmitting: {seg.seq_start}-{seg.seq_end}")
mgr.segment_sent(seg) # Re-register with new timestamp
print(f"\n=== After retransmission ===")
print(f"RTO={mgr.rto:.3f}s (doubled for seg2: {mgr.rto * 2:.3f}s)")
This implementation incorporates two critical RFC 6298 details: Karn's algorithm (don't use RTT samples from retransmitted segments to avoid ambiguity) and exponential backoff (RTO doubles on each retransmission). The Jacobson/Karels algorithm provides smooth, stable RTO estimates that prevent both premature timeouts and excessive latency.
TCP Connection Teardown and TIME_WAIT
Connection termination follows a four-way handshake. The active closer sends FIN, receives ACK, receives the passive closer's FIN, and sends the final ACK. The active closer