defined client runtime + tests

Author: GioeleB00

src/app/config/rqs_state.py

@@ -1,50 +1,51 @@
-"""
-defining a state in a one to one correspondence
-with the requests generated that will go through
-all the node necessary to accomplish the user request
-"""
+"""Data structures representing the life-cycle of a single request."""
 
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, NamedTuple
 
+if TYPE_CHECKING:
+    from app.config.constants import SystemEdges, SystemNodes
 
-@dataclass
-class RequestState:
-    """
-    State object carried by each request through the simulation.
-
-    Attributes:
-        id: Unique identifier of the request.
-        t0: Timestamp (simulated env.now) when the request was generated.
-        history: List of hop records, each noting a node/edge visit.
-        finish_time: Timestamp when the requests is satisfied
 
-    """
+class Hop(NamedTuple):
+    """A single traversal of a node or edge."""
 
-    id: int                                # Unique request identifier
-    initial_time: float                    # Generation timestamp (env.now)
-    finish_time: float | None = None       # a requests might be dropped
-    history: list[str] = field(default_factory=list)  # Trace of hops
+    component_type: SystemNodes | SystemEdges
+    component_id: str
+    timestamp: float
 
-    def record_hop(self, node_name: str, now: float) -> None:
-        """
-        Append a record of visiting a node or edge.
 
-        Args:
-            node_name: Name of the node or edge being recorded.
-            now: register the time of the operation
-
-        """
-        # Record hop as "NodeName@Timestamp"
-        self.history.append(f"{node_name}@{now:.3f}")
+@dataclass
+class RequestState:
+    """Mutable state carried by each request throughout the simulation."""
+
+    id: int
+    initial_time: float
+    finish_time: float | None = None
+    history: list[Hop] = field(default_factory=list)
+
+    # ------------------------------------------------------------------ #
+    # API                                                                 #
+    # ------------------------------------------------------------------ #
+
+    def record_hop(
+        self,
+        component_type: SystemNodes | SystemEdges,
+        component_id: str,
+        now: float,
+    ) -> None:
+        """Append a new hop in chronological order."""
+        self.history.append(Hop(component_type, component_id, now))
+
+    # ------------------------------------------------------------------ #
+    # Derived metrics                                                     #
+    # ------------------------------------------------------------------ #
 
     @property
     def latency(self) -> float | None:
-        """
-        Return the total time in the system (finish_time - initial_time),
-        or None if the request hasn't completed yet.
-        """
+        """Total time inside the system or ``None`` if not yet completed."""
         if self.finish_time is None:
             return None
         return self.finish_time - self.initial_time

src/app/core/runtime/client.py

@@ -0,0 +1,59 @@
+"""defining the object client for the simulation"""
+
+from collections.abc import Generator
+from typing import TYPE_CHECKING
+
+import simpy
+
+from app.config.constants import SystemNodes
+from app.core.runtime.edge import EdgeRuntime
+from app.schemas.system_topology_schema.full_system_topology_schema import Client
+
+if TYPE_CHECKING:
+    from app.config.rqs_state import RequestState
+
+
+
+class ClientRuntime:
+    """class to define the client runtime"""
+
+    def __init__(
+        self,
+        env: simpy.Environment,
+        out_edge: EdgeRuntime,
+        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
+
+
+    def _forwarder(self) -> Generator[simpy.Event, None, None]:
+        """Updtate the state before passing it to another node"""
+        while True:
+            state: RequestState = yield self.client_box.get()  # type: ignore[assignment]
+
+            state.record_hop(
+                    SystemNodes.CLIENT,
+                    self.client_config.id,
+                    self.env.now,
+                )
+
+            # by checking the previous node (-2 the previous component is an edge)
+            # we are able to understand if the request should be elaborated
+            # when the type is Generator, or the request is completed, in this case
+            # the client is the target and the previous node type is not a rqs generator
+            if state.history[-2].component_type != SystemNodes.GENERATOR:
+                state.finish_time = self.env.now
+                yield self.completed_box.put(state)
+            else:
+                self.out_edge.transport(state)
+
+    def client_run(self) -> simpy.Process:
+        """Initialization of the process"""
+        return self.env.process(self._forwarder())

src/app/core/runtime/edge.py

@@ -12,6 +12,7 @@
 import numpy as np
 import simpy
 
+from app.config.constants import SystemEdges
 from app.config.rqs_state import RequestState
 from app.core.event_samplers.common_helpers import general_sampler
 from app.schemas.system_topology_schema.full_system_topology_schema import Edge
@@ -46,12 +47,20 @@ def _deliver(self, state: RequestState) -> Generator[simpy.Event, None, None]:
         uniform_variable = self.rng.uniform()
         if uniform_variable < self.edge_config.dropout_rate:
             state.finish_time = self.env.now
-            state.record_hop(f"{self.edge_config.id}-dropped", state.finish_time)
+            state.record_hop(
+                SystemEdges.NETWORK_CONNECTION,
+                f"{self.edge_config.id}-dropped",
+                state.finish_time,
+            )
             return
 
         transit_time = general_sampler(random_variable, self.rng)
         yield self.env.timeout(transit_time)
-        state.record_hop(self.edge_config.id, self.env.now)
+        state.record_hop(
+            SystemEdges.NETWORK_CONNECTION,
+            self.edge_config.id,
+            self.env.now,
+            )
         yield self.target_box.put(state)
 
 

src/app/core/runtime/rqs_generator.py

@@ -106,7 +106,11 @@ def _event_arrival(self) -> Generator[simpy.Event, None, None]:
                 initial_time=self.env.now,
 
             )
