-
-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* feat: implement logical clock Previously, the benchmark runner uses physical time to track the benchmark progress. This is not ideal because the physical time keeps moving forward even when the benchmark is paused. Also the throughput calculation is not accurate when there are pauses. This commit introduces a logical clock that can be paused and resumed to improve the pause functionality. Signed-off-by: Wenxuan Zhang <wenxuangm@gmail.com>
- Loading branch information
Showing
9 changed files
with
157 additions
and
73 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| use std::sync::Arc; | ||
|
|
||
| use parking_lot::Mutex; | ||
| use tokio::time::{self, Duration, Instant}; | ||
|
|
||
| /// A logical clock that can be paused | ||
| #[derive(Debug, Clone, Default)] | ||
| pub struct Clock { | ||
| inner: Arc<Mutex<InnerClock>>, | ||
| } | ||
|
|
||
| #[derive(Debug, Clone, Default)] | ||
| pub(crate) struct InnerClock { | ||
| status: Status, | ||
| elapsed: Duration, | ||
| } | ||
|
|
||
| #[derive(Debug, Clone, Copy, Default)] | ||
| pub(crate) enum Status { | ||
| #[default] | ||
| Paused, | ||
| Running(Instant), | ||
| } | ||
|
|
||
| impl Clock { | ||
| pub fn start_at(instant: Instant) -> Self { | ||
| let inner = InnerClock { | ||
| status: Status::Running(instant), | ||
| elapsed: Duration::default(), | ||
| }; | ||
| Self { inner: Arc::new(Mutex::new(inner)) } | ||
| } | ||
|
|
||
| pub fn resume(&mut self) { | ||
| let mut inner = self.inner.lock(); | ||
| if let Status::Paused = inner.status { | ||
| inner.status = Status::Running(Instant::now()); | ||
| } | ||
| } | ||
|
|
||
| pub fn pause(&mut self) { | ||
| let mut inner = self.inner.lock(); | ||
| if let Status::Running(checkpoint) = inner.status { | ||
| inner.elapsed += checkpoint.elapsed(); | ||
| inner.status = Status::Paused; | ||
| } | ||
| } | ||
|
|
||
| pub fn elapsed(&self) -> Duration { | ||
| let inner = self.inner.lock(); | ||
| match inner.status { | ||
| Status::Paused => inner.elapsed, | ||
| Status::Running(checkpoint) => inner.elapsed + checkpoint.elapsed(), | ||
| } | ||
| } | ||
|
|
||
| pub async fn sleep(&self, mut duration: Duration) { | ||
| let wake_time = self.elapsed() + duration; | ||
| loop { | ||
| time::sleep(duration).await; | ||
| let elapsed = self.elapsed(); | ||
| if elapsed >= wake_time { | ||
| break; | ||
| } | ||
| duration = wake_time - elapsed; | ||
| } | ||
| } | ||
|
|
||
| async fn sleep_until(&self, deadline: Duration) { | ||
| let now = self.elapsed(); | ||
| if deadline <= now { | ||
| return; | ||
| } | ||
| self.sleep(deadline - now).await; | ||
| } | ||
|
|
||
| pub fn ticker(&self, duration: Duration) -> Ticker { | ||
| Ticker::new(self.clone(), duration) | ||
| } | ||
| } | ||
|
|
||
| /// A ticker that ticks at a fixed logical interval | ||
| #[derive(Debug, Clone)] | ||
| pub struct Ticker { | ||
| clock: Clock, | ||
| interval: Duration, | ||
| next_tick: Duration, | ||
| } | ||
|
|
||
| impl Ticker { | ||
| pub fn new(clock: Clock, duration: Duration) -> Self { | ||
| Self { clock, interval: duration, next_tick: duration } | ||
| } | ||
|
|
||
| pub async fn tick(&mut self) { | ||
| self.clock.sleep_until(self.next_tick).await; | ||
| self.next_tick += self.interval; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.