Implemented fcfs with tests

Author: GioeleB00

asyncflow_queue_limit/asyncflow_mm1.ipynb

@@ -5,12 +5,12 @@
    "id": "870337dc",
    "metadata": {},
    "source": [
-    "# AsyncFlow — MMc Theory vs Simulation (Guided Notebook)\n",
+    "# AsyncFlow — MMc split Theory vs Simulation (Guided Notebook)\n",
     "\n",
     "This notebook shows how to:\n",
     "\n",
     "1. Make imports work inside a notebook (src-layout or package install)\n",
-    "2. Build a **multi-server** scenario compatible with **M/M/c** assumptions\n",
+    "2. Build a **multi-server** scenario compatible with **M/M/c** assumptions in the case of n parallel M/M/1\n",
     "3. Run the simulation and collect results\n",
     "4. Compare theory vs observed KPIs (pretty-printed table)\n",
     "5. Plot the standard dashboards (latency, throughput, server time series)\n",
@@ -84,15 +84,11 @@
     "  Service times follow an **Exponential** distribution with mean \\$E\\[S]\\$ (service rate \\$\\mu = 1/E\\[S]\\$). No RAM/IO steps are included in the pipeline.\n",
     "\n",
     "* **Load balancer with round-robin dispatch**\n",
-    "  A **single load balancer** is required when \\$c > 1\\$. It splits arrivals **round-robin** across servers, so each server has its own local queue.\n",
+    "  A **single load balancer** is required when \\$c > 1\\$. It splits arrivals **randomly** across servers, so each server has its own local queue.\n",
     "  This corresponds to a **split M/M/c** model, not the textbook pooled queue.\n",
     "\n",
-    "* **Deterministic, very small network latency**\n",
-    "  All edges have **fixed latency** \\$\\ll 1,\\mathrm{ms}\\$. Queueing behavior is therefore dominated by CPU service, closely matching textbook assumptions.\n",
-    "\n",
     "* **“Poisson arrivals” via the generator**\n",
-    "  Arrivals are produced by the same **two-stage, windowed Poisson sampler**: in each user-sampling window \\$\\Delta\\$, we draw the active users \\$U\\$ (Poisson or Normal, per config).\n",
-    "  Within the window, arrivals are a **homogeneous Poisson process** with rate \\$\\Lambda = U \\cdot \\lambda\\_r/60\\$.\n",
+    "  \n",
     "\n",
     "  \n",
     "\n",
@@ -234,7 +230,7 @@
    "id": "da98b8b9",
    "metadata": {},
    "source": [
-    "## 3) MMc (Round-Robin) — theory vs observed comparison\n",
+    "## 3) MMc (Random) — theory vs observed comparison\n",
     "\n",
     "If the payload violates MMc assumptions, a readable error is shown instead.\n",
     "This section matches exactly what the analyzer computes **now**: a **split random model** (not the pooled FCFS/Erlang-C model). When we add **FCFS** in the LB, we’ll also expose the textbook Erlang-C formulas side-by-side.\n",
@@ -366,12 +362,7 @@
     "\n",
     "### Why small deltas appear\n",
     "\n",
-    "* **Windowed arrivals:** The generator uses a **windowed** active-user sampler; across windows the rate is piecewise-constant (Cox process), not one global Poisson.\n",
-    "* **Finite horizon / warm-up:** Short runs and lack of warm-up bias early windows.\n",
-    "* **Network latency:** Small but non-zero link delays add a constant component to $W$.\n",
-    "\n",
-    "\n",
-    "Increasing the run length, keeping edges tiny and deterministic, and ensuring identical servers will usually shrink Theory–Observed gaps. "
+    "* **Finite horizon / warm-up:** Short runs and lack of warm-up bias early windows.\n"
    ]
   },
   {

asyncflow_queue_limit/asyncflow_mmc.ipynb

@@ -134,6 +134,7 @@ class LbAlgorithmsName(StrEnum):
   ROUND_ROBIN = "round_robin"
   LEAST_CONNECTIONS = "least_connection"
   RANDOM = "random"
+  FCFS = "fcfs"
 
 
 # ======================================================================

asyncflow_queue_limit/asyncflow_mmc_split.ipynb

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import math
 import sys
 from typing import TYPE_CHECKING, Literal, TextIO, TypedDict, cast
 from weakref import WeakSet
@@ -127,8 +128,8 @@ def _check_topology(self, payload: SimulationPayload) -> list[str]:
                 errs.append("for c=1 the load balancer must be absent.")
         elif lb is None:
             errs.append("for c>1 a load balancer is required.")
-        elif lb.algorithms != LbAlgorithmsName.RANDOM:
-            errs.append("only random is supported for the split M/M/c model.")
+        elif lb.algorithms not in {LbAlgorithmsName.RANDOM, LbAlgorithmsName.FCFS}:
+             errs.append("supported lb algorithms: RANDOM (split) or FCFS (pooled).")
 
         return errs
 
@@ -350,11 +351,14 @@ def _build_observed_params(
         )
 
     # ────────────────────────────────────────────────────────────────────
-    # Closed form (split RR): theory → MMcResults
+    # Closed form (Random): theory → MMcResults split model ( n parallel mm1)
     # ────────────────────────────────────────────────────────────────────
 
     def _theoretical_kpis_split(self, payload: SimulationPayload) -> MMcResults:
-        """Closed forms for RR split: λ_i=λ/c; Wq=rho/(μ-λ_i); W=1/μ+Wq; Lq=λWq; L=λW"""
+        """
+        Closed forms for Random split: λ_i=λ/c;
+        Wq=rho/(μ-λ_i); W=1/μ+Wq; Lq=λWq; L=λW
+        """
         self.validate_or_raise(payload)
         params = self._build_params(payload)
 
@@ -394,10 +398,68 @@ def _theoretical_kpis_split(self, payload: SimulationPayload) -> MMcResults:
             Wq=wq,
         )
 