-            state.record_hop(SystemNodes.GENERATOR, self.env.now)
+            state.record_hop(
+                SystemNodes.GENERATOR,
+                self.rqs_generator_data.id,
+                self.env.now,
+            )
             # transport is a method of the edge runtime
             # which define the step of how the state is moving
             # from one node to another

src/app/core/simulation_run.py

@@ -19,7 +19,7 @@
 
 
 
-
+### TO MODIFY EVERYTHING WORK IN PROGRESS
 
 def run_simulation(
     input_data: SimulationPayload,

src/app/schemas/requests_generator_input.py

@@ -3,13 +3,15 @@
 
 from pydantic import BaseModel, Field, field_validator
 
-from app.config.constants import Distribution, TimeDefaults
+from app.config.constants import Distribution, SystemNodes, TimeDefaults
 from app.schemas.random_variables_config import RVConfig
 
 
 class RqsGeneratorInput(BaseModel):
     """Define the expected variables for the simulation"""
 
+    id: str
+    type: SystemNodes = SystemNodes.GENERATOR
     avg_active_users: RVConfig
     avg_request_per_minute_per_user: RVConfig
 

tests/conftest.py

@@ -153,14 +153,14 @@ def rng() -> NpGenerator:
     return default_rng(0)
 
 
-# ---------------------------------------------------------------------------
-# Metrics sets
-# ---------------------------------------------------------------------------
+# --------------------------------------------------------------------------- #
+# Metric sets                                                                 #
+# --------------------------------------------------------------------------- #
 
 
 @pytest.fixture(scope="session")
 def enabled_sample_metrics() -> set[SampledMetricName]:
-    """Default sample-level KPIs tracked in most tests."""
+    """Default time-series KPIs collected in most tests."""
     return {
         SampledMetricName.READY_QUEUE_LEN,
         SampledMetricName.RAM_IN_USE,
@@ -169,59 +169,75 @@ def enabled_sample_metrics() -> set[SampledMetricName]:
 
 @pytest.fixture(scope="session")
 def enabled_event_metrics() -> set[EventMetricName]:
-    """Default event-level KPIs tracked in most tests."""
-    return {EventMetricName.RQS_LATENCY}
+    """Default per-event KPIs collected in most tests."""
+    return {
+        EventMetricName.RQS_LATENCY,
+    }
 
 
-# ---------------------------------------------------------------------------
-# Global simulation settings
-# ---------------------------------------------------------------------------
+# --------------------------------------------------------------------------- #
+# Global simulation settings                                                  #
+# --------------------------------------------------------------------------- #
 
 
 @pytest.fixture
 def sim_settings(
     enabled_sample_metrics: set[SampledMetricName],
     enabled_event_metrics: set[EventMetricName],
 ) -> SimulationSettings:
-    """A minimal `SimulationSettings` instance for unit tests."""
+    """
+    Minimal :class:`SimulationSettings` instance.
+
+    The simulation horizon is fixed to the lowest allowed value so that unit
+    tests run quickly.
+    """
     return SimulationSettings(
         total_simulation_time=TimeDefaults.MIN_SIMULATION_TIME,
         enabled_sample_metrics=enabled_sample_metrics,
         enabled_event_metrics=enabled_event_metrics,
     )
 
 
-# ---------------------------------------------------------------------------
-# Traffic profile
-# ---------------------------------------------------------------------------
+# --------------------------------------------------------------------------- #
+# Traffic profile                                                             #
+# --------------------------------------------------------------------------- #
 
 
 @pytest.fixture
 def rqs_input() -> RqsGeneratorInput:
-    """`RqsGeneratorInput` with 1 user and 2 req/min for quick tests."""
+    """
+    One active user issuing two requests per minute—sufficient to
+    exercise the entire request-generator pipeline with minimal overhead.
+    """
     return RqsGeneratorInput(
+        id= "rqs-1",
         avg_active_users=RVConfig(mean=1.0),
         avg_request_per_minute_per_user=RVConfig(mean=2.0),
         user_sampling_window=TimeDefaults.USER_SAMPLING_WINDOW,
     )
 
 
-# ---------------------------------------------------------------------------
-# Minimal topology (one client, no servers, no edges)
-# ---------------------------------------------------------------------------
+# --------------------------------------------------------------------------- #
+# Minimal topology (one client, no servers, no edges)                         #
+# --------------------------------------------------------------------------- #
 
 
 @pytest.fixture
 def topology_minimal() -> TopologyGraph:
-    """Valid topology with a single client and zero servers/edges."""
+    """
+    A valid topology containing a single client and **no** servers or edges.
+
+    Suitable for low-level tests that do not need to traverse the server
+    layer or network graph.
+    """
     client = Client(id="client-1")
     nodes = TopologyNodes(servers=[], client=client)
     return TopologyGraph(nodes=nodes, edges=[])
 
 
-# ---------------------------------------------------------------------------
-# Full simulation payload
-# ---------------------------------------------------------------------------
+# --------------------------------------------------------------------------- #
+# Complete simulation payload                                                 #
+# --------------------------------------------------------------------------- #
 
 
 @pytest.fixture
@@ -230,7 +246,12 @@ def payload_base(
     sim_settings: SimulationSettings,
     topology_minimal: TopologyGraph,
 ) -> SimulationPayload:
-    """End-to-end payload used by high-level simulation tests."""
+    """
+    End-to-end payload used by integration tests and FastAPI endpoint tests.
+
+    It wires together the individual fixtures into the single object expected
+    by the simulation engine.
+    """
     return SimulationPayload(
         rqs_input=rqs_input,
         topology_graph=topology_minimal,