Logic to handle completed rqs move to arr-gen

Author: GioeleB00

asyncflow_queue_limit/asyncflow_mmc.ipynb

@@ -20,7 +20,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 96,
+   "execution_count": 111,
    "id": "3e168d4a",
    "metadata": {},
    "outputs": [],
@@ -45,7 +45,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 97,
+   "execution_count": 112,
    "id": "dd39a8e3",
    "metadata": {},
    "outputs": [
@@ -112,7 +112,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 98,
+   "execution_count": 113,
    "id": "d2937e5e",
    "metadata": {},
    "outputs": [],
@@ -201,7 +201,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 99,
+   "execution_count": 114,
    "id": "d0634bc8",
    "metadata": {},
    "outputs": [
@@ -339,27 +339,27 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 100,
+   "execution_count": 115,
    "id": "ccd7379b",
    "metadata": {},
    "outputs": [
     {
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "=================================================================\n",
+      "===================================================================\n",
       "MMc (FCFS/Erlang-C) — Theory vs Observed\n",
-      "-----------------------------------------------------------------\n",
-      "sym  metric                    theory    observed       abs  rel%\n",
-      "-----------------------------------------------------------------\n",
-      "λ    Arrival rate (1/s)    270.000000  270.258333  0.258333  0.10\n",
-      "μ    Service rate (1/s)    100.000000  100.036707  0.036707  0.04\n",
-      "rho  Utilization             0.900000    0.900531  0.000531  0.06\n",
-      "L    Mean items in sys      10.053549   10.073544  0.019994  0.20\n",
-      "Lq   Mean items in queue     7.353549    7.371934  0.018385  0.25\n",
-      "W    Mean time in sys (s)    0.037235    0.037274  0.000038  0.10\n",
-      "Wq   Mean waiting (s)        0.027235    0.027277  0.000042  0.15\n",
-      "=================================================================\n"
+      "-------------------------------------------------------------------\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",
+      "===================================================================\n"
      ]
     }
    ],

src/asyncflow/metrics/collector.py

@@ -67,9 +67,9 @@ def _build_time_series(self) -> Generator[simpy.Event, None, None]:
 
 
 
-    def start(self) -> simpy.Process:
+    def start(self) -> None:
         """Definition of the process to collect sampled metrics"""
-        return self.env.process(self._build_time_series())
+        self.env.process(self._build_time_series())
 
 
 

src/asyncflow/metrics/simulation_analyzer.py

@@ -28,7 +28,7 @@
     from matplotlib.axes import Axes
     from matplotlib.lines import Line2D
 
-    from asyncflow.runtime.actors.client import ClientRuntime
+    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
@@ -69,14 +69,14 @@ class ResultsAnalyzer:
     def __init__(
         self,
         *,
-        client: ClientRuntime,
+        generator: ArrivalsGeneratorRuntime,
         servers: list[ServerRuntime],
         edges: list[EdgeRuntime],
         settings: SimulationSettings,
         lb: LoadBalancerRuntime | None = None,
     ) -> None:
         """Initialize with the runtime objects and original settings."""
-        self._client = client
+        self._generator = generator
         self._servers = servers
         self._edges = edges
         self._settings = settings
@@ -103,8 +103,8 @@ def __init__(
     # ─────────────────────────────────────────────
     def process_all_metrics(self) -> None:
         """Compute all aggregated and sampled metrics if not already done."""
-        # Client-side: end-to-end latencies + 1s throughput
-        if self.latency_stats is None and self._client.rqs_clock:
+        # generator-side: end-to-end latencies + 1s throughput
+        if self.latency_stats is None and self._generator.rqs_clock:
             self._process_event_metrics()
 
         # Sampled time series from servers/edges (RAM, queues, etc.)
@@ -209,7 +209,7 @@ def _process_event_metrics(self) -> None:
         # 1) Latencies
         self.latencies = [
             clock.finish - clock.start
-            for clock in self._client.rqs_clock
+            for clock in self._generator.rqs_clock
         ]
 
         # 2) Summary stats
@@ -229,7 +229,7 @@ def _process_event_metrics(self) -> None:
             self.latency_stats = {}
 
         # 3) Throughput per 1s window (cached)
-        completion_times = sorted(clock.finish for clock in self._client.rqs_clock)
+        completion_times = sorted(clock.finish for clock in self._generator.rqs_clock)
         end_time = self._settings.total_simulation_time
 
         timestamps: list[float] = []
@@ -320,7 +320,7 @@ def get_throughput_series(
             return self.throughput_series or ([], [])
 
         # Recompute with a custom window size.
-        completion_times = sorted(clock.finish for clock in self._client.rqs_clock)
+        completion_times = sorted(clock.finish for clock in self._generator.rqs_clock)
         end_time = self._settings.total_simulation_time
 
         timestamps: list[float] = []

src/asyncflow/runner/simulation.py

@@ -44,7 +44,7 @@
 class Startable(Protocol):
     """A protocol for runtime actors that can be started."""
 
-    def start(self) -> simpy.Process:
+    def start(self) -> None:
         """Starts the main process loop for the actor."""
         ...
 
@@ -141,6 +141,9 @@ def _build_rqs_generator(self) -> None:
             arrivals=self.arrivals,
             sim_settings=self.simulation_settings,
             rng=self.rng,
+            arrivals_generator_box=self._make_inbox(),
+            completed_box=self._make_inbox(),
+
         )
 
 
@@ -155,7 +158,6 @@ def _build_client(self) -> None:
         self._client_runtime[self.client.id] = ClientRuntime(
             env=self.env,
             out_edge=None,
-            completed_box=self._make_inbox(),
             client_box=self._make_inbox(),
             client_config=self.client,
         )
@@ -227,6 +229,8 @@ def _build_edges(self) -> None:
                 target_box = target_object.client_box
             elif isinstance(target_object, LoadBalancerRuntime):
                 target_box = target_object.lb_box
+            elif isinstance(target_object, ArrivalsGeneratorRuntime):
+                target_box = target_object.arrivals_generator_box
 
 
             else:
@@ -404,7 +408,7 @@ def run(self) -> ResultsAnalyzer:
         self.env.run(until=self.simulation_settings.total_simulation_time)
 
         return ResultsAnalyzer(
-            client=next(iter(self._client_runtime.values())),
+            generator=next(iter(self._arrivals_runtime.values())),
             servers=list(self._servers_runtime.values()),
             edges=list(self._edges_runtime.values()),
             settings=self.simulation_settings,

src/asyncflow/runtime/actors/arrivals_generator.py

@@ -10,6 +10,7 @@
 import numpy as np
 
 from asyncflow.config.enums import SystemNodes
+from asyncflow.metrics.client import RqsClock
 from asyncflow.runtime.rqs_state import RequestState
 from asyncflow.samplers.arrivals import general_interarrivals
 
@@ -25,18 +26,20 @@
 
 class ArrivalsGeneratorRuntime:
     """
-    A “node” that produces request contexts at stochastic inter-arrival times
+    A node that produces request contexts at stochastic inter-arrival times
     and immediately pushes them down the pipeline via an EdgeRuntime.
     """
 
-    def __init__(
+    def __init__( # noqa: PLR0913
         self,
         *,
         env: simpy.Environment,
         out_edge: EdgeRuntime | None,
         arrivals: ArrivalsGenerator,
         sim_settings: SimulationSettings,
         rng: np.random.Generator | None = None,
+        arrivals_generator_box: simpy.Store,
+        completed_box: simpy.Store,
         ) -> None:
         """
         Definition of the instance attributes for the ArrivalsGeneratorRuntime
@@ -47,14 +50,19 @@ def __init__(
             arrivals (ArrivalsGenerator): data do define the sampler
             sim_settings (SimulationSettings): settings to start the simulation
             rng (np.random.Generator | None, optional): random variable generator.
+            arrivals_generator_box: simpy box to collect request when they come back
+            completed_box: box to collect all satisfied requests
 
         """
         self.arrivals = arrivals
         self.sim_settings = sim_settings
         self.rng =  rng or np.random.default_rng()
         self.out_edge = out_edge
         self.env = env
+        self.arrivals_generator_box = arrivals_generator_box
+        self.completed_box = completed_box
         self.id_counter = 0
+        self._rqs_clock: list[RqsClock] = []
 
 
     def _next_id(self) -> int:
@@ -90,6 +98,28 @@ def _event_arrival(self) -> Generator[simpy.Event, None, None]:
             # from one node to another
             self.out_edge.transport(state)
 
+    def _collector(self) -> Generator[simpy.Event, None, None]:
+        """The request has been satisfied"""
+        while True:
+            state: RequestState = yield self.arrivals_generator_box.get()  # type: ignore[assignment]
+
+            state.finish_time = self.env.now
+            clock_data = RqsClock(
+                start=state.initial_time,
+                finish=state.finish_time,
+            )
+            self._rqs_clock.append(clock_data)
+            yield self.completed_box.put(state)
+
+
     def start(self) -> simpy.Process:
         """Passing the structure as a simpy process"""
-        return self.env.process(self._event_arrival())
+        self.env.process(self._event_arrival())
+        self.env.process(self._collector())
+
+
+
+    @property
+    def rqs_clock(self) -> list[RqsClock]:
+        """Readable version to compute aggregate metrics"""
+        return self._rqs_clock

src/asyncflow/runtime/actors/client.py

@@ -6,7 +6,6 @@
 import simpy
 
 from asyncflow.config.enums import SystemNodes
-from asyncflow.metrics.client import RqsClock
 from asyncflow.runtime.actors.edge import EdgeRuntime
 from asyncflow.schemas.topology.nodes import Client
 
@@ -24,20 +23,13 @@ def __init__(
         env: simpy.Environment,
         out_edge: EdgeRuntime | None,
         client_box: simpy.Store,
-        completed_box: simpy.Store,
         client_config: Client,
         ) -> None:
         """Definition of attributes for the client"""
         self.env = env
         self.out_edge = out_edge
         self.client_config = client_config
         self.client_box = client_box
-        self.completed_box = completed_box
-        # This list will be enough to calculate at the end
-        # of the simulation both the throughput and the
-        # latency distribution
-
-        self._rqs_clock: list[RqsClock] = []
 
 
     def _forwarder(self) -> Generator[simpy.Event, None, None]:
@@ -53,31 +45,9 @@ def _forwarder(self) -> Generator[simpy.Event, None, None]:
                     self.env.now,
                 )
 
-            # if the length of the list is bigger than two
-            # it means that the state is coming back to the
-            # client after being elaborated, since if the value
-            # would be equal to two would mean that the state
-            # went through the mandatory path to be generated
-            # rqs generator and client registration
-            if len(state.history) > 3:
-                state.finish_time = self.env.now
-                clock_data = RqsClock(
-                    start=state.initial_time,
-                    finish=state.finish_time,
-                )
-                self._rqs_clock.append(clock_data)
-                yield self.completed_box.put(state)
-            else:
-                self.out_edge.transport(state)
+            self.out_edge.transport(state)
 
-    def start(self) -> simpy.Process:
+    def start(self) -> None:
         """Initialization of the process"""
-        return self.env.process(self._forwarder())
+        self.env.process(self._forwarder())
 
-    @property
-    def rqs_clock(self) -> list[RqsClock]:
-        """
-        Expose the value of the private list of the starting
-        and arrival time for each rqs just for reading purpose
-        """
-        return self._rqs_clock

src/asyncflow/runtime/actors/load_balancer.py

@@ -149,10 +149,10 @@ 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) -> simpy.Process:
+    def start(self) -> None:
         """Start the process and populate FIFO"""
         self._prime_free_edges()
-        return self.env.process(self._forwarder())
+        self.env.process(self._forwarder())
 
     @property
     def lb_waiting_times(self) -> Sequence[float]:

src/asyncflow/runtime/actors/server.py

@@ -189,9 +189,9 @@ 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) -> simpy.Process:
+    def start(self) -> None:
         """Generate the process to simulate the server inside simpy env"""
-        return self.env.process(self._dispatcher())
+        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

src/asyncflow/runtime/events/injection.py

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