fix: sim data recv only after cold started and parents all done

ActivePeter · ActivePeter · commit fff6bcd99b41 · 2024-07-23T22:28:14.000+08:00
diff --git a/serverless_sim/Cargo.toml b/serverless_sim/Cargo.toml
@@ -32,17 +32,19 @@ rand_pcg = "0.3.1"
 async-trait = "0.1.51"
 parking_lot = "0.12"
 rand_distr = "0.4.3"
+futures = "0.3"
+cpu-time = "1.0.0"
+
+[target.'cfg(windows)'.dependencies]
 thread-priority = "0.3"
-# futures = "0.3"
-# cpu-time = "1.0.0"
 
-# [dependencies.windows]
-# version = "0.52"
-# features = [
-#     "Win32_Foundation",
-#     "Win32_System_Threading",
-#     "Win32_System_Diagnostics_Etw",
-#     "Win32_System_Time",
-#     "Win32_Foundation",
-#     "Win32_Security"
-# ]
+[dependencies.windows]
+version = "0.52"
+features = [
+    "Win32_Foundation",
+    "Win32_System_Threading",
+    "Win32_System_Diagnostics_Etw",
+    "Win32_System_Time",
+    "Win32_Foundation",
+    "Win32_Security",
+]
diff --git a/serverless_sim/src/mechanism_thread.rs b/serverless_sim/src/mechanism_thread.rs
@@ -1,11 +1,12 @@
 use std::sync::mpsc;
 
-use enum_as_inner::EnumAsInner;
-use thread_priority::{ set_current_thread_priority, ThreadPriority, WinAPIThreadPriority };
-// use windows::Win32::System::Threading::{ SetThreadPriority, GetCurrentThread, THREAD_PRIORITY };
+#[cfg(target_os = "windows")]
+use thread_priority::{set_current_thread_priority, ThreadPriority, WinAPIThreadPriority};
+#[cfg(target_os = "windows")]
+use windows::Win32::System::Threading::{GetCurrentThread, SetThreadPriority, THREAD_PRIORITY};
 
 use crate::actions::ESActionWrapper;
-use crate::mechanism::{ DownCmd, Mechanism, MechanismImpl, ScheCmd, SimEnvObserve, UpCmd };
+use crate::mechanism::{DownCmd, Mechanism, MechanismImpl, ScheCmd, SimEnvObserve, UpCmd};
 
 use crate::util;
 use crate::with_env_sub::WithEnvHelp;