+
+    def _theoretical_mmc_erlang_c_kpis(
+        self,
+        lambda_rate: float,
+        mu_rate: float,
+        c: int,
+    ) -> MMcResults:
+        """Closed forms for pooled M/M/c (FCFS, Erlang-C)."""
+        rho = self._rho_from(lambda_rate, c, mu_rate)
+        if rho >= 1.0:
+            inf = float("inf")
+            return MMcResults(
+                lambda_rate=lambda_rate,
+                mu_rate=mu_rate,
+                c=c,
+                rho=rho,
+                L=inf, Lq=inf, W=inf, Wq=inf,
+            )
+
+        a = lambda_rate / mu_rate  # offered traffic
+        # P0
+        s = sum((a**n) / math.factorial(n) for n in range(c))
+        tail = (a**c) / math.factorial(c) * (1.0 / (1.0 - rho))
+        p0 = 1.0 / (s + tail)
+        pw = ((a**c) / math.factorial(c)) * (1.0 / (1.0 - rho)) * p0
+        lq = pw * (rho / (1.0 - rho))
+        wq = lq / lambda_rate
+        w = wq + 1.0 / mu_rate
+        lam = lambda_rate * w
+
+        return MMcResults(
+            lambda_rate=lambda_rate,
+            mu_rate=mu_rate,
+            c=c,
+            rho=rho,
+            L=lam,
+            Lq=lq,
+            W=w,
+            Wq=wq,
+        )
+
+    def _theoretical_kpis_pooled(self, payload: SimulationPayload) -> MMcResults:
+        """Closed forms for pooled M/M/c (LB=FCFS, central queue)."""
+        # riusa i builder che hai già
+        params = self._build_params(payload)
+        return self._theoretical_mmc_erlang_c_kpis(
+            lambda_rate=params["lambda_rate"],
+            mu_rate=params["mu_rate"],
+            c=params["c"],
+    )
+
+
     def evaluate(self, payload: SimulationPayload) -> MMcResults:
-        """Public entry-point: return closed-form KPIs for split RR."""
+        """Return closed-form KPIs: split (RR/RANDOM) or pooled (FCFS)."""
+        self.validate_or_raise(payload)
+        lb = payload.topology_graph.nodes.load_balancer
+        if (lb is not None) and (lb.algorithms == LbAlgorithmsName.FCFS):
+            return self._theoretical_kpis_pooled(payload)
+        # default: split (random/RR)
         return self._theoretical_kpis_split(payload)
 
+
     # ────────────────────────────────────────────────────────────────────
     # Observed KPIs → MMcResults (coerenti con definizioni sopra)
     # ────────────────────────────────────────────────────────────────────
