Loading src/tests/tools/simap_server/simap_client/SimapMetricsGenerator.py +37 −13 Original line number Diff line number Diff line Loading @@ -13,19 +13,26 @@ # limitations under the License. import random import math import logging from typing import Dict, List, Tuple LOGGER = logging.getLogger(__name__) # Link profiles: (base_bw%, base_latency_ms, sensitivity) # Sensitivity: 1.0 = highly affected by load, 0.3 = minimally affected # Congestion curve types CURVE_LINEAR = 'linear' # x - steady increase CURVE_EXPONENTIAL = 'exponential' # exp(x)-1 - slow start, rapid end CURVE_LOGARITHMIC = 'logarithmic' # log(1+x) - fast start, plateau # Link profiles: (base_bw%, base_latency_ms, sensitivity, curve_type) # - sensitivity: 1.0 = highly affected by load, 0.3 = minimally affected # - curve_type: how congestion scales with load LINK_PROFILES = { 'L1' : (15.0, 1.0, 1.0), 'L3' : (10.0, 0.8, 0.7), 'L5' : ( 8.0, 0.3, 0.3), 'L9' : ( 8.0, 0.3, 0.3), 'L13': (12.0, 0.5, 0.5), 'L1' : (15.0, 1.0, 1.0, CURVE_EXPONENTIAL), 'L3' : (10.0, 0.8, 0.7, CURVE_EXPONENTIAL), 'L5' : ( 8.0, 0.3, 0.3, CURVE_LINEAR), 'L9' : ( 8.0, 0.3, 0.3, CURVE_LINEAR), 'L13': (12.0, 0.5, 0.5, CURVE_LOGARITHMIC), } MAX_SERVICES = 5 Loading Loading @@ -77,22 +84,39 @@ class SimapMetricsGenerator: """Return all domain service IDs.""" return {k: v.copy() for k, v in self._service_ids.items()} def generate_link_metrics(self, link_id: str) -> Tuple[float, float]: def _compute_congestion_factor(self, curve_type: str, load_ratio: float) -> float: """ Compute congestion factor based on curve type and load ratio (0-1). """ Generate BW and latency for a specific TE link using non-linear congestion. if curve_type == CURVE_LINEAR: return load_ratio elif curve_type == CURVE_EXPONENTIAL: # Exponential: slow start, rapid increase at high load return (math.exp(load_ratio * 2) - 1) / (math.e ** 2 - 1) elif curve_type == CURVE_LOGARITHMIC: # Logarithmic: fast initial increase, then plateau return math.log1p(load_ratio * 2.7) / math.log1p(2.7) else: return load_ratio # Default to linear def generate_link_metrics(self, link_id: str) -> Tuple[float, float]: """ Generate BW and latency for a specific TE link using distinct congestion patterns. Returns: Tuple of (bandwidth_utilization%, latency_ms) """ if link_id not in LINK_PROFILES: raise ValueError(f"Unknown link ID: {link_id}") base_bw, base_latency, sensitivity = LINK_PROFILES[link_id] base_bw, base_latency, sensitivity, curve_type = LINK_PROFILES[link_id] # Load ratio (0 to 1) load_ratio = self._service_count / MAX_SERVICES # Non-linear congestion factor: quadratic scaling congestion_factor = (self._service_count / MAX_SERVICES) ** 2 # Compute congestion factor using link-specific curve congestion_factor = self._compute_congestion_factor(curve_type, load_ratio) # Calculate metrics with congestion # Calculate base metrics with congestion bw_utilization = base_bw + (congestion_factor * sensitivity * 60.0) latency = base_latency * (1.0 + congestion_factor * sensitivity * 4.0) Loading src/tests/tools/simap_server/simap_client/__main__.py +3 −2 Original line number Diff line number Diff line Loading @@ -61,7 +61,8 @@ def main() -> None: generator = SimapMetricsGenerator(service_count=4) for i in range(1000): # Randomly change service count (1-5) each iteration # Randomly change service count (1-5) every 5 iterations if i % 5 == 0: generator.set_service_count(random.randint(1, 5)) # Generate TE link metrics based on current service count Loading Loading
src/tests/tools/simap_server/simap_client/SimapMetricsGenerator.py +37 −13 Original line number Diff line number Diff line Loading @@ -13,19 +13,26 @@ # limitations under the License. import random import math import logging from typing import Dict, List, Tuple LOGGER = logging.getLogger(__name__) # Link profiles: (base_bw%, base_latency_ms, sensitivity) # Sensitivity: 1.0 = highly affected by load, 0.3 = minimally affected # Congestion curve types CURVE_LINEAR = 'linear' # x - steady increase CURVE_EXPONENTIAL = 'exponential' # exp(x)-1 - slow start, rapid end CURVE_LOGARITHMIC = 'logarithmic' # log(1+x) - fast start, plateau # Link profiles: (base_bw%, base_latency_ms, sensitivity, curve_type) # - sensitivity: 1.0 = highly affected by load, 0.3 = minimally affected # - curve_type: how congestion scales with load LINK_PROFILES = { 'L1' : (15.0, 1.0, 1.0), 'L3' : (10.0, 0.8, 0.7), 'L5' : ( 8.0, 0.3, 0.3), 'L9' : ( 8.0, 0.3, 0.3), 'L13': (12.0, 0.5, 0.5), 'L1' : (15.0, 1.0, 1.0, CURVE_EXPONENTIAL), 'L3' : (10.0, 0.8, 0.7, CURVE_EXPONENTIAL), 'L5' : ( 8.0, 0.3, 0.3, CURVE_LINEAR), 'L9' : ( 8.0, 0.3, 0.3, CURVE_LINEAR), 'L13': (12.0, 0.5, 0.5, CURVE_LOGARITHMIC), } MAX_SERVICES = 5 Loading Loading @@ -77,22 +84,39 @@ class SimapMetricsGenerator: """Return all domain service IDs.""" return {k: v.copy() for k, v in self._service_ids.items()} def generate_link_metrics(self, link_id: str) -> Tuple[float, float]: def _compute_congestion_factor(self, curve_type: str, load_ratio: float) -> float: """ Compute congestion factor based on curve type and load ratio (0-1). """ Generate BW and latency for a specific TE link using non-linear congestion. if curve_type == CURVE_LINEAR: return load_ratio elif curve_type == CURVE_EXPONENTIAL: # Exponential: slow start, rapid increase at high load return (math.exp(load_ratio * 2) - 1) / (math.e ** 2 - 1) elif curve_type == CURVE_LOGARITHMIC: # Logarithmic: fast initial increase, then plateau return math.log1p(load_ratio * 2.7) / math.log1p(2.7) else: return load_ratio # Default to linear def generate_link_metrics(self, link_id: str) -> Tuple[float, float]: """ Generate BW and latency for a specific TE link using distinct congestion patterns. Returns: Tuple of (bandwidth_utilization%, latency_ms) """ if link_id not in LINK_PROFILES: raise ValueError(f"Unknown link ID: {link_id}") base_bw, base_latency, sensitivity = LINK_PROFILES[link_id] base_bw, base_latency, sensitivity, curve_type = LINK_PROFILES[link_id] # Load ratio (0 to 1) load_ratio = self._service_count / MAX_SERVICES # Non-linear congestion factor: quadratic scaling congestion_factor = (self._service_count / MAX_SERVICES) ** 2 # Compute congestion factor using link-specific curve congestion_factor = self._compute_congestion_factor(curve_type, load_ratio) # Calculate metrics with congestion # Calculate base metrics with congestion bw_utilization = base_bw + (congestion_factor * sensitivity * 60.0) latency = base_latency * (1.0 + congestion_factor * sensitivity * 4.0) Loading
src/tests/tools/simap_server/simap_client/__main__.py +3 −2 Original line number Diff line number Diff line Loading @@ -61,7 +61,8 @@ def main() -> None: generator = SimapMetricsGenerator(service_count=4) for i in range(1000): # Randomly change service count (1-5) each iteration # Randomly change service count (1-5) every 5 iterations if i % 5 == 0: generator.set_service_count(random.randint(1, 5)) # Generate TE link metrics based on current service count Loading