Merge pull request #136 from matsim-vsp/vehicle-driver

add vehicle driver and agents

src/experiments/mod.rs

@@ -6,4 +6,6 @@ mod link_automat_with_references;
 mod link_automata_with_routes;
 mod ownership;
 mod pointers;
+mod run_process_as_service;
+mod stack_vs_heap;
 mod super_simple_link_automat;

src/experiments/run_process_as_service.rs

@@ -0,0 +1,97 @@
+use crate::experiments::run_process_as_service::Message::{Request, Response};
+use std::collections::HashMap;
+use std::sync::mpsc;
+use std::thread;
+use std::time::Duration;
+
+#[derive(Debug)]
+enum Message {
+    //number, id, more
+    Request(u32, u32, bool),
+    Response(u32),
+}
+
+fn run() {
+    // Create a channel
+    let (sender1, receiver_service) = mpsc::channel();
+    let (sender_service_to1, receiver1) = mpsc::channel();
+    let (sender_service_to2, receiver2) = mpsc::channel();
+
+    let sender2 = sender1.clone();
+    // Spawn the first thread
+    thread::spawn(move || {
+        for i in 1..=5 {
+            let request = Request(i, 1, i != 5);
+            println!("Thread 1: Sending number {:?}", request);
+            sender1.send(request).unwrap();
+            // Wait for the response
+            let response = receiver1.recv().unwrap();
+            println!("Thread 1: Received response {:?}", response);
+            thread::sleep(Duration::from_millis(500));
+        }
+    });
+
+    // Spawn the first thread
+    thread::spawn(move || {
+        for i in 6..=10 {
+            let request = Request(i, 2, i != 10);
+            println!("Thread 2: Sending number {:?}", request);
+            sender2.send(request).unwrap();
+            // Wait for the response
+            let response = receiver2.recv().unwrap();
+            println!("Thread 2: Received response {:?}", response);
+            thread::sleep(Duration::from_millis(500));
+        }
+    });
+
+    // Spawn the second thread
+    thread::spawn(move || {
+        let mut expect_more = HashMap::new();
+        expect_more.insert(1, true);
+        expect_more.insert(2, true);
+        while !expect_more.values().all(|v| v == &false) {
+            // Receive the number
+            let in_message = receiver_service.recv().unwrap();
+            // println!("Thread 2: Received number {:?}", in_message);
+            // Send back the doubled number
+            let res;
+            let to_thread_id;
+            match in_message {
+                Request(i, id, more) => {
+                    res = i * 2;
+                    expect_more.insert(id, more);
+                    to_thread_id = id;
+                }
+                Response(_) => {
+                    panic!("Only expect Requests here.")
+                }
+            }
+            let out_message = Response(res);
+            // println!("Thread 2: Sent response {:?}", out_message);
+            match to_thread_id {
+                1 => {
+                    sender_service_to1.send(out_message).unwrap();
+                }
+                2 => {
+                    sender_service_to2.send(out_message).unwrap();
+                }
+                _ => panic!(),
+            }
+        }
+    })
+    .join()
+    .unwrap(); // Ensure the second thread completes
+
+    // Give some time for the first thread to complete
+    // thread::sleep(Duration::from_secs(3));
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::experiments::run_process_as_service::run;
+
+    #[test]
+    fn test_run() {
+        run();
+    }
+}

src/experiments/stack_vs_heap.rs

