[runtime] Enhancement: Adaptable timer res

src/rust/runtime/mod.rs

@@ -27,6 +27,7 @@ pub use queue::{
     QType,
 };
 pub use scheduler::TaskId;
+use x86::time::rdtscp;
 
 #[cfg(feature = "libdpdk")]
 pub use dpdk_rs as libdpdk;
@@ -109,6 +110,12 @@ pub struct DemiRuntime {
     network_table: NetworkQueueTable,
     /// Number of iterations that we have polled since advancing the clock.
     ts_iters: usize,
+    /// Last tsc counter value.
+    last_tsc: u64,
+    /// List time instant.
+    last_time: Instant,
+    /// Timer resolution.
+    timer_resolution: usize,
     /// Tasks that have been completed and removed from the
     completed_tasks: HashMap<QToken, (QDesc, OperationResult)>,
 }
@@ -146,6 +153,12 @@ impl SharedDemiRuntime {
             ephemeral_ports: EphemeralPorts::default(),
             network_table: NetworkQueueTable::default(),
             ts_iters: 0,
+            last_tsc: unsafe {
+                let (rdtscp, _) = rdtscp();
+                rdtscp
+            },
+            last_time: Instant::now(),
+            timer_resolution: TIMER_RESOLUTION,
             completed_tasks: HashMap::<QToken, (QDesc, OperationResult)>::new(),
         }))
     }
@@ -212,11 +225,12 @@ impl SharedDemiRuntime {
             return Err(Fail::new(libc::EINVAL, &cause));
         }
 
-        // 2. None of the tasks have already completed, so start a timer and move the clock.
-        self.advance_clock_to_now();
+        // None of the tasks have already completed.
 
         loop {
-            if let Some(boxed_task) = self.scheduler.get_next_completed_task(TIMER_RESOLUTION) {
+            self.advance_clock_to_now();
+            let max_iterations = self.timer_resolution;
+            if let Some(boxed_task) = self.scheduler.get_next_completed_task(max_iterations) {
                 // Perform bookkeeping for the completed and removed task.
                 trace!("Removing coroutine: {:?}", boxed_task.get_name());
                 let completed_qt: QToken = boxed_task.get_id().into();
@@ -240,9 +254,6 @@ impl SharedDemiRuntime {
                     return Err(Fail::new(libc::ETIMEDOUT, "wait timed out"));
                 }
             }
-
-            // Advance the clock and continue running tasks.
-            self.advance_clock_to_now();
         }
     }
 
@@ -359,7 +370,7 @@ impl SharedDemiRuntime {
     /// the clock.
     fn run_next(&mut self, timeout: Duration) -> Option<(QToken, QDesc, OperationResult)> {
         let iterations: usize = match timeout {
-            timeout if timeout.as_secs() > 0 => TIMER_RESOLUTION,
+            timeout if timeout.as_secs() > 0 => self.timer_resolution,
             _ => TIMER_FINER_RESOLUTION,
         };
         if let Some(boxed_task) = self.scheduler.get_next_completed_task(iterations) {
@@ -478,15 +489,16 @@ impl SharedDemiRuntime {
 
     /// Moves time forward deterministically.
     pub fn advance_clock(&mut self, now: Instant) {
-        timer::global_advance_clock(now)
+        timer::global_advance_clock(now);
+        self.adjust_time_resolution(now);
     }
 
     /// Moves time forward to the current real time.
     fn advance_clock_to_now(&mut self) {
         if self.ts_iters == 0 {
             self.advance_clock(Instant::now());
         }
-        self.ts_iters = (self.ts_iters + 1) % TIMER_RESOLUTION;
+        self.ts_iters = (self.ts_iters + 1) % self.timer_resolution;
     }
 
     /// Gets the current time according to our internal timer.
@@ -535,6 +547,24 @@ impl SharedDemiRuntime {
         trace!("Check address in use: {:?}", local);
         self.network_table.addr_in_use(local)
     }
+
+    fn adjust_time_resolution(&mut self, now: Instant) {
+        let curr_tsc: u64 = unsafe {
+            let (tsc, _): (u64, u32) = rdtscp();
+            tsc
+        };
+        let cycles: usize = (curr_tsc - self.last_tsc) as usize;
+        let time: Duration = now - self.last_time;
+        let cycles_per_second: usize = cycles / time.as_secs() as usize;
+
+        if cycles_per_second > 0 {
+            let cycles_per_quanta: usize = cycles_per_second / TIMER_RESOLUTION;
+            self.timer_resolution = cycles_per_quanta;
+            trace!("Adjusted time resolution: resolution={:?}", self.timer_resolution);
+        }
+        self.last_tsc = curr_tsc;
+        self.last_time = now;
+    }
 }
 
 impl<T> SharedObject<T> {
@@ -597,6 +627,12 @@ impl Default for SharedDemiRuntime {
             ephemeral_ports: EphemeralPorts::default(),
             network_table: NetworkQueueTable::default(),
             ts_iters: 0,
+            last_tsc: unsafe {
+                let (tsc, _) = rdtscp();
+                tsc
+            },
+            last_time: Instant::now(),
+            timer_resolution: TIMER_RESOLUTION,
             completed_tasks: HashMap::<QToken, (QDesc, OperationResult)>::new(),
         }))
     }