@@ -43,6 +44,8 @@ pub fn spawn(mech: MechanismImpl) -> mpsc::Sender<MechScheduleOnce> {
         //         THREAD_PRIORITY(WinAPIThreadPriority::TimeCritical as i32)
         //     ).unwrap();
         // }
+
+        #[cfg(target_os = "windows")]
         if let Err(e) = set_current_thread_priority(ThreadPriority::Max) {
             eprintln!("设置线程优先级失败: {:?}", e);
         }
@@ -71,7 +74,11 @@ fn mechanism_loop(rx: mpsc::Receiver<MechScheduleOnce>, mech: MechanismImpl) {
         // let passed_ms = measure.passed_100ns();
         let end_ms = util::now_ms();
         // log::info!("master mech run cpu:{:?}, total:{} ms", begin_cpu.elapsed(), end_ms - begin_ms);
-        let mech_latency = if mech.config.no_mech_latency { 0 } else { end_ms - begin_ms };
+        let mech_latency = if mech.config.no_mech_latency {
+            0
+        } else {
+            end_ms - begin_ms
+        };
         res.responser
             .send(MechScheduleOnceRes::End {
                 mech_run_ms: mech_latency,
@@ -85,11 +92,15 @@ fn mechanism_loop(rx: mpsc::Receiver<MechScheduleOnce>, mech: MechanismImpl) {
 pub mod tests {
     use std::sync::mpsc;
 
-    use crate::{ actions::ESActionWrapper, mechanism_thread::MechScheduleOnceRes, sim_env::SimEnv };
+    use crate::{actions::ESActionWrapper, mechanism_thread::MechScheduleOnceRes, sim_env::SimEnv};
 
     #[test]
     pub fn test_algo_latency() {
-        use std::{ cell::RefCell, rc::Rc, sync::{ atomic::AtomicU64, Arc } };
+        use std::{
+            cell::RefCell,
+            rc::Rc,
+            sync::{atomic::AtomicU64, Arc},
+        };
 
         use crate::config::Config;
         let _ = env_logger::try_init();
@@ -120,24 +131,20 @@ pub mod tests {
             ESActionWrapper::Int(0),
             None,
             None,
-            Some(
-                Box::new(move |env: &SimEnv| {
-                    *begin_frame.borrow_mut() = env.current_frame();
-                })
-            ),
-            Some(
-                Box::new(move |env: &SimEnv| {
-                    // calltime = env.current_frame() - begin_frame;
-                    assert!(
-                        env.current_frame() - *begin_frame2.borrow() == calltime,
-                        "begin_frame:{} current_frame:{} calltime:{}",
-                        begin_frame2.borrow(),
-                        env.current_frame(),
-                        calltime
-                    );
-                    calltime += 1;
-                })
-            )
+            Some(Box::new(move |env: &SimEnv| {
+                *begin_frame.borrow_mut() = env.current_frame();
+            })),
+            Some(Box::new(move |env: &SimEnv| {
+                // calltime = env.current_frame() - begin_frame;
+                assert!(
+                    env.current_frame() - *begin_frame2.borrow() == calltime,
+                    "begin_frame:{} current_frame:{} calltime:{}",
+                    begin_frame2.borrow(),
+                    env.current_frame(),
+                    calltime
+                );
+                calltime += 1;
+            })),
         );
     }
 }
diff --git a/serverless_sim/src/request.rs b/serverless_sim/src/request.rs
@@ -104,6 +104,7 @@ impl Request {
                 ),
             );
         }
+        // log::info!("endtime_fn: {:?}", endtime_fn);
         let first = endtime_fn.iter().next().unwrap().1.clone().1;
         let mut cur: (usize, FnId) = endtime_fn.iter().next_back().unwrap().1.clone();
         let mut recur_path = vec![cur.1];
@@ -112,7 +113,12 @@ impl Request {
                 break;
             }
             // use cur fn's begin time to get prev fn's end time
-            let prev: (usize, FnId) = endtime_fn.get(&cur.0).unwrap().clone();
+            let prev: (usize, FnId) = endtime_fn
+                .get(&cur.0)
+                .unwrap_or_else(|| {
+                    panic!("can't find fn end at {}", cur.0);
+                })
+                .clone();
             recur_path.push(prev.1);
             if prev.1 == first {
                 break;
diff --git a/serverless_sim/src/sim_env.rs b/serverless_sim/src/sim_env.rs
@@ -385,8 +385,8 @@ impl SimEnv {
                     rng
                 );
             }
-            // mkdir
-            std::fs::create_dir("cache");
+            // mkdir, allow failure
+            let _ = std::fs::create_dir("cache");
             // write to file
             let mut file = std::fs::File::create(cache_req_freq).unwrap();
             serde_json::to_writer(&mut file, &*self.help.fn_call_frequency()).unwrap();
@@ -507,7 +507,8 @@ impl SimEnv {
 
         // 自增 frame
         let mut cur_frame = self.core.current_frame.borrow_mut();
-        // log::info!("frame done: {}", *cur_frame);
+
+        log::info!("frame done: {}", *cur_frame);
         *cur_frame += 1;
     }
 }
diff --git a/serverless_sim/src/sim_events/on_task_done.rs b/serverless_sim/src/sim_events/on_task_done.rs
@@ -23,12 +23,11 @@ impl SimEnv {
     }
     pub fn on_task_done(&self, req: &mut Request, fnid: FnId) {
         self.check_sub_tasks_ready_sche(req, fnid);
-        assert!(req
-            .fn_metric
-            .get_mut(&fnid)
-            .unwrap()
+        let fn_metric = req.fn_metric.get_mut(&fnid).unwrap();
+        assert!(fn_metric
             .fn_done_time
             .replace(self.current_frame())
             .is_none());
+        assert!(fn_metric.ready_sche_time.is_some());
     }
 }
diff --git a/serverless_sim/src/sim_events/on_task_ready_schedule.rs b/serverless_sim/src/sim_events/on_task_ready_schedule.rs
@@ -2,13 +2,10 @@ use crate::{fn_dag::FnId, request::Request, sim_env::SimEnv};
 
 impl SimEnv {
     pub fn on_task_ready_sche(&self, req: &mut Request, fnid: FnId) {
-        assert!(req
-            .fn_metric
-            .get_mut(&fnid)
-            .unwrap()
-            .ready_sche_time
-            .is_none());
-        req.fn_metric.get_mut(&fnid).unwrap().ready_sche_time = Some(self.current_frame());
+        let fnmetric = req.fn_metric.get_mut(&fnid).unwrap();
+        assert!(fnmetric.ready_sche_time.is_none());
+        fnmetric.ready_sche_time = Some(self.current_frame());
+        assert!(fnmetric.data_recv_done_time.is_none());
         // Happend in this frame. So real ready is next frame
     }
 }
diff --git a/serverless_sim/src/sim_loop.rs b/serverless_sim/src/sim_loop.rs
@@ -1,3 +1,4 @@
+#[cfg(target_os = "windows")]
 use thread_priority::{set_current_thread_priority, ThreadPriority};
 
 use crate::{
@@ -62,9 +63,11 @@ impl SimEnv {
         mut hook_algo_end: Option<Box<dyn FnMut(&SimEnv) + 'static>>,
     ) -> (f32, String) {
         // 尝试设置当前线程的优先级
+        #[cfg(target_os = "windows")]
         if let Err(e) = set_current_thread_priority(ThreadPriority::Min) {
             eprintln!("设置线程优先级失败: {:?}", e);
         }
+
         self.avoid_gc();
         let mut master_mech_resp_rx: Option<Receiver<MechScheduleOnceRes>> = None;
         let mut frame_when_master_mech_begin = 0;
@@ -180,6 +183,7 @@ impl SimEnv {
                 hook_algo_begin.as_mut().map(|f| f(self));
             }
             if !self.one_frame(&mut hook_frame_begin, &mut hook_req_gen) {
+                log::info!("simulation end");
                 break;
             }
 
diff --git a/serverless_sim/src/sim_run.rs b/serverless_sim/src/sim_run.rs

Original file line number	Diff line number	Diff line change
`@@ -385,8 +385,8 @@ impl SimEnv {`
`385`	`385`	`rng`
`386`	`386`	`);`
`387`	`387`	`}`
`388`		`- // mkdir`
`389`		`- std::fs::create_dir("cache");`
	`388`	`+ // mkdir, allow failure`
	`389`	`+ let _ = std::fs::create_dir("cache");`
`390`	`390`	`// write to file`
`391`	`391`	`let mut file = std::fs::File::create(cache_req_freq).unwrap();`
`392`	`392`	`serde_json::to_writer(&mut file, &*self.help.fn_call_frequency()).unwrap();`
`@@ -507,7 +507,8 @@ impl SimEnv {`
`507`	`507`
`508`	`508`	`// 自增 frame`
`509`	`509`	`let mut cur_frame = self.core.current_frame.borrow_mut();`
`510`		`- // log::info!("frame done: {}", *cur_frame);`
	`510`	`+`
	`511`	`+ log::info!("frame done: {}", *cur_frame);`
`511`	`512`	`*cur_frame += 1;`
`512`	`513`	`}`
`513`	`514`	`}`
Original file line number	Diff line number	Diff line change
`@@ -23,12 +23,11 @@ impl SimEnv {`
`23`	`23`	`}`
`24`	`24`	`pub fn on_task_done(&self, req: &mut Request, fnid: FnId) {`
`25`	`25`	`self.check_sub_tasks_ready_sche(req, fnid);`
`26`		`- assert!(req`
`27`		`- .fn_metric`
`28`		`- .get_mut(&fnid)`
`29`		`- .unwrap()`
	`26`	`+ let fn_metric = req.fn_metric.get_mut(&fnid).unwrap();`
	`27`	`+ assert!(fn_metric`
`30`	`28`	`.fn_done_time`
`31`	`29`	`.replace(self.current_frame())`
`32`	`30`	`.is_none());`
	`31`	`+ assert!(fn_metric.ready_sche_time.is_some());`
`33`	`32`	`}`
`34`	`33`	`}`