Added metrics L system and Lq LB

Author: GioeleB00

src/asyncflow/config/enums.py

@@ -189,10 +189,13 @@ class SampledMetricName(StrEnum):
   """
 
   # Mandatory metrics to collect
-  READY_QUEUE_LEN = "ready_queue_len" #length of the event loop ready q
-  EVENT_LOOP_IO_SLEEP = "event_loop_io_sleep"
+  LQ_SERVER = "ready_queue_len" #length of the event loop ready q
+  LQ_IO = "event_loop_io_sleep"
   RAM_IN_USE = "ram_in_use"
   EDGE_CONCURRENT_CONNECTION = "edge_concurrent_connection"
+  L_SYSTEM = "l_system"
+  LQ_LB = "lq_lb"
+
 
 
 class SamplePeriods(float, Enum):

src/asyncflow/metrics/collector.py

@@ -4,8 +4,10 @@
 
 import simpy
 
-from asyncflow.config.enums import SampledMetricName
+from asyncflow.config.enums import LbAlgorithmsName, SampledMetricName
+from asyncflow.runtime.actors.arrivals_generator import ArrivalsGeneratorRuntime
 from asyncflow.runtime.actors.edge import EdgeRuntime
+from asyncflow.runtime.actors.load_balancer import LoadBalancerRuntime
 from asyncflow.runtime.actors.server import ServerRuntime
 from asyncflow.schemas.settings.simulation import SimulationSettings
 
@@ -17,34 +19,42 @@
 class SampledMetricCollector:
     """class to define a centralized object to collect sampled metrics"""
 
-    def __init__(
+    def __init__(# noqa: PLR0913
         self,
         *,
+        arrivals: ArrivalsGeneratorRuntime,
         edges: list[EdgeRuntime],
         servers: list[ServerRuntime],
+        lb: LoadBalancerRuntime | None,
         env:  simpy.Environment,
         sim_settings: SimulationSettings,
         ) -> None:
         """
         Args:
+            arrivals: (ArrivalsGeneratorRuntime): usefull to compute l_system
             edges (list[EdgeRuntime]): list of the class EdgeRuntime
             servers (list[ServerRuntime]): list of server of the class ServerRuntime
+            lb (LoadBalancerRuntime): useful to compute Lq
             env (simpy.Environment): environment for the simulation
             sim_settings (SimulationSettings): general settings for the simulation
 
         """
+        self.arrivals = arrivals
         self.edges = edges
         self.servers = servers
+        self.lb = lb
         self.sim_settings = sim_settings
         self.env = env
         self._sample_period = sim_settings.sample_period_s
 
 
         # enum keys instance-level for mandatory sampled metrics to collect
-        self._conn_key   = SampledMetricName.EDGE_CONCURRENT_CONNECTION
-        self._ram_key    = SampledMetricName.RAM_IN_USE
-        self._io_key     = SampledMetricName.EVENT_LOOP_IO_SLEEP
-        self._ready_key  = SampledMetricName.READY_QUEUE_LEN
+        self._conn_key = SampledMetricName.EDGE_CONCURRENT_CONNECTION
+        self._ram_key = SampledMetricName.RAM_IN_USE
+        self._io_key = SampledMetricName.LQ_IO
+        self._ready_key = SampledMetricName.LQ_SERVER
+        self._l_system = SampledMetricName.L_SYSTEM
+        self._lq_lb = SampledMetricName.LQ_LB
 
 
     def _build_time_series(self) -> Generator[simpy.Event, None, None]:
@@ -65,6 +75,16 @@ def _build_time_series(self) -> Generator[simpy.Event, None, None]:
                     server.enabled_metrics[self._io_key].append(server.io_queue_len)
                     server.enabled_metrics[self._ready_key].append(server.ready_queue_len)
 
+            if self._l_system in self.arrivals.enabled_metrics:
+                self.arrivals.enabled_metrics[self._l_system].append(
+                    float(self.arrivals.l_system),
+                )
+
+            if (self.lb is not None and
+                self.lb is not None and
+                self.lb.lb_config.algorithms == LbAlgorithmsName.FCFS):
+                self.lb.enabled_metrics[self._lq_lb].append(self.lb.lq_lb)
+
 
 
     def start(self) -> None:

src/asyncflow/metrics/server.py

@@ -13,8 +13,8 @@
 # will be considered
 
 SERVER_METRICS = (
-    SampledMetricName.READY_QUEUE_LEN,
-    SampledMetricName.EVENT_LOOP_IO_SLEEP,
+    SampledMetricName.LQ_SERVER,
+    SampledMetricName.LQ_IO,
     SampledMetricName.RAM_IN_USE,
 )
 

src/asyncflow/metrics/simulation_analyzer.py

@@ -10,6 +10,7 @@
 from asyncflow.config.enums import (
     EventMetricName,
     LatencyKey,
+    LbAlgorithmsName,
     SampledMetricName,
 )
 from asyncflow.config.plot_constants import (
@@ -263,6 +264,17 @@ def _extract_sampled_metrics(self) -> None:
             for name, values in edge.enabled_metrics.items():
                 metrics[name.value][eid] = values
 
+        for name, values in self._generator.enabled_metrics.items():
+            aid = self._generator.arrivals.id
+            metrics[name.value][aid] = values
+
+        if (self.lb is not None and
+            self.lb.lb_config.algorithms == LbAlgorithmsName.FCFS):
+            lb_id = self.lb.lb_config.id
+            for name, values in self.lb.enabled_metrics.items():
+                # es. SampledMetricName.LQ_LB → “lq_lb”
+                metrics[name.value][lb_id] = values
+
         self.sampled_metrics = metrics
 
     # ─────────────────────────────────────────────
@@ -559,7 +571,7 @@ def plot_single_server_ready_queue(self, ax: Axes, server_id: str) -> None:
         """Plot Ready queue with mean/min/max lines and a single legend box with
         values. No trend/ewma, no legend entry for the main series.
         """
-        times, vals = self.get_series(SampledMetricName.READY_QUEUE_LEN, server_id)
+        times, vals = self.get_series(SampledMetricName.LQ_SERVER, server_id)
         if not vals:
             ax.text(0.5, 0.5, SERVER_QUEUES_PLOT.no_data, ha="center", va="center")
             return
@@ -617,7 +629,7 @@ def plot_single_server_io_queue(self, ax: Axes, server_id: str) -> None:
         """Plot I/O queue with mean/min/max lines and a single legend box with
         values. No trend/ewma, no legend entry for the main series.
         """
-        times, vals = self.get_series(SampledMetricName.EVENT_LOOP_IO_SLEEP, server_id)
+        times, vals = self.get_series(SampledMetricName.LQ_IO, server_id)
         if not vals:
             ax.text(0.5, 0.5, SERVER_QUEUES_PLOT.no_data, ha="center", va="center")
             return

src/asyncflow/runner/simulation.py

@@ -44,7 +44,7 @@
 class Startable(Protocol):
     """A protocol for runtime actors that can be started."""
 
-    def start(self) -> None:
+    def start(self) -> simpy.Process:
         """Starts the main process loop for the actor."""
         ...
 
@@ -368,8 +368,10 @@ def _start_all_processes(self) -> None:
     def _start_metric_collector(self) -> None:
         """One coroutine that snapshots RAM / queues / connections."""
         SampledMetricCollector(
+            arrivals=next(iter(self._arrivals_runtime.values())),
             edges=list(self._edges_runtime.values()),
             servers=list(self._servers_runtime.values()),
+            lb=self._lb_runtime,
             env=self.env,
             sim_settings=self.simulation_settings,
         ).start()

src/asyncflow/runtime/actors/arrivals_generator.py

@@ -20,6 +20,7 @@
 
     import simpy
 
+    from asyncflow.config.enums import SampledMetricName
     from asyncflow.runtime.actors.edge import EdgeRuntime
     from asyncflow.schemas.arrivals.generator import ArrivalsGenerator
     from asyncflow.schemas.settings.simulation import SimulationSettings
@@ -62,7 +63,14 @@ def __init__( # noqa: PLR0913
         self.arrivals_generator_box = arrivals_generator_box
         self.completed_box = completed_box
         self.id_counter = 0
+
+
+        # necessary to collect metrics (global throughput and latency)
         self._rqs_clock: list[RqsClock] = []
+        # necessary to collect simultaneous rqs in the system
+        self._l_system: int = 0
+        # dict for the collector to have the time series
+        self.enabled_metrics: dict[SampledMetricName, list[float]] = {}
 
 
     def _next_id(self) -> int:
@@ -93,6 +101,7 @@ def _event_arrival(self) -> Generator[simpy.Event, None, None]:
                 self.arrivals.id,
                 self.env.now,
             )
+            self._l_system += 1
             # transport is a method of the edge runtime
             # which define the step of how the state is moving
             # from one node to another
@@ -109,17 +118,22 @@ def _collector(self) -> Generator[simpy.Event, None, None]:
                 finish=state.finish_time,
             )
             self._rqs_clock.append(clock_data)
+            self._l_system -= 1
             yield self.completed_box.put(state)
 
 
     def start(self) -> simpy.Process:
-        """Passing the structure as a simpy process"""
-        self.env.process(self._event_arrival())
+        """Start the simpy processes"""
+        p_arr = self.env.process(self._event_arrival())
         self.env.process(self._collector())
-
-
+        return p_arr
 
     @property
     def rqs_clock(self) -> list[RqsClock]:
         """Readable version to compute aggregate metrics"""
         return self._rqs_clock
+
+    @property
+    def l_system(self) -> int:
+        """Readable version to sample the metric"""
+        return self._l_system

src/asyncflow/runtime/actors/client.py

@@ -47,7 +47,7 @@ def _forwarder(self) -> Generator[simpy.Event, None, None]:
 
             self.out_edge.transport(state)
 
-    def start(self) -> None:
+    def start(self) -> simpy.Process:
         """Initialization of the process"""
-        self.env.process(self._forwarder())
+        return self.env.process(self._forwarder())
 

src/asyncflow/runtime/actors/load_balancer.py

@@ -7,7 +7,7 @@
 
 import simpy
 
-from asyncflow.config.enums import LbAlgorithmsName, SystemNodes
+from asyncflow.config.enums import LbAlgorithmsName, SampledMetricName, SystemNodes
 from asyncflow.runtime.actors.edge import EdgeRuntime
 from asyncflow.runtime.actors.routing.lb_algorithms import (
     LB_TABLE,
@@ -65,6 +65,11 @@ def __init__(
         # queue theory
         self._lb_waiting_time: list[float] = []
 
+        # counter to collect when algo is fcfs lq to compare with queue theory
+        self._lq_lb: int = 0
+        # dict to collect the time series
+        self.enabled_metrics: dict[SampledMetricName, list[float]] = {}
+
 
     # Helpers FCFS
 
@@ -99,7 +104,7 @@ def _forwarder(self) -> Generator[simpy.Event, None, None]:
             state: RequestState = yield self.lb_box.get()  # type: ignore[assignment]
 
             if self.lb_config.algorithms == LbAlgorithmsName.FCFS:
-
+                self._lq_lb += 1
                 hist = getattr(state, "history", None)
                 if hist:
                     last = hist[-1]
@@ -139,6 +144,7 @@ def _forwarder(self) -> Generator[simpy.Event, None, None]:
                     self._lb_waiting_time.append(waiting_time)
 
                 edge_rt = self.lb_out_edges[edge_id]
+                self._lq_lb -= 1
                 edge_rt.transport(state)
             else:
                 state.record_hop(
@@ -149,12 +155,17 @@ def _forwarder(self) -> Generator[simpy.Event, None, None]:
                 edge_rt = LB_TABLE[self.lb_config.algorithms](self.lb_out_edges)
                 edge_rt.transport(state)
 
-    def start(self) -> None:
+    def start(self) -> simpy.Process:
         """Start the process and populate FIFO"""
         self._prime_free_edges()
-        self.env.process(self._forwarder())
+        return self.env.process(self._forwarder())
 
     @property
     def lb_waiting_times(self) -> Sequence[float]:
         """Read-only view of LB FCFS waiting times (one per waited request)."""
         return tuple(self._lb_waiting_time)
+
+    @property
+    def lq_lb(self) -> int:
+        """Readable version to sample the metric"""
+        return self._lq_lb

src/asyncflow/runtime/actors/server.py

@@ -189,10 +189,6 @@ def _dispatcher(self) -> Generator[simpy.Event, None, None]:
             # Spawn a new, independent process to handle this request
             self.env.process(self._handle_request(request_state))
 
-    def start(self) -> None:
-        """Generate the process to simulate the server inside simpy env"""
-        self.env.process(self._dispatcher())
-
     # right now we disable the warnings but a refactor will be done soon
     def _handle_request( # noqa: PLR0915, PLR0912, C901
         self,
@@ -444,6 +440,9 @@ def _handle_request( # noqa: PLR0915, PLR0912, C901
         assert self.out_edge is not None
         self.out_edge.transport(state)
 
+    def start(self) -> simpy.Process:
+        """Generate the process to simulate the server inside simpy env"""
+        return self.env.process(self._dispatcher())
 
     # we need these accessor because we need to read these private attribute
     # in the sampled metric collector

src/asyncflow/runtime/events/injection.py

@@ -254,10 +254,11 @@ def _assign_server_state(self) -> Generator[simpy.Event, None, None]:
 
 
 
-    def start(self) -> None:
+    def start(self) -> tuple[simpy.Process, simpy.Process]:
         """Start both edge-spike and server-outage timelines."""
-        self.env.process(self._assign_edges_spike())
-        self.env.process(self._assign_server_state())
+        p1 = self.env.process(self._assign_edges_spike())
+        p2 = self.env.process(self._assign_server_state())
+        return p1, p2
 
 
     @property