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

def calculate_sr_buffer_requirements(
    window_size: int,
    max_frame_size: int = 1500,
    num_connections: int = 1
) -> dict:
    """
    Calculate complete buffer requirements for Selective Repeat.
    
    Args:
        window_size: Number of frames in sliding window (W)
        max_frame_size: Maximum frame size in bytes (default: 1500 MTU)
        num_connections: Number of concurrent connections
    
    Returns:
        Dictionary with buffer requirements
    """
    
    # Per-connection sender buffer
    # Must store W frames for potential retransmission
    sender_buffer_per_conn = window_size * max_frame_size
    
    # Per-connection receiver buffer  
    # Must store up to W frames for out-of-order handling
    receiver_buffer_per_conn = window_size * max_frame_size
    
    # Per-connection total
    total_per_conn = sender_buffer_per_conn + receiver_buffer_per_conn
    
    # Per-connection overhead (metadata, timers, etc.)
    # Typical: ~100 bytes per frame for tracking
    metadata_per_conn = window_size * 100  # Conservative estimate
    
    # Full system requirements
    total_sender = sender_buffer_per_conn * num_connections
    total_receiver = receiver_buffer_per_conn * num_connections
    total_metadata = metadata_per_conn * num_connections
    
    return {
        'window_size': window_size,
        'frame_size': max_frame_size,
        'connections': num_connections,
        
        # Per-connection
        'sender_per_conn_kb': sender_buffer_per_conn / 1024,
        'receiver_per_conn_kb': receiver_buffer_per_conn / 1024,
        'total_per_conn_kb': total_per_conn / 1024,
        
        # System-wide
        'total_sender_mb': total_sender / (1024 * 1024),
        'total_receiver_mb': total_receiver / (1024 * 1024),
        'total_data_mb': (total_sender + total_receiver) / (1024 * 1024),
        'total_with_overhead_mb': (total_sender + total_receiver + total_metadata) / (1024 * 1024),
    }
 
 
# Example calculations
print("=== Low-latency LAN (W=8) ===")
lan = calculate_sr_buffer_requirements(window_size=8, num_connections=100)
print(f"Per connection: {lan['total_per_conn_kb']:.1f} KB")
print(f"100 connections: {lan['total_with_overhead_mb']:.1f} MB")
 
print("\n=== Satellite Link (W=64) ===")
sat = calculate_sr_buffer_requirements(window_size=64, num_connections=10)
print(f"Per connection: {sat['total_per_conn_kb']:.1f} KB") 
print(f"10 connections: {sat['total_with_overhead_mb']:.1f} MB")
 
print("\n=== High-throughput Server (W=128, 10K connections) ===")
server = calculate_sr_buffer_requirements(window_size=128, num_connections=10000)
print(f"Per connection: {server['total_per_conn_kb']:.1f} KB")
print(f"10K connections: {server['total_with_overhead_mb']:.1f} MB = {server['total_with_overhead_mb']/1024:.1f} GB")

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

import mmap
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class FrameMetadata:
    """Compact per-frame tracking information."""
    sequence_number: int = -1    # -1 means slot is free
    data_offset: int = 0         # Offset into data buffer
    data_length: int = 0         # Actual frame size
    timestamp: float = 0.0       # For timeout calculation
    status: int = 0              # 0=free, 1=sent, 2=acked
 
