Merge branch '205-rename-protofiles-to-inputfiles' of github.com:matsim-vsp/parallel_qsim_rust into 205-rename-protofiles-to-inputfiles

Author: frievoe97

rust_qsim/assets/equil/equil-1-plan.xml

@@ -22,7 +22,7 @@
                     <attribute name="routingMode" class="java.lang.String">car</attribute>
                 </attributes>
                 <route type="links" start_link="1" end_link="20" trav_time="undefined" distance="25000.0"
-                       vehicleRefId="1_car">1 6 15 20
+                       vehicleRefId="1">1 6 15 20
                 </route>
             </leg>
             <activity type="w" link="20" x="3456.0" y="4242.0" max_dur="00:10:00">
@@ -32,7 +32,7 @@
                     <attribute name="routingMode" class="java.lang.String">car</attribute>
                 </attributes>
                 <route type="links" start_link="20" end_link="20" trav_time="undefined" distance="0.0"
-                       vehicleRefId="1_car">20
+                       vehicleRefId="1">20
                 </route>
             </leg>
             <activity type="w" link="20" x="10000.0" y="0.0" max_dur="03:30:00">
@@ -42,7 +42,7 @@
                     <attribute name="routingMode" class="java.lang.String">car</attribute>
                 </attributes>
                 <route type="links" start_link="20" end_link="1" trav_time="undefined" distance="65000.0"
-                       vehicleRefId="1_car">20 21 22 23 1
+                       vehicleRefId="1">20 21 22 23 1
                 </route>
             </leg>
             <activity type="h" link="1" x="-25000.0" y="0.0">

rust_qsim/assets/equil/equil-vehicles.xml

@@ -1,9 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<vehicleDefinitions xmlns="http://www.matsim.org/files/dtd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-                    xsi:schemaLocation="http://www.matsim.org/files/dtd http://www.matsim.org/files/dtd/vehicleDefinitions_v1.0.xsd">
+<vehicleDefinitions xmlns="http://www.matsim.org/files/dtd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.matsim.org/files/dtd http://www.matsim.org/files/dtd/vehicleDefinitions_v2.0.xsd">
     <vehicleType id="car">
-        <attributes>
-        </attributes>
         <description>abc</description>
         <length meter="9.5"/>
         <width meter="3.0"/>
@@ -13,9 +10,6 @@
         <flowEfficiencyFactor factor="1.5"/>
     </vehicleType>
     <vehicleType id="walk">
-        <attributes>
-            <attribute name="lod" class="java.lang.String">teleported</attribute>
-        </attributes>
         <description>This is a pair of shoes</description>
         <length meter="0.5"/>
         <width meter="1.0"/>

rust_qsim/src/bin/local_qsim_routing.rs

@@ -27,29 +27,44 @@ fn main() {
     let total_thread_count = config.partitioning().num_parts + 1;
     let barrier = Arc::new(Barrier::new(total_thread_count as usize));
 
+    // Configuring the routing adapter. We need
+    // - the IP address of the router service
+    // - the configuration of the simulation
+    // - the shutdown handles of the executor (= receiver of shutdown signals from the controller)
+    // The AsyncExecutor will spawn a thread for the routing service adapter and an async runtime.
     let executor = AsyncExecutor::from_config(&config, barrier.clone());
     let factory = RoutingServiceAdapterFactory::new(
         vec![&args.router_ip],
         config.clone(),
         executor.shutdown_handles(),
     );
 
+    // Spawning the routing service adapter in a separate thread. The adapter will be run in its own tokio runtime.
+    // This function returns
+    // - the join handle of the adapter thread
+    // - a channel for sending requests to the adapter
+    // - a channel for sending shutdown signal for the adapter
     let (router_handle, send, send_sd) = executor.spawn_thread("router", factory);
 
+    // The request sender is passed to the controller.
     let mut services = ExternalServices::default();
     services.insert(ExternalServiceType::Routing("pt".into()), send.into());
 
-    let scenario = GlobalScenario::build(config);
+    // Load scenario
+    let scenario = GlobalScenario::load(config);
 
+    // Create controller
     let controller = LocalControllerBuilder::default()
         .global_scenario(scenario)
         .external_services(services)
         .global_barrier(barrier)
         .build()
         .unwrap();
 
+    // Run controller
     let sim_handles = controller.run();
 
+    // Wait for the controller to finish and the routing adapter to finish.
     controller::try_join(
         sim_handles,
         vec![AdapterHandleBuilder::default()

rust_qsim/src/external_services/routing/mod.rs

@@ -109,14 +109,14 @@ impl From<Response> for InternalRoutingResponse {
 pub struct RoutingServiceAdapterFactory {
     ip: Vec<String>,
     config: Arc<Config>,
-    shutdown_handles: Arc<Mutex<Vec<tokio::task::JoinHandle<()>>>>,
+    shutdown_handles: Arc<Mutex<Vec<JoinHandle<()>>>>,
 }
 
 impl RoutingServiceAdapterFactory {
     pub fn new(
         ip: Vec<impl Into<String>>,
         config: Arc<Config>,
-        shutdown_handles: Arc<Mutex<Vec<tokio::task::JoinHandle<()>>>>,
+        shutdown_handles: Arc<Mutex<Vec<JoinHandle<()>>>>,
     ) -> Self {
         Self {
             ip: ip.into_iter().map(|s| s.into()).collect(),

rust_qsim/src/generated/mod.rs

@@ -225,6 +225,12 @@ impl From<&str> for AttributeValue {
     }
 }
 
+impl From<f64> for AttributeValue {
+    fn from(value: f64) -> Self {
+        AttributeValue::new_double(value)
+    }
+}
+
 // we can't tag the enum as non-exhaustive because prost generates it. This is why the warning is manually disabled.
 #[allow(unreachable_patterns)]
 impl Serialize for AttributeValue {

rust_qsim/src/simulation/agents/agent.rs

@@ -68,16 +68,19 @@ impl SimulationAgentLogic for SimulationAgent {
     fn advance_plan(&mut self) {
         self.logic.advance_plan();
     }
-    fn wakeup_time(&self, now: u32) -> u32 {
-        self.logic.wakeup_time(now)
-    }
     fn state(&self) -> SimulationAgentState {
         self.logic.state()
     }
+    fn is_wanting_to_arrive_on_current_link(&self) -> bool {
+        self.logic.is_wanting_to_arrive_on_current_link()
+    }
     fn curr_link_id(&self) -> Option<&Id<Link>> {
         self.logic.curr_link_id()
     }
     fn peek_next_link_id(&self) -> Option<&Id<Link>> {
         self.logic.peek_next_link_id()
     }
+    fn wakeup_time(&self, now: u32) -> u32 {
+        self.logic.wakeup_time(now)
+    }
 }

rust_qsim/src/simulation/agents/agent_logic.rs

@@ -139,17 +139,6 @@ impl SimulationAgentLogic for PlanBasedSimulationLogic {
         );
     }
 
-    fn wakeup_time(&self, now: u32) -> u32 {
-        match self
-            .basic_agent_delegate
-            .plan_element_at(self.curr_plan_element)
-            .unwrap()
-        {
-            InternalPlanElement::Activity(a) => a.cmp_end_time(now),
-            InternalPlanElement::Leg(_) => panic!("Cannot wake up on a leg!"),
-        }
-    }
-
     fn state(&self) -> SimulationAgentState {
         match self.curr_plan_element % 2 {
             0 => SimulationAgentState::ACTIVITY,
@@ -158,6 +147,10 @@ impl SimulationAgentLogic for PlanBasedSimulationLogic {
         }
     }
 
+    fn is_wanting_to_arrive_on_current_link(&self) -> bool {
+        self.peek_next_link_id().is_none()
+    }
+
     fn curr_link_id(&self) -> Option<&Id<Link>> {
         if self.state() != SimulationAgentState::LEG {
             return None;
@@ -192,6 +185,17 @@ impl SimulationAgentLogic for PlanBasedSimulationLogic {
             .unwrap()
             .route_element_at(next_i)
     }
+
+    fn wakeup_time(&self, now: u32) -> u32 {
+        match self
+            .basic_agent_delegate
+            .plan_element_at(self.curr_plan_element)
+            .unwrap()
+        {
+            InternalPlanElement::Activity(a) => a.cmp_end_time(now),
+            InternalPlanElement::Leg(_) => panic!("Cannot wake up on a leg!"),
+        }
+    }
 }
 
 impl EndTime for PlanBasedSimulationLogic {
@@ -228,6 +232,22 @@ impl SimulationAgentLogic for AdaptivePlanBasedSimulationLogic {
         self.delegate.advance_plan();
     }
 
+    fn state(&self) -> SimulationAgentState {
+        self.delegate.state()
+    }
+
+    fn is_wanting_to_arrive_on_current_link(&self) -> bool {
+        self.delegate.is_wanting_to_arrive_on_current_link()
+    }
+
+    fn curr_link_id(&self) -> Option<&Id<Link>> {
+        self.delegate.curr_link_id()
+    }
+
+    fn peek_next_link_id(&self) -> Option<&Id<Link>> {
+        self.delegate.peek_next_link_id()
+    }
+
     fn wakeup_time(&self, now: u32) -> u32 {
         let mut end = self.delegate.curr_act().cmp_end_time(now);
         if let Some(l) = self.delegate.next_leg() {
@@ -240,18 +260,6 @@ impl SimulationAgentLogic for AdaptivePlanBasedSimulationLogic {
         }
         end
     }
-
-    fn state(&self) -> SimulationAgentState {
-        self.delegate.state()
-    }
-
-    fn curr_link_id(&self) -> Option<&Id<Link>> {
-        self.delegate.curr_link_id()
-    }
-
-    fn peek_next_link_id(&self) -> Option<&Id<Link>> {
-        self.delegate.peek_next_link_id()
-    }
 }
 
 impl EndTime for AdaptivePlanBasedSimulationLogic {

rust_qsim/src/simulation/agents/mod.rs

@@ -16,10 +16,13 @@ pub trait SimulationAgentLogic:
     fn curr_leg(&self) -> &InternalLeg;
     fn next_leg(&self) -> Option<&InternalLeg>;
     fn advance_plan(&mut self);
-    fn wakeup_time(&self, now: u32) -> u32;
     fn state(&self) -> SimulationAgentState;
+
+    // Having these functions here is not ideal. See https://github.com/matsim-vsp/parallel_qsim_rust/issues/203 for more details.
+    fn is_wanting_to_arrive_on_current_link(&self) -> bool;
     fn curr_link_id(&self) -> Option<&Id<Link>>;
     fn peek_next_link_id(&self) -> Option<&Id<Link>>;
+    fn wakeup_time(&self, now: u32) -> u32;
 }
 
 pub trait EnvironmentalEventObserver {
@@ -28,12 +31,12 @@ pub trait EnvironmentalEventObserver {
 
 #[non_exhaustive]
 pub enum AgentEvent<'a> {
-    // activity related events
+    // activity-related events
     ActivityStarted(),
     WokeUp(WokeUpEvent<'a>),
     ActivityFinished(),
 
-    // teleportation related events
+    // teleportation-related events
     TeleportationStarted(),
     TeleportationFinished(),
 

rust_qsim/src/simulation/config.rs

@@ -436,10 +436,15 @@ pub struct Simulation {
     pub start_time: u32,
     pub end_time: u32,
     pub sample_size: f32,
+    #[serde(default = "default_to_10")]
     pub stuck_threshold: u32,
     pub main_modes: Vec<String>,
 }
 
+fn default_to_10() -> u32 {
+    10
+}
+
 #[derive(Serialize, Deserialize, Clone, Copy, Debug)]
 pub struct ComputationalSetup {
     pub global_sync: bool,

rust_qsim/src/simulation/controller/local_controller.rs

@@ -122,7 +122,7 @@ pub fn run_channel_from_args() {
     let config = Arc::new(Config::from(args));
 
     // Load and adapt scenario
-    let scenario = GlobalScenario::build(config);
+    let scenario = GlobalScenario::load(config);
 
     // Create and run simulation
     let controller = LocalControllerBuilder::default()