@@ -483,7 +545,12 @@ def compare_against_run(
         """Build a table with theory vs observed and deltas."""
         self.validate_or_raise(payload)
 
-        theory = self._theoretical_kpis_split(payload)
+        lb = payload.topology_graph.nodes.load_balancer
+        if (lb is not None) and (lb.algorithms == LbAlgorithmsName.FCFS):
+            theory = self._theoretical_kpis_pooled(payload)
+        else:
+            theory = self._theoretical_kpis_split(payload)
+
         observed = self._observed_kpis(payload, results_analyzer)
 
         rows: list[MMcKPIRow] = []
@@ -514,16 +581,21 @@ def add(symbol: str, name: str, key: MMcResultKey) -> None:
 
         return rows
 
-
-    # --------------- PRETTY PRINT -------------------
-
     # ────────────────────────────────────────────────────────────────────
     # Pretty table (KPI):
     # ────────────────────────────────────────────────────────────────────
+    
+    def _title_for(self, payload: "SimulationPayload") -> str:
+        lb = payload.topology_graph.nodes.load_balancer
+        if lb is not None and lb.algorithms == LbAlgorithmsName.FCFS:
+            return "MMc (FCFS/Erlang-C) — Theory vs Observed"
+        # default to random split when no LB or non-FCFS
+        return "MMc (Random split) — Theory vs Observed"
+    
     @staticmethod
     def _format_kpi_table(
-        rows: list[MMcKPIRow],
-        title: str = "MMc (RR) — Theory vs Observed",
+        rows: list["MMcKPIRow"],
+        title: str = "MMc — Theory vs Observed",
     ) -> str:
         data = [
             (
@@ -536,12 +608,11 @@ def _format_kpi_table(
             )
             for r in rows
         ]
-
         headers = ("sym", "metric", "theory", "observed", "abs", "rel%")
         w_sym = max(len(headers[0]), *(len(d[0]) for d in data))
         w_met = max(len(headers[1]), *(len(d[1]) for d in data))
-        w_th = max(len(headers[2]), *(len(d[2]) for d in data))
-        w_ob = max(len(headers[3]), *(len(d[3]) for d in data))
+        w_th  = max(len(headers[2]), *(len(d[2]) for d in data))
+        w_ob  = max(len(headers[3]), *(len(d[3]) for d in data))
         w_abs = max(len(headers[4]), *(len(d[4]) for d in data))
         w_rel = max(len(headers[5]), *(len(d[5]) for d in data))
 
@@ -564,12 +635,12 @@ def _format_kpi_table(
 
     def compare_and_format(
         self,
-        payload: SimulationPayload,
-        results_analyzer: ResultsAnalyzer,
+        payload: "SimulationPayload",
+        results_analyzer: "ResultsAnalyzer",
     ) -> str:
-        """Return a formatted KPI table for theory vs observed."""
         rows = self.compare_against_run(payload, results_analyzer)
-        return self._format_kpi_table(rows)
+        title = self._title_for(payload)
+        return self._format_kpi_table(rows, title=title)
 
     def print_comparison(
         self,

src/asyncflow/config/enums.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from collections import OrderedDict
+from functools import partial
 from itertools import chain
 from pathlib import Path
 from types import MappingProxyType
@@ -12,6 +13,7 @@
 import simpy
 import yaml
 
+from asyncflow.config.enums import LbAlgorithmsName
 from asyncflow.metrics.collector import SampledMetricCollector
 from asyncflow.metrics.simulation_analyzer import ResultsAnalyzer
 from asyncflow.resources.registry import ResourcesRuntime
@@ -229,7 +231,9 @@ def _build_edges(self) -> None:
                 msg = f"Unknown runtime for {edge.target!r}"
                 raise TypeError(msg)
 
-
+            # prepare a dict of edges runtime with unique key as a tuple
+            # once all are ready we have to assign each one to the source node
+            # to allow the transport of the state through the edge
             self._edges_runtime[(edge.source, edge.target)] = (
                 EdgeRuntime(
                     env=self.env,
@@ -251,10 +255,31 @@ def _build_edges(self) -> None:
                     edge.source,
                     edge.target)
                 ]
+
+            # since multiple edges fan out from the LB we use a dict
+            # to have access in o(1) and assign the correct Edge runtime
             elif isinstance(source_object, LoadBalancerRuntime):
                 self._lb_out_edges[edge.id] = (
                     self._edges_runtime[(edge.source, edge.target)]
                 )
+                if isinstance(target_object, ServerRuntime) and (
+                    source_object.lb_config.algorithms == LbAlgorithmsName.FCFS
+                    ):
+                    # if the target is a server we pass the callback to comunicate
+                    # to the Lb that the server is free
+
+                    assert self._lb_runtime is not None
+                    lb_rt = self._lb_runtime
+                    edge_id = edge.id
+
+                    # We use functools.partial here to "pre-bind" the edge_id argument
+                    # of LoadBalancerRuntime.mark_free.
+                    # This turns it into a zero-argument
+                    # callable, so the ServerRuntime can simply
+                    # call notify_server_free()
+                    # when done, without needing to know its own edge_id.
+                    target_object.notify_server_free = partial(lb_rt.mark_free, edge_id)
+
             else:
                 msg =  f"Unknown runtime for {edge.source!r}"
                 raise TypeError(msg)

src/asyncflow/queue_theory_analysis/mmc.py

@@ -1,17 +1,20 @@
 """Definition of the node represented by the LB in the simulation"""
 
 
-from collections import OrderedDict
+from collections import OrderedDict, defaultdict
 from collections.abc import Generator
 from typing import (
     TYPE_CHECKING,
 )
 
 import simpy
 
-from asyncflow.config.enums import SystemNodes
+from asyncflow.config.enums import LbAlgorithmsName, SystemNodes
 from asyncflow.runtime.actors.edge import EdgeRuntime
-from asyncflow.runtime.actors.routing.lb_algorithms import LB_TABLE
+from asyncflow.runtime.actors.routing.lb_algorithms import (
+    LB_TABLE,
+    fcfs_picker,
+)
 from asyncflow.schemas.topology.nodes import LoadBalancer
 
 if TYPE_CHECKING:
@@ -55,7 +58,33 @@ def __init__(
         self.lb_out_edges = lb_out_edges
         self.lb_box = lb_box
 
-
+        # we need to keep track of the server that are busy
+        # right now we don't have multiprocess so each server has
+        # one core, to handle multiprocess in the feature we would
+        # have to add a new structure called capacity
+        self._busy: dict[str, int] = defaultdict(int)
+
+        # In the case of a FCFS algo or similar if no servers
+        # are available we need to wait before starting the algo
+        self._wait_ev: simpy.Event | None = None
+
+    # Helpers FCFS
+    # We manage here the logic for this algo because we need to be carefull
+    # to dont have collision with eventinjectionruntime, that's why we are
+    # not popping and rotating from the ordered dict with key edge_id and
+    # value edge runtime, but we manage adding an extra structure to know
+    # the state of the servers through the edges connecting them with the LB
+    def mark_busy(self, edge_id: str) -> None:
+        """Helper to manage the state of the available server"""
+        self._busy[edge_id] += 1
+
+    def mark_free(self, edge_id: str) -> None:
+        """Helper to manage the state of the available server"""
+        b = self._busy.get(edge_id, 0)
+        if b > 0:
+            self._busy[edge_id] = b - 1
+        if self._wait_ev is not None and not self._wait_ev.triggered:
+            self._wait_ev.succeed()
 
     def _forwarder(self) -> Generator[simpy.Event, None, None]:
         """Updtate the state before passing it to another node"""
@@ -68,8 +97,24 @@ def _forwarder(self) -> Generator[simpy.Event, None, None]:
                     self.env.now,
                 )
 
-            out_edge = LB_TABLE[self.lb_config.algorithms](self.lb_out_edges)
-            out_edge.transport(state)
+            if self.lb_config.algorithms == LbAlgorithmsName.FCFS:
+                # FCFS: wait for an edge to be free
+                pick = fcfs_picker(self.lb_out_edges, self._busy)
+                while pick is None:
+                    if self._wait_ev is None or self._wait_ev.triggered:
+                        self._wait_ev = self.env.event()
+                    yield self._wait_ev
+                    pick = fcfs_picker(self.lb_out_edges, self._busy)
+
+                edge_id, edge_rt = pick
+                # mark the server as occupied
+                self.mark_busy(edge_id)
+                # transport the request
+                edge_rt.transport(state)
+            else:
+                # Algo different from FCFS
+                edge_rt = LB_TABLE[self.lb_config.algorithms](self.lb_out_edges)
+                edge_rt.transport(state)
 
     def start(self) -> simpy.Process:
         """Initialization of the simpy process for the LB"""