class EfficientSRBuffer:
    """
    Memory-efficient buffer for Selective Repeat.
    
    Uses contiguous pre-allocated memory for both data and metadata,
    ensuring cache efficiency and zero per-frame allocation overhead.
    """
    
    # Frame status constants
    FREE = 0
    SENT = 1  
    ACKED = 2
    RECEIVED = 3
    
    def __init__(self, window_size: int, max_frame_size: int = 1500):
        self.W = window_size
        self.frame_size = max_frame_size
        
        # Pre-allocate contiguous data buffer
        # Using mmap for efficient memory management
        self.data_buffer_size = window_size * max_frame_size
        self.data_buffer = mmap.mmap(-1, self.data_buffer_size)
        
        # Pre-allocate metadata array (fixed-size, no dict overhead)
        self.metadata = [FrameMetadata() for _ in range(window_size)]
        
        # Free list for O(1) slot allocation
        self.free_slots = list(range(window_size))
        
        # Statistics
        self.stats = {
            'stores': 0,
            'retrievals': 0,
            'max_utilization': 0
        }
    
    def _slot_for_seq(self, seq_num: int) -> int:
        """Map sequence number to buffer slot (circular)."""
        return seq_num % self.W
    
    def store(self, seq_num: int, data: bytes) -> bool:
        """
        Store a frame in the buffer.
        
        Zero-copy when possible: data goes directly into pre-allocated memory.
        """
        slot = self._slot_for_seq(seq_num)
        
        if self.metadata[slot].status != self.FREE:
            return False  # Slot in use
        
        if len(data) > self.frame_size:
            raise ValueError(f"Frame too large: {len(data)} > {self.frame_size}")
        
        # Calculate offset into data buffer
        offset = slot * self.frame_size
        
        # Copy data into pre-allocated buffer (single memcpy)
        self.data_buffer[offset:offset + len(data)] = data
        
        # Update metadata
        self.metadata[slot] = FrameMetadata(
            sequence_number=seq_num,
            data_offset=offset,
            data_length=len(data),
            timestamp=time.time(),
            status=self.SENT
        )
        
        self.stats['stores'] += 1
        current_util = sum(1 for m in self.metadata if m.status != self.FREE)
        self.stats['max_utilization'] = max(self.stats['max_utilization'], current_util)
        
        return True
    
    def retrieve(self, seq_num: int) -> Optional[bytes]:
        """
        Retrieve a frame from the buffer.
        
        Returns copy of data (caller owns it).
        """
        slot = self._slot_for_seq(seq_num)
        meta = self.metadata[slot]
        
        if meta.status == self.FREE or meta.sequence_number != seq_num:
            return None
        
        self.stats['retrievals'] += 1
        
        # Return copy of data from buffer
        return bytes(self.data_buffer[meta.data_offset:meta.data_offset + meta.data_length])
    
    def mark_acked(self, seq_num: int) -> bool:
        """Mark a frame as acknowledged (ready for reuse)."""
        slot = self._slot_for_seq(seq_num)
        meta = self.metadata[slot]
        
        if meta.sequence_number == seq_num:
            meta.status = self.ACKED
            return True
        return False
    
    def release(self, seq_num: int):
        """Release a slot for reuse."""
        slot = self._slot_for_seq(seq_num)
        self.metadata[slot] = FrameMetadata()  # Reset to free
    
    def get_memory_usage(self) -> dict:
        """Return current memory usage statistics."""
        used_slots = sum(1 for m in self.metadata if m.status != self.FREE)
        
        return {
            'total_allocated_kb': self.data_buffer_size / 1024,
            'used_slots': used_slots,
            'used_data_kb': used_slots * self.frame_size / 1024,
            'utilization_pct': used_slots / self.W * 100,
            'metadata_overhead_bytes': len(self.metadata) * 48,  # ~48 bytes per FrameMetadata
        }
    
    def __del__(self):
        """Clean up memory-mapped buffer."""
        self.data_buffer.close()

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

from heapq import heappush, heappop
from dataclasses import dataclass, field
from typing import Callable
import time
 
@dataclass(order=True)
class TimerEvent:
    """Represents a scheduled timeout event."""
    expire_time: float
    seq_num: int = field(compare=False)
    callback: Callable = field(compare=False)
    cancelled: bool = field(default=False, compare=False)
 
class EfficientTimerWheel:
    """
    Timer wheel for efficient per-frame timeout management.
    
    Instead of W separate timer objects, maintains a single
    priority queue of timeout events. O(log n) insert/cancel,
    O(1) check for next expiry.
    
    Memory: Single heap + event objects (~50 bytes each)
    vs W separate threading.Timer objects (~300 bytes each)
    """
    
    def __init__(self, default_timeout: float = 1.0):
        self.timeout = default_timeout
        self.timer_heap = []  # Min-heap by expire_time
        self.active_timers = {}  # seq_num -> TimerEvent (for cancellation)
        
    def schedule(self, seq_num: int, callback: Callable):
        """Schedule a timeout for a specific frame."""
        # Cancel any existing timer for this seq_num
        self.cancel(seq_num)
        
        expire_time = time.time() + self.timeout
        event = TimerEvent(expire_time, seq_num, callback)
        
        heappush(self.timer_heap, event)
        self.active_timers[seq_num] = event
        
    def cancel(self, seq_num: int):
        """Cancel timeout for a frame (e.g., when ACKed)."""
        if seq_num in self.active_timers:
            # Mark as cancelled (lazy deletion)
            self.active_timers[seq_num].cancelled = True
            del self.active_timers[seq_num]
    
    def get_next_expiry(self) -> float:
        """Get time until next timeout (for select/poll)."""
        while self.timer_heap:
            event = self.timer_heap[0]
            if event.cancelled:
                heappop(self.timer_heap)
                continue
            return max(0, event.expire_time - time.time())
        return float('inf')  # No pending timers
    
    def process_expired(self):
        """Process all expired timers, calling their callbacks."""
        now = time.time()
        expired_count = 0
        
        while self.timer_heap:
            event = self.timer_heap[0]
            
            if event.cancelled:
                heappop(self.timer_heap)
                continue
                
            if event.expire_time > now:
                break  # No more expired
            
            # Remove and process
            heappop(self.timer_heap)
            
            if event.seq_num in self.active_timers:
                del self.active_timers[event.seq_num]
                event.callback(event.seq_num)  # Trigger retransmission
                expired_count += 1
        
        return expired_count
    
    def get_memory_usage(self) -> dict:
        """Return timer subsystem memory usage."""
        return {
            'heap_entries': len(self.timer_heap),
            'active_timers': len(self.active_timers),
            'estimated_bytes': len(self.timer_heap) * 50 + len(self.active_timers) * 8,
        }