@@ -0,0 +1,61 @@
+use std::time::Instant;
+
+#[derive(Debug)]
+struct DummyStruct {
+    i: usize,
+}
+
+impl DummyStruct {
+    fn add(&mut self, i: usize) {
+        self.i += i;
+    }
+}
+
+fn stack_allocation(n: usize) -> DummyStruct {
+    let mut sum = DummyStruct { i: 0 };
+    for i in 0..n {
+        sum.add(i)
+    }
+    sum
+}
+
+fn heap_allocation(n: usize) -> Box<DummyStruct> {
+    let mut sum = Box::new(DummyStruct { i: 0 });
+    for i in 0..n {
+        sum.add(i)
+    }
+    sum
+}
+
+fn run() {
+    let n = 10_000_000;
+
+    let start = Instant::now();
+    let stack_result = stack_allocation(n);
+    let stack_duration = start.elapsed();
+
+    let start = Instant::now();
+    let heap_result = heap_allocation(n);
+    let heap_duration = start.elapsed();
+
+    println!(
+        "Stack result: {:?}, Time: {:?}",
+        stack_result, stack_duration
+    );
+    println!("Heap result: {:?}, Time: {:?}", heap_result, heap_duration);
+
+    println!(
+        "Ratio Heap/Stack: {:?}",
+        heap_duration.as_nanos() / stack_duration.as_nanos()
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::experiments::stack_vs_heap::run;
+
+    #[test]
+    fn test_run() {
+        run()
+    }
+}

src/simulation/messaging/messages.rs

@@ -110,19 +110,20 @@ impl Ord for SyncMessage {
 
 impl Vehicle {
     // todo, fix type and mode
-    pub fn new(id: u64, veh_type: u64, max_v: f32, pce: f32, agent: Option<Person>) -> Vehicle {
+    pub fn new(id: u64, veh_type: u64, max_v: f32, pce: f32, driver: Option<Person>) -> Vehicle {
         Vehicle {
             id,
-            agent,
+            driver,
             curr_route_elem: 0,
             r#type: veh_type,
             max_v,
             pce,
+            passengers: vec![],
         }
     }
 
-    pub fn agent(&self) -> &Person {
-        self.agent.as_ref().unwrap()
+    pub fn driver(&self) -> &Person {
+        self.driver.as_ref().unwrap()
     }
 
     pub fn id(&self) -> usize {
@@ -138,33 +139,33 @@ impl Vehicle {
     /// the vehicle is independent of whether the leg has a Generic-Teleportation route or a network
     /// route.
     pub fn route_index_to_last(&mut self) {
-        let route_len = self.agent().curr_leg().route.as_ref().unwrap().route.len() as u32;
+        let route_len = self.driver().curr_leg().route.as_ref().unwrap().route.len() as u32;
         self.curr_route_elem = route_len - 1;
     }
 
     pub fn curr_link_id(&self) -> Option<u64> {
-        let leg = self.agent().curr_leg();
+        let leg = self.driver().curr_leg();
         let route = leg.route.as_ref().unwrap();
         let index = self.curr_route_elem as usize;
         route.route.get(index).copied()
     }
 
     // todo same as above
     pub fn is_current_link_last(&self) -> bool {
-        let leg = self.agent().curr_leg();
+        let leg = self.driver().curr_leg();
         let route = leg.route.as_ref().unwrap();
         self.curr_route_elem + 1 >= route.route.len() as u32
     }
 
     pub fn peek_next_route_element(&self) -> Option<u64> {
-        let route = self.agent().curr_leg().route.as_ref().unwrap();
+        let route = self.driver().curr_leg().route.as_ref().unwrap();
         let next_i = self.curr_route_elem as usize + 1;
         route.route.get(next_i).copied()
     }
 }
 
 impl EndTime for Vehicle {
     fn end_time(&self, now: u32) -> u32 {
-        self.agent().end_time(now)
+        self.driver().end_time(now)
     }
 }

src/simulation/network/sim_network.rs

@@ -182,7 +182,7 @@ impl SimNetworkPartition {
             panic!("Vehicle is expected to have a current link id if it is sent onto the network")
         });
         let link = self.links.get_mut(&link_id).unwrap_or_else(|| {
-            let agent_id = Id::<Person>::get(vehicle.agent().id());
+            let agent_id = Id::<Person>::get(vehicle.driver().id());
             panic!(
                 "#{} Couldn't find link for id {:?}.for Agent {}. \n\n The vehicle: {:?}",
                 self.partition,

src/simulation/simulation.rs

@@ -128,7 +128,7 @@ where
                 LevelOfDetail::Network => {
                     self.events.publish_event(
                         now,
-                        &Event::new_person_enters_veh(vehicle.agent().id, vehicle.id),
+                        &Event::new_person_enters_veh(vehicle.driver().id, vehicle.id),
                     );
                     //we don't pass the event publisher because a link enter event should not be published
                     self.network.send_veh_en_route(vehicle, None, now);
@@ -213,7 +213,7 @@ where
 
         for veh in exited_vehicles {
             self.events
-                .publish_event(now, &Event::new_person_leaves_veh(veh.agent().id, veh.id));
+                .publish_event(now, &Event::new_person_leaves_veh(veh.driver().id, veh.id));
             let veh_type_id = Id::get(veh.r#type);
             let veh_type = self.garage.vehicle_types.get(&veh_type_id).unwrap();
             let mode = veh_type.net_mode;
@@ -266,7 +266,7 @@ where
     }
 
     fn is_local_route(veh: &Vehicle, message_broker: &NetMessageBroker<C>) -> bool {
-        let leg = veh.agent.as_ref().unwrap().curr_leg();
+        let leg = veh.driver.as_ref().unwrap().curr_leg();
         let route = leg.route.as_ref().unwrap();
         let to = message_broker.rank_for_link(route.end_link());
         message_broker.rank() == to

src/simulation/vehicles/garage.rs

@@ -99,7 +99,7 @@ impl Garage {
         // the above logic would park a vehicle within a garage. This only works if we have mass
         // conservation enabled. The scenario we're testing with doesn't. Therfore, we just take
         // the agent out of the vehicle and pretend we have parked the car.
-        vehicle.agent.unwrap()
+        vehicle.driver.unwrap()
     }
 
     pub fn unpark_veh(&mut self, person: Person, id: &Id<Vehicle>) -> Vehicle {
@@ -133,7 +133,8 @@ impl Garage {
             r#type: veh_type.id,
             max_v: veh_type.max_v,
             pce: veh_type.pce,
-            agent: Some(person),
+            driver: Some(person),
+            passengers: vec![],
         }
     }
 }

src/simulation/wire_types/messages.proto

@@ -37,6 +37,6 @@ message Vehicle {
   uint64 type = 3;
   float max_v = 4;
   float pce = 5;
-  population.Person agent = 6;
-  // this must be extended once more people use one vehicle
+  population.Person driver = 6;
+  repeated population.Person passengers = 7;
 }