Merge branch 'refs/heads/main' into internal-interface-experiments

# Conflicts:
#	src/simulation/controller.rs

Author: paulheinr

.github/workflows/rust.yml

@@ -18,10 +18,10 @@ jobs:
       - name: Install other libraries
         run: sudo apt -y install libclang-dev llvm-dev libmetis-dev libopenmpi-dev
       - name: Build
-        run: cargo build
+        run: cargo build --release
       - name: Install mpirun
         run: cargo install --force cargo-mpirun
       - name: Run tests
         env:
           RUST_BACKTRACE: 1
-        run: cargo test --verbose -- --test-threads=1
+        run: cargo test --release --verbose -- --test-threads=1

assets/adhoc_routing/vehicles.xml renamed to assets/adhoc_routing/no_updates/vehicles.xml

@@ -0,0 +1,38 @@
+<?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_v2.0.xsd">
+    <vehicleType id="car">
+        <attributes>
+        </attributes>
+        <description>abc</description>
+        <length meter="9.5"/>
+        <width meter="3.0"/>
+        <maximumVelocity meterPerSecond="20.0"/>
+        <passengerCarEquivalents pce="1.0"/>
+        <networkMode networkMode="car"/>
+        <flowEfficiencyFactor factor="1.5"/>
+    </vehicleType>
+    <vehicleType id="bike">
+        <attributes>
+        </attributes>
+        <description>This is a bike</description>
+        <length meter="2.0"/>
+        <width meter="1.0"/>
+        <maximumVelocity meterPerSecond="5.0"/>
+        <passengerCarEquivalents pce="0.25"/>
+        <networkMode networkMode="bike"/>
+        <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"/>
+        <maximumVelocity meterPerSecond="1.2"/>
+        <passengerCarEquivalents pce="0.1"/>
+        <networkMode networkMode="walk"/>
+        <flowEfficiencyFactor factor="10.0"/>
+    </vehicleType>
+</vehicleDefinitions>

assets/adhoc_routing/with_updates/vehicles.xml

@@ -28,8 +28,8 @@ fn main() {
     rust_q_sim::simulation::logging::init_std_out_logging();
     let args = InputArgs::parse();
 
-    let mut net = Network::from_file_path(&args.network, 1, PartitionMethod::None);
     let mut veh = Garage::from_file(&args.vehicles);
+    let mut net = Network::from_file_path(&args.network, 1, PartitionMethod::None);
     let pop = Population::from_file(&args.population, &mut veh);
 
     let cmp_weights = compute_computational_weights(&pop);
@@ -39,6 +39,7 @@ fn main() {
     net.to_file(&create_file_path(&args, "network"));
     veh.to_file(&create_file_path(&args, "vehicles"));
     pop.to_file(&create_file_path(&args, "plans"));
+    info!("Finished conversion. Exiting.")
 }
 
 fn create_file_path(args: &InputArgs, extension: &str) -> PathBuf {

src/bin/convert_to_binary.rs

@@ -44,9 +44,10 @@ fn run() {
     );
     println!("Heap result: {:?}, Time: {:?}", heap_result, heap_duration);
 
+    // divide by at least one
     println!(
         "Ratio Heap/Stack: {:?}",
-        heap_duration.as_nanos() / stack_duration.as_nanos()
+        heap_duration.as_nanos() / stack_duration.as_nanos().max(1)
     )
 }
 

src/experiments/stack_vs_heap.rs

@@ -43,6 +43,7 @@ impl Config {
             config.set_output(Output {
                 output_dir: out_dir,
                 profiling: config.output().profiling,
+                write_events: Default::default(),
             });
         }
         config
@@ -90,6 +91,7 @@ impl Config {
             let default = Output {
                 output_dir: "./".to_string(),
                 profiling: Profiling::None,
+                write_events: Default::default(),
             };
             self.modules
                 .borrow_mut()
@@ -157,6 +159,8 @@ pub struct Output {
     pub output_dir: String,
     #[serde(default)]
     pub profiling: Profiling,
+    #[serde(default)]
+    pub write_events: WriteEvents,
 }
 
 #[derive(Serialize, Deserialize, Clone)]
@@ -244,6 +248,13 @@ pub enum Profiling {
     CSV(ProfilingLevel),
 }
 
+#[derive(PartialEq, Debug, Clone, Serialize, Deserialize, Default)]
+pub enum WriteEvents {
+    #[default]
+    None,
+    Proto,
+}
+
 #[derive(PartialEq, Debug, Clone, Serialize, Deserialize, Default)]
 pub struct ProfilingLevel {
     #[serde(default = "default_profiling_level")]

src/simulation/config.rs

@@ -10,15 +10,16 @@ use mpi::traits::{Communicator, CommunicatorCollectives};
 use nohash_hasher::IntMap;
 use tracing::info;
 
-use crate::simulation::config::{CommandLineArgs, Config, PartitionMethod, RoutingMode};
-use crate::simulation::id::Id;
+use crate::simulation::config::{
+    CommandLineArgs, Config, PartitionMethod, RoutingMode, WriteEvents,
+};
 use crate::simulation::io::proto_events::ProtoEventsWriter;
 use crate::simulation::messaging::communication::communicators::{
     ChannelSimCommunicator, MpiSimCommunicator, SimCommunicator,
 };
 use crate::simulation::messaging::communication::message_broker::NetMessageBroker;
 use crate::simulation::messaging::events::EventsPublisher;
-use crate::simulation::network::global_network::{Link, Network};
+use crate::simulation::network::global_network::Network;
 use crate::simulation::network::sim_network::SimNetworkPartition;
 use crate::simulation::population::population::Population;
 use crate::simulation::replanning::replanner::{DummyReplanner, ReRouteTripReplanner, Replanner};
@@ -111,13 +112,11 @@ fn execute_partition<C: SimCommunicator + 'static>(comm: C, args: &CommandLineAr
     ));
     let mut garage = Garage::from_file(&PathBuf::from(config.proto_files().vehicles));
 
-    let population: Population = Population::from_file(
-        &get_numbered_output_filename(
-            &output_path,
-            &PathBuf::from(config.proto_files().population),
-            rank,
-        ),
+    let population = Population::from_file_filtered_part(
+        &PathBuf::from(config.proto_files().population),
+        &network,
         &mut garage,
+        comm.rank(),
     );
 
     let network_partition = SimNetworkPartition::from_network(&network, rank, config.simulation());
@@ -130,14 +129,13 @@ fn execute_partition<C: SimCommunicator + 'static>(comm: C, args: &CommandLineAr
 
     let events = Rc::new(RefCell::new(EventsPublisher::new()));
 
-    let events_file = format!("events.{rank}.binpb");
-    let events_path = output_path.join(events_file);
-    events
-        .borrow_mut()
-        .add_subscriber(Box::new(ProtoEventsWriter::new(&events_path)));
+    if config.output().write_events == WriteEvents::Proto {
+        let events_file = format!("events.{rank}.binpb");
+        let events_path = output_path.join(events_file);
+        events.borrow_mut().add_subscriber(Box::new(ProtoEventsWriter::new(&events_path)));
+    }
     let travel_time_collector = Box::new(TravelTimeCollector::new());
     events.borrow_mut().add_subscriber(travel_time_collector);
-    //events.add_subscriber(Box::new(EventsLogger {}));
 
     let rc = Rc::new(comm);
 
@@ -186,15 +184,13 @@ fn try_join(mut handles: IntMap<u32, JoinHandle<()>>) {
 
 pub fn partition_input(config: &Config) {
     id::load_from_file(&PathBuf::from(config.proto_files().ids));
-    let net = if let PartitionMethod::Metis(_) = config.partitioning().method {
+    let _net = if let PartitionMethod::Metis(_) = config.partitioning().method {
         info!("Config param Partition method was set to metis. Loading input network, running metis conversion and then store it into output folder");
         partition_network(config)
     } else {
         info!("Config param Partition method was set to none. Loading network from input, assuming it has partitioning information");
         copy_network_into_output(config)
     };
-
-    partition_population(config, &net);
 }
 
 fn partition_network(config: &Config) -> Network {
@@ -220,38 +216,6 @@ fn copy_network_into_output(config: &Config) -> Network {
     network
 }
 
-fn partition_population(config: &Config, network: &Network) {
-    info!(
-        "Partition population into {} parts",
-        config.partitioning().num_parts
-    );
-    let pop_in_path = PathBuf::from(config.proto_files().population);
-    let num_parts = config.partitioning().num_parts;
-    let pop = Population::from_file(&pop_in_path, &mut Garage::new());
-    let mut part_pops = vec![];
-    for _i in 0..num_parts {
-        part_pops.push(Population::new())
-    }
-
-    let pop_out_path =
-        create_output_filename(&PathBuf::from(config.output().output_dir), &pop_in_path);
-
-    for (id, person) in pop.persons.into_iter() {
-        let link_id: Id<Link> = Id::get(person.curr_act().link_id);
-        let partition = network.get_link(&link_id).partition;
-        part_pops
-            .get_mut(partition as usize)
-            .unwrap()
-            .persons
-            .insert(id, person);
-    }
-
-    for (i, population) in part_pops.iter().enumerate() {
-        let part_out_path = insert_number_in_proto_filename(&pop_out_path, i as u32);
-        population.to_file(&part_out_path);
-    }
-}
-
 pub fn get_numbered_output_filename(output_dir: &Path, input_file: &Path, part: u32) -> PathBuf {
     let out = create_output_filename(output_dir, input_file);
     insert_number_in_proto_filename(&out, part)

src/simulation/controller.rs

@@ -37,6 +37,39 @@ pub fn write_to_file<T: Message>(message: T, path: &Path) {
     info!("Finished writing message to file: {path:?}");
 }
 
+pub fn read_delimiter<R>(reader: &mut BufReader<R>) -> Option<usize>
+where
+    R: Read + Seek,
+{
+    // read the delimiter of the message. Prost says delimiter is between 1 and 10 bytes
+    // so, read the first 10 bytes of the buffer
+    let mut delim_buffer: [u8; 10] = [0; 10];
+    // this could crash
+    match reader.read_exact(&mut delim_buffer) {
+        Ok(_) => {} // go on.
+        Err(e) => match e.kind() {
+            ErrorKind::UnexpectedEof => return None,
+            _ => {
+                panic!("Error while reading file: {}", e);
+            }
+        },
+    };
+
+    let delimiter =
+        prost::decode_length_delimiter(delim_buffer.as_slice()).expect("error reading delimiter");
+
+    // since the delimiter is a varint figure out how many bytes the delimiter was actually taking
+    // up in the buffer. Set the buffers position to the first byte after the delimiter, which
+    // should be the start of the TimeStep message
+    let delim_encoded_len = prost::encoding::encoded_len_varint(delimiter as u64) as i64;
+    let offset = delim_encoded_len - (delim_buffer.len() as i64);
+    reader
+        .seek_relative(offset)
+        .expect("Seeking relative failed");
+
+    Some(delimiter)
+}
+
 pub struct MessageIter<T, R>
 where
     T: Message + Default,
@@ -54,7 +87,7 @@ where
     type Item = T;
 
     fn next(&mut self) -> Option<Self::Item> {
-        if let Some(delimiter) = self.read_delimiter() {
+        if let Some(delimiter) = read_delimiter(&mut self.internal_reader) {
             let mut bytes: Vec<u8> = vec![0; delimiter];
             self.internal_reader
                 .read_exact(&mut bytes)
@@ -78,33 +111,4 @@ where
             internal_reader: BufReader::new(reader),
         }
     }
-    fn read_delimiter(&mut self) -> Option<usize> {
-        // read the delimiter of the message. Prost says delimiter is between 1 and 10 bytes
-        // so, read the first 10 bytes of the buffer
-        let mut delim_buffer: [u8; 10] = [0; 10];
-        // this could crash
-        match self.internal_reader.read_exact(&mut delim_buffer) {
-            Ok(_) => {} // go on.
-            Err(e) => match e.kind() {
-                ErrorKind::UnexpectedEof => return None,
-                _ => {
-                    panic!("Error while reading file: {}", e);
-                }
-            },
-        };
-
-        let delimiter = prost::decode_length_delimiter(delim_buffer.as_slice())
-            .expect("error reading delimiter");
-
-        // since the delimiter is a varint figure out how many bytes the delimiter was actually taking
-        // up in the buffer. Set the buffers position to the first byte after the delimiter, which
-        // should be the start of the TimeStep message
-        let delim_encoded_len = prost::encoding::encoded_len_varint(delimiter as u64) as i64;
-        let offset = delim_encoded_len - (delim_buffer.len() as i64);
-        self.internal_reader
-            .seek_relative(offset)
-            .expect("Seeking relative failed");
-
-        Some(delimiter)
-    }
 }