|
| 1 | +# SPDX-License-Identifier: BSD-3-Clause |
| 2 | +# Copyright (C) 2026 CESNET z. s. p. o. |
| 3 | +# Author(s): Daniel Kondys <kondys@cesnet.cz> |
| 4 | + |
| 5 | +"""Address range tracker to prevent overlapping memory accesses.""" |
| 6 | + |
| 7 | +from typing import List, Tuple, Optional, Dict |
| 8 | +from random import randint |
| 9 | + |
| 10 | + |
| 11 | +class AddressRangeTracker: |
| 12 | + """ |
| 13 | + Tracks address ranges that are currently in use (written to RAM but not yet read). |
| 14 | + Ensures new address allocations don't overlap with existing ranges. |
| 15 | +
|
| 16 | + Supports tracking by packet ID, allowing ranges to be freed when complete |
| 17 | + packets are received rather than individual PCIe responses. |
| 18 | + """ |
| 19 | + |
| 20 | + def __init__(self, max_addr: int): |
| 21 | + """ |
| 22 | + Args: |
| 23 | + max_addr: Maximum valid address |
| 24 | + """ |
| 25 | + self.max_addr = max_addr |
| 26 | + # List of (start_addr, end_addr) tuples representing in-use ranges |
| 27 | + # end_addr is exclusive (i.e., range is [start, end)) |
| 28 | + self.in_use_ranges: List[Tuple[int, int]] = [] |
| 29 | + # Map packet ID to (start_addr, length) for tracking by ID |
| 30 | + self._range_by_id: Dict[int, Tuple[int, int]] = {} |
| 31 | + |
| 32 | + def add_range(self, start: int, length: int, pkt_id: int = None, allow_wrap: bool = False) -> bool: |
| 33 | + """ |
| 34 | + Add a new address range to track. |
| 35 | +
|
| 36 | + Args: |
| 37 | + start: Starting address |
| 38 | + length: Length of the range in bytes |
| 39 | + pkt_id: Optional packet ID to associate with this range for later removal |
| 40 | + allow_wrap: If True, the range can wrap around from end to beginning |
| 41 | +
|
| 42 | + Returns: |
| 43 | + True if range was added successfully, False if it overlaps with existing range |
| 44 | + """ |
| 45 | + end = start + length |
| 46 | + |
| 47 | + # Check for overlap with existing ranges |
| 48 | + # For wrap-around ranges, we need to check both the high and low parts |
| 49 | + for existing_start, existing_end in self.in_use_ranges: |
| 50 | + if self._ranges_overlap_wrap(start, end, existing_start, existing_end, allow_wrap): |
| 51 | + return False |
| 52 | + |
| 53 | + self.in_use_ranges.append((start, end)) |
| 54 | + if pkt_id is not None: |
| 55 | + self._range_by_id[pkt_id] = (start, length) |
| 56 | + return True |
| 57 | + |
| 58 | + def remove_range(self, start: int, length: int) -> bool: |
| 59 | + """ |
| 60 | + Remove an address range from tracking (when read is complete). |
| 61 | +
|
| 62 | + Args: |
| 63 | + start: Starting address |
| 64 | + length: Length of the range in bytes |
| 65 | +
|
| 66 | + Returns: |
| 67 | + True if range was found and removed, False otherwise |
| 68 | + """ |
| 69 | + end = start + length |
| 70 | + target = (start, end) |
| 71 | + |
| 72 | + if target in self.in_use_ranges: |
| 73 | + self.in_use_ranges.remove(target) |
| 74 | + return True |
| 75 | + return False |
| 76 | + |
| 77 | + def remove_range_by_id(self, pkt_id: int) -> bool: |
| 78 | + """ |
| 79 | + Remove an address range by packet ID. |
| 80 | +
|
| 81 | + This is used when a complete packet is received on the response interface, |
| 82 | + which may consist of multiple PCIe read completions. |
| 83 | +
|
| 84 | + Args: |
| 85 | + pkt_id: Packet ID associated with the range |
| 86 | +
|
| 87 | + Returns: |
| 88 | + True if range was found and removed, False otherwise |
| 89 | + """ |
| 90 | + if pkt_id not in self._range_by_id: |
| 91 | + return False |
| 92 | + |
| 93 | + start, length = self._range_by_id.pop(pkt_id) |
| 94 | + return self.remove_range(start, length) |
| 95 | + |
| 96 | + def find_non_overlapping_address(self, length: int, max_attempts: int = 1000, allow_wrap: bool = False) -> Optional[int]: |
| 97 | + """ |
| 98 | + Find a random address that doesn't overlap with any in-use range. |
| 99 | +
|
| 100 | + Args: |
| 101 | + length: Required length of the range |
| 102 | + max_attempts: Maximum number of random attempts before giving up |
| 103 | + allow_wrap: If True, allow addresses that wrap around from end to beginning |
| 104 | +
|
| 105 | + Returns: |
| 106 | + A valid starting address or None if no space available |
| 107 | + """ |
| 108 | + if length > self.max_addr: |
| 109 | + raise ValueError(f"Requested length ({length}) is greater than the whole address range ({self.max_addr}).") |
| 110 | + |
| 111 | + for _ in range(max_attempts): |
| 112 | + # Generate random address |
| 113 | + if allow_wrap: |
| 114 | + # Allow any address from 0 to max_addr-1 |
| 115 | + addr = randint(0, self.max_addr - 1) |
| 116 | + else: |
| 117 | + # Current behavior: only addresses that don't wrap |
| 118 | + addr = randint(0, self.max_addr - length) |
| 119 | + |
| 120 | + end = addr + length |
| 121 | + |
| 122 | + # Check if it overlaps with any in-use range |
| 123 | + overlaps = False |
| 124 | + for existing_start, existing_end in self.in_use_ranges: |
| 125 | + if self._ranges_overlap_wrap(addr, end, existing_start, existing_end, allow_wrap): |
| 126 | + overlaps = True |
| 127 | + break |
| 128 | + |
| 129 | + if not overlaps: |
| 130 | + return addr |
| 131 | + |
| 132 | + return None |
| 133 | + |
| 134 | + def _ranges_overlap(self, start1: int, end1: int, start2: int, end2: int) -> bool: |
| 135 | + """ |
| 136 | + Check if two ranges overlap. |
| 137 | + Ranges are [start, end) - inclusive start, exclusive end. |
| 138 | + """ |
| 139 | + return start1 < end2 and start2 < end1 |
| 140 | + |
| 141 | + def _ranges_overlap_wrap(self, start1: int, end1: int, start2: int, end2: int, allow_wrap: bool) -> bool: |
| 142 | + """ |
| 143 | + Check if two ranges overlap, with optional wrap-around support. |
| 144 | +
|
| 145 | + Args: |
| 146 | + start1, end1: First range [start1, end1) |
| 147 | + start2, end2: Second range [start2, end2) |
| 148 | + allow_wrap: If True, handle wrap-around ranges correctly |
| 149 | +
|
| 150 | + Returns: |
| 151 | + True if ranges overlap, False otherwise |
| 152 | + """ |
| 153 | + if not allow_wrap: |
| 154 | + # Standard overlap check |
| 155 | + return self._ranges_overlap(start1, end1, start2, end2) |
| 156 | + |
| 157 | + # With wrap-around, a range can span across max_addr boundary |
| 158 | + # We need to check if either range wraps around |
| 159 | + |
| 160 | + # Normalize ranges to be within [0, max_addr) |
| 161 | + # A range wraps if end > max_addr |
| 162 | + wraps1 = end1 > self.max_addr |
| 163 | + wraps2 = end2 > self.max_addr |
| 164 | + |
| 165 | + if not wraps1 and not wraps2: |
| 166 | + # Neither wraps - standard overlap check |
| 167 | + return self._ranges_overlap(start1, end1, start2, end2) |
| 168 | + |
| 169 | + if wraps1 and wraps2: |
| 170 | + # Both wrap - they overlap if their "wrapped parts" overlap |
| 171 | + # Wrapped part of range 1: [0, end1 % max_addr) and [start1 % max_addr, max_addr) |
| 172 | + # But since both wrap, we check if the non-wrapped portions don't cover everything |
| 173 | + # Actually, if both wrap, they always overlap (they both cover the middle) |
| 174 | + # unless one is completely contained in the other's "hole" |
| 175 | + # The "hole" of a wrapped range is [end1 % max_addr, start1 % max_addr) |
| 176 | + # This is complex - let's simplify by checking if the ranges together don't cover everything |
| 177 | + |
| 178 | + # Simpler approach: check if there's any gap in either range |
| 179 | + # Range 1 covers: [start1, max_addr) U [0, end1 - max_addr) |
| 180 | + # Range 2 covers: [start2, max_addr) U [0, end2 - max_addr) |
| 181 | + # They don't overlap only if one's covered area is completely outside the other's |
| 182 | + |
| 183 | + # Actually, if both wrap, they always overlap because they both include |
| 184 | + # addresses near max_addr and addresses near 0 |
| 185 | + return True |
| 186 | + |
| 187 | + # One wraps, one doesn't |
| 188 | + if wraps1: |
| 189 | + # Range 1 wraps: [start1, max_addr) U [0, end1 - max_addr) |
| 190 | + # Range 2 doesn't wrap: [start2, end2) |
| 191 | + # They don't overlap if range2 is completely in the "hole" of range1 |
| 192 | + # Hole of range1: [end1 - max_addr, start1) |
| 193 | + hole_start = end1 - self.max_addr |
| 194 | + hole_end = start1 |
| 195 | + # Range2 doesn't overlap with wrapped range1 if it's entirely in the hole |
| 196 | + if start2 >= hole_start and end2 <= hole_end: |
| 197 | + return False |
| 198 | + return True |
| 199 | + |
| 200 | + else: # wraps2 |
| 201 | + # Range 2 wraps: [start2, max_addr) U [0, end2 - max_addr) |
| 202 | + # Range 1 doesn't wrap: [start1, end1) |
| 203 | + # Same logic as above, just swap |
| 204 | + hole_start = end2 - self.max_addr |
| 205 | + hole_end = start2 |
| 206 | + if start1 >= hole_start and end1 <= hole_end: |
| 207 | + return False |
| 208 | + return True |
| 209 | + |
| 210 | + def is_range_available(self, start: int, length: int, allow_wrap: bool = False) -> bool: |
| 211 | + """ |
| 212 | + Check if a range is available (doesn't overlap with any in-use range). |
| 213 | +
|
| 214 | + Args: |
| 215 | + start: Starting address |
| 216 | + length: Length of the range |
| 217 | + allow_wrap: If True, handle wrap-around ranges correctly |
| 218 | +
|
| 219 | + Returns: |
| 220 | + True if range is available, False otherwise |
| 221 | + """ |
| 222 | + end = start + length |
| 223 | + for existing_start, existing_end in self.in_use_ranges: |
| 224 | + if self._ranges_overlap_wrap(start, end, existing_start, existing_end, allow_wrap): |
| 225 | + return False |
| 226 | + return True |
| 227 | + |
| 228 | + def __len__(self) -> int: |
| 229 | + """Return the number of tracked ranges.""" |
| 230 | + return len(self.in_use_ranges) |
0 commit comments