Calculations theory vs observed indipendent

Author: GioeleB00

asyncflow_queue_limit/asyncflow_mmc.ipynb

@@ -20,7 +20,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 111,
+   "execution_count": 9,
    "id": "3e168d4a",
    "metadata": {},
    "outputs": [],
@@ -45,7 +45,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 112,
+   "execution_count": 10,
    "id": "dd39a8e3",
    "metadata": {},
    "outputs": [
@@ -112,7 +112,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 113,
+   "execution_count": 11,
    "id": "d2937e5e",
    "metadata": {},
    "outputs": [],
@@ -168,9 +168,9 @@
     "        LinkEdge(id=\"lb-srv1\",     source=\"lb-1\",   target=\"srv-1\",   ),\n",
     "        LinkEdge(id=\"lb-srv2\",     source=\"lb-1\",   target=\"srv-2\",   ),\n",
     "        LinkEdge(id=\"lb-srv3\",     source=\"lb-1\",   target=\"srv-3\",   ),\n",
-    "        LinkEdge(id=\"srv1-client\", source=\"srv-1\",  target=\"client-1\",),\n",
-    "        LinkEdge(id=\"srv2-client\", source=\"srv-2\",  target=\"client-1\",),\n",
-    "        LinkEdge(id=\"srv3-client\", source=\"srv-3\",  target=\"client-1\",),\n",
+    "        LinkEdge(id=\"srv1-client\", source=\"srv-1\",  target=\"rqs-1\",),\n",
+    "        LinkEdge(id=\"srv2-client\", source=\"srv-2\",  target=\"rqs-1\",),\n",
+    "        LinkEdge(id=\"srv3-client\", source=\"srv-3\",  target=\"rqs-1\",),\n",
     "    ]\n",
     "\n",
     "    settings = SimulationSettings(\n",
@@ -201,7 +201,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 114,
+   "execution_count": 12,
    "id": "d0634bc8",
    "metadata": {},
    "outputs": [
@@ -285,47 +285,73 @@
     "\n",
     "---\n",
     "\n",
-    "## Observed (from simulation)\n",
+    "## Observed (from simulation, FCFS/Erlang-C)\n",
     "\n",
-    "After processing metrics:\n",
+    "Under FCFS we derive **each KPI directly** from sampled time series or per-request buckets, **not** by deriving one observed KPI from another. This keeps Little’s law checks as **independent validations**, not definitions.\n",
     "\n",
-    "1. **Arrival rate**:\n",
+    "1. **Arrival rate**\n",
     "\n",
     "$$\n",
-    "\\lambda_{\\text{Observed}} = \\text{mean throughput (client completions)}\n",
+    "\\lambda_{\\text{obs}}=\\mathrm{mean}(\\text{Throughput RPS})\n",
     "$$\n",
     "\n",
-    "2. **Service rate**:\n",
+    "Mean of the throughput series (client completions per fixed window, typically 1 s).\n",
+    "\n",
+    "2. **Service rate**\n",
     "\n",
     "$$\n",
-    "\\mu_{\\text{Observed}} = 1 / \\overline{S}, \\quad \\overline{S} = \\text{mean(service\\_time)}\n",
+    "\\mu_{\\text{obs}}=\\frac{1}{\\overline{S}},\\quad \n",
+    "\\overline{S}=\\mathrm{mean}(\\texttt{SERVICE\\_TIME})\n",
     "$$\n",
     "\n",
-    "3. **End-to-end latency**:\n",
+    "Computed from per-request server buckets, aggregated across all servers.\n",
+    "\n",
+    "3. **End-to-end time**\n",
     "\n",
     "$$\n",
-    "W_{\\text{Observed}} = \\text{mean(client latencies)}\n",
+    "W_{\\text{obs}}=\\mathrm{mean}(\\text{client latencies})\n",
     "$$\n",
     "\n",
-    "4. **Waiting time**:\n",
+    "From the generator’s request clocks (start/finish per request).\n",
+    "\n",
+    "4. **Queue waiting time (LB)**\n",
     "\n",
     "$$\n",
-    "W_{q,\\text{Observed}} = \\text{mean(waiting\\_time)} \n",
+    "W_{q,\\text{obs}}=\\mathrm{mean}(\\text{LB waiting times})\n",
     "$$\n",
     "\n",
-    "5. **Little’s law check**:\n",
+    "Mean of the **load balancer FCFS** waiting time recorded per request (not inferred from queue length).\n",
+    "\n",
+    "5. **Mean number in system**\n",
     "\n",
     "$$\n",
-    "L_{\\text{Observed}} = \\lambda_{\\text{Observed}} W_{\\text{Observed}}, \\qquad\n",
-    "L_{q,\\text{Observed}} = \\lambda_{\\text{Observed}} W_{q,\\text{Observed}}\n",
+    "L_{\\text{obs}}=\\mathrm{mean}(\\texttt{L\\_SYSTEM})\n",
     "$$\n",
     "\n",
-    "6. **Utilization**:\n",
+    "Time average of the sampled series of **concurrent requests in the system** (sampled every $\\Delta t$).\n",
+    "*Fallback if the series is unavailable in tests:* $L_{\\text{obs}}=\\lambda_{\\text{obs}} W_{\\text{obs}}$.\n",
+    "\n",
+    "6. **Mean number in queue (LB)**\n",
     "\n",
     "$$\n",
-    "\\rho_{\\text{Observed}} = \\lambda_{\\text{Observed}}/(c\\,\\mu_{\\text{Observed}})\n",
+    "L_{q,\\text{obs}}=\\mathrm{mean}(\\texttt{LQ\\_LB})\n",
     "$$\n",
     "\n",
+    "Time average of the **LB queue length** series.\n",
+    "*Fallback:* $L_{q,\\text{obs}}=\\lambda_{\\text{obs}} W_{q,\\text{obs}}$.\n",
+    "\n",
+    "7. **Global utilization**\n",
+    "\n",
+    "$$\n",
+    "\\rho_{\\text{obs}}=\\mathrm{mean}(\\texttt{SERVER\\_UTILIZATION})\n",
+    "$$\n",
+    "\n",
+    "Each server samples a 0/1 “busy” indicator; $\\rho_{\\text{obs}}$ is the time average (with 1 core/server it matches the busy fraction).\n",
+    "*Fallback:* $\\rho_{\\text{obs}}=\\lambda_{\\text{obs}}/(c\\,\\mu_{\\text{obs}})$.\n",
+    "\n",
+    "> Because each quantity comes from its own primary measurement (series or buckets), cross-relations like $L=\\lambda W$ and $L_q=\\lambda W_q$ are genuine **consistency checks**, not tautologies.\n",
+    "\n",
+    "\n",
     "---\n",
     "\n",
     "## Comparison\n",
@@ -339,7 +365,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 115,
+   "execution_count": 13,
    "id": "ccd7379b",
    "metadata": {},
    "outputs": [
@@ -352,13 +378,13 @@
       "-------------------------------------------------------------------\n",
       "sym  metric                    theory    observed        abs   rel%\n",
       "-------------------------------------------------------------------\n",
-      "λ    Arrival rate (1/s)    270.000000  269.997500  -0.002500  -0.00\n",
-      "μ    Service rate (1/s)    100.000000   99.925514  -0.074486  -0.07\n",
-      "rho  Utilization             0.900000    0.900663   0.000663   0.07\n",
-      "L    Mean items in sys      10.053549    9.991820  -0.061730  -0.61\n",
-      "Lq   Mean items in queue     7.353549    7.289848  -0.063701  -0.87\n",
-      "W    Mean time in sys (s)    0.037235    0.037007  -0.000228  -0.61\n",
-      "Wq   Mean waiting (s)        0.027235    0.027000  -0.000236  -0.87\n",
+      "λ    Arrival rate (1/s)    270.000000  269.696111  -0.303889  -0.11\n",
+      "μ    Service rate (1/s)    100.000000  100.016488   0.016488   0.02\n",
+      "rho  Utilization             0.900000    0.898839  -0.001161  -0.13\n",
+      "L    Mean items in sys      10.053549    9.849654  -0.203895  -2.03\n",
+      "Lq   Mean items in queue     7.353549    7.153121  -0.200428  -2.73\n",
+      "W    Mean time in sys (s)    0.037235    0.036521  -0.000714  -1.92\n",
+      "Wq   Mean waiting (s)        0.027235    0.026523  -0.000712  -2.62\n",
       "===================================================================\n"
      ]
     }

src/asyncflow/config/enums.py

@@ -195,6 +195,7 @@ class SampledMetricName(StrEnum):
   EDGE_CONCURRENT_CONNECTION = "edge_concurrent_connection"
   L_SYSTEM = "l_system"
   LQ_LB = "lq_lb"
+  SERVER_UTILIZATION = "server_utilization"
 
 
 

src/asyncflow/metrics/collector.py

@@ -53,8 +53,9 @@ def __init__(# noqa: PLR0913
         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
+        self._l_system_key = SampledMetricName.L_SYSTEM
+        self._lq_lb_key = SampledMetricName.LQ_LB
+        self._server_utilization_key = SampledMetricName.SERVER_UTILIZATION
 
 
     def _build_time_series(self) -> Generator[simpy.Event, None, None]:
@@ -74,16 +75,18 @@ def _build_time_series(self) -> Generator[simpy.Event, None, None]:
                     server.enabled_metrics[self._ram_key].append(server.ram_in_use)
                     server.enabled_metrics[self._io_key].append(server.io_queue_len)
                     server.enabled_metrics[self._ready_key].append(server.ready_queue_len)
+                    server.enabled_metrics[
+                        self._server_utilization_key
+                        ].append(server.server_utilization)
 
-            if self._l_system in self.arrivals.enabled_metrics:
-                self.arrivals.enabled_metrics[self._l_system].append(
+            if self._l_system_key in self.arrivals.enabled_metrics:
+                self.arrivals.enabled_metrics[self._l_system_key].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)
+                self.lb.enabled_metrics[self._lq_lb_key].append(self.lb.lq_lb)
 
 
 

src/asyncflow/metrics/server.py

@@ -16,6 +16,7 @@
     SampledMetricName.LQ_SERVER,
     SampledMetricName.LQ_IO,
     SampledMetricName.RAM_IN_USE,
+    SampledMetricName.SERVER_UTILIZATION,
 )
 
 def build_server_metrics(

src/asyncflow/queue_theory_analysis/mmc.py

@@ -15,6 +15,7 @@
     EndpointStepCPU,
     LatencyKey,
     LbAlgorithmsName,
+    SampledMetricName,
 )
 from asyncflow.queue_theory_analysis.base import QueueTheoryBase
 from asyncflow.schemas.common.random_variables import RVConfig
@@ -474,18 +475,20 @@ def _observed_kpis(
         - λ̂: mean throughput
         - μ̂: 1 / mean(service_time)
         - Ŵ: mean client latency
-        - Wq̂: mean waiting_time (server arrays)
-        - L̂: λ̂ * Ŵ
-        - Lq̂: λ̂ * Wq̂
-        - rhô: λ̂ / (c μ̂)
+        - Wq̂: FCFS -> mean LB waits; else -> mean server waiting_time
+        - L̂: FCFS -> mean L_SYSTEM; else -> λ̂ * Ŵ
+        - Lq̂: FCFS -> mean LQ_LB; else -> λ̂ * Wq̂
+        - rho: FCFS -> mean SERVER_UTILIZATION; else -> λ̂ / (c μ̂)
+        all computations are independent: we calculate each
+        variable in a separate way
         """
         self._ensure_metrics_processed(results_analyzer)
 
         lambda_hat = self._observed_lambda_rate(results_analyzer)
         mu_hat = self._observed_mu_rate(results_analyzer)
         server_count = self._server_count(payload)
 
-        # Ŵ from latency stats (client-side);
+        # Ŵ from latency stats (generator-side);
         lat_stats = results_analyzer.get_latency_stats()
         w_hat = float(lat_stats.get(LatencyKey.MEAN, 0.0))
 
@@ -494,8 +497,42 @@ def _observed_kpis(
 
         # Collect waiting time from LB if the algo is FCFS
         if is_fcfs:
+            # --- L̂: mean L_SYSTEM timeseries (fallback λ̂·Ŵ) ---
+            lsys_map = results_analyzer.get_metric_map(SampledMetricName.L_SYSTEM)
+            aid = payload.arrivals.id
+            lsys_series = lsys_map.get(aid, [])
+            l_hat = (
+                (sum(lsys_series) / len(lsys_series))
+                if lsys_series else (lambda_hat * w_hat)
+            )
+
+            # --- Wq̂: keep LB waiting times ---
             lb_waits = list(results_analyzer.get_lb_waiting_times())
             wq_hat = (sum(lb_waits) / len(lb_waits)) if lb_waits else 0.0
+
+            # --- Lq̂: mean LQ_LB timeseries (fallback λ̂·Wq̂) ---
+            lq_map = results_analyzer.get_metric_map(SampledMetricName.LQ_LB)
+            lb_id = lb.id if lb is not None else None
+            lq_series = lq_map.get(lb_id, []) if lb_id is not None else []
+            lq_hat = (
+                (sum(lq_series) / len(lq_series))
+                if lq_series else (lambda_hat * wq_hat)
+            )
+
+            # ---- rho  mean SERVER_UTILIZATION timeseries (fallback λ̂/(cμ̂)) ---
+            util_map = (
+                results_analyzer.get_metric_map(SampledMetricName.SERVER_UTILIZATION)
+            )
+            total_busy = 0.0
+            total_samples = 0
+            for series in util_map.values():
+                total_busy += float(sum(series))
+                total_samples += len(series)
+            rho_hat = (total_busy / total_samples) if total_samples > 0 else (
+                lambda_hat / (server_count * mu_hat)
+                if mu_hat not in (0.0, float("inf")) else 0.0
+        )
+
         else:
             arrays_map = results_analyzer.get_server_event_arrays()
             wait_sum = 0.0
@@ -506,12 +543,12 @@ def _observed_kpis(
                 wait_count += len(vals)
             wq_hat = (wait_sum / wait_count) if wait_count > 0 else 0.0
 
-        l_hat = lambda_hat * w_hat
-        lq_hat = lambda_hat * wq_hat
-        rho_hat = (
-            lambda_hat / (server_count * mu_hat)
-            if mu_hat not in (0.0, float("inf")) else 0.0
-        )
+            l_hat = lambda_hat * w_hat
+            lq_hat = lambda_hat * wq_hat
+            rho_hat = (
+                lambda_hat / (server_count * mu_hat)
+                if mu_hat not in (0.0, float("inf")) else 0.0
+            )
 
         return MMcResults(
             lambda_rate=lambda_hat,

src/asyncflow/runtime/actors/arrivals_generator.py

@@ -9,7 +9,7 @@
 
 import numpy as np
 
-from asyncflow.config.enums import SystemNodes
+from asyncflow.config.enums import SampledMetricName, SystemNodes
 from asyncflow.metrics.client import RqsClock
 from asyncflow.runtime.rqs_state import RequestState
 from asyncflow.samplers.arrivals import general_interarrivals
@@ -20,7 +20,6 @@
 
     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
@@ -71,6 +70,7 @@ def __init__( # noqa: PLR0913
         self._l_system: int = 0
         # dict for the collector to have the time series
         self.enabled_metrics: dict[SampledMetricName, list[float]] = {}
+        self.enabled_metrics[SampledMetricName.L_SYSTEM] = []
 
 
     def _next_id(self) -> int:

src/asyncflow/runtime/actors/load_balancer.py

@@ -65,10 +65,14 @@ 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]] = {}
+        # initialize list
+        if self.lb_config.algorithms == LbAlgorithmsName.FCFS:
+            self.enabled_metrics[SampledMetricName.LQ_LB] = []
+
+        # track if LB is holding one request while waiting for a free edge
+        self._holding_waiting: bool = False
 
 
     # Helpers FCFS
@@ -104,7 +108,6 @@ 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]
@@ -119,6 +122,9 @@ def _forwarder(self) -> Generator[simpy.Event, None, None]:
                         self.env.now,
                     )
 
+                # We're now holding one request while waiting for a free edge
+                self._holding_waiting = True
+
             # The idea is the following: when a request arrives and the algorithm
             # is FCFS, we maintain a FIFO of available edges. If an edge connected
             # to a server is ready, the loop continues and (assuming no event injection
@@ -144,7 +150,8 @@ 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
+                self._holding_waiting = False
+
                 edge_rt.transport(state)
             else:
                 state.record_hop(
@@ -167,5 +174,8 @@ def lb_waiting_times(self) -> Sequence[float]:
 
     @property
     def lq_lb(self) -> int:
-        """Readable version to sample the metric"""
-        return self._lq_lb
+        """
+        FCFS queue length at the LB: items still in lb_box plus the one
+        currently held by the LB while waiting for a free edge.
+        """
+        return len(self.lb_box.items) + (1 if self._holding_waiting else 0)