mohr/nexosim
1
0
Fork 0
forked from ROMEO/nexosim
Code Pull Requests Activity

Overload schedule_*event methods

Browse Source 
The `schedule_*event_in` and `schedule_*event_at` pairs of methods are
each merged into a single overloaded method accepting either a relative
`Duration`or an absolute `MonotonicTime`.
This commit is contained in:
Serge Barral 2023-08-14 15:47:08 +02:00
parent a036630c4e
commit 22516fe190
9 changed files with 132 additions and 418 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -88,7 +88,7 @@ pub struct DelayedMultiplier {
impl DelayedMultiplier { impl DelayedMultiplier {
pub fn input(&mut self, value: f64, scheduler: &Scheduler<Self>) { pub fn input(&mut self, value: f64, scheduler: &Scheduler<Self>) {
scheduler scheduler
.schedule_event_in(Duration::from_secs(1), Self::send, 2.0 * value) .schedule_event(Duration::from_secs(1), Self::send, 2.0 * value)
.unwrap(); .unwrap();
} }
async fn send(&mut self, value: f64) { async fn send(&mut self, value: f64) {

6
asynchronix/examples/espresso_machine.rs
View File

@ -141,7 +141,7 @@ impl Controller {
// Schedule the `stop_brew()` method and turn on the pump. // Schedule the `stop_brew()` method and turn on the pump.
self.stop_brew_key = Some( self.stop_brew_key = Some(
scheduler scheduler
.schedule_keyed_event_in(self.brew_time, Self::stop_brew, ()) .schedule_keyed_event(self.brew_time, Self::stop_brew, ())
.unwrap(), .unwrap(),
); );
self.pump_cmd.send(PumpCommand::On).await; self.pump_cmd.send(PumpCommand::On).await;
@ -274,7 +274,7 @@ impl Tank {
let duration_until_empty = Duration::from_secs_f64(duration_until_empty); let duration_until_empty = Duration::from_secs_f64(duration_until_empty);
// Schedule the next update. // Schedule the next update.
match scheduler.schedule_keyed_event_in(duration_until_empty, Self::set_empty, ()) { match scheduler.schedule_keyed_event(duration_until_empty, Self::set_empty, ()) {
Ok(set_empty_key) => { Ok(set_empty_key) => {
let state = TankDynamicState { let state = TankDynamicState {
last_volume_update: time, last_volume_update: time,
@ -431,7 +431,7 @@ fn main() {
assert_eq!(flow_rate.take(), Some(0.0)); assert_eq!(flow_rate.take(), Some(0.0));
// Interrupt the brew after 15s by pressing again the brew button. // Interrupt the brew after 15s by pressing again the brew button.
simu.schedule_event_in( simu.schedule_event(
Duration::from_secs(15), Duration::from_secs(15),
Controller::brew_cmd, Controller::brew_cmd,
(), (),

4
asynchronix/examples/stepper_motor.rs
View File

@ -174,7 +174,7 @@ impl Driver {
// Schedule the next pulse. // Schedule the next pulse.
scheduler scheduler
.schedule_event_in(pulse_duration, Self::send_pulse, ()) .schedule_event(pulse_duration, Self::send_pulse, ())
.unwrap(); .unwrap();
} }
} }
@ -224,7 +224,7 @@ fn main() {
assert!(position.next().is_none()); assert!(position.next().is_none());
// Start the motor in 2s with a PPS of 10Hz. // Start the motor in 2s with a PPS of 10Hz.
simu.schedule_event_in( simu.schedule_event(
Duration::from_secs(2), Duration::from_secs(2),
Driver::pulse_rate, Driver::pulse_rate,
10.0, 10.0,

16
asynchronix/src/lib.rs
View File

@ -113,7 +113,7 @@
//! } //! }
//! impl Delay { //! impl Delay {
//! pub fn input(&mut self, value: f64, scheduler: &Scheduler<Self>) { //! pub fn input(&mut self, value: f64, scheduler: &Scheduler<Self>) {
//! scheduler.schedule_event_in(Duration::from_secs(1), Self::send, value).unwrap(); //! scheduler.schedule_event(Duration::from_secs(1), Self::send, value).unwrap();
//! } //! }
//! //!
//! async fn send(&mut self, value: f64) { //! async fn send(&mut self, value: f64) {
@ -184,7 +184,7 @@
//! # } //! # }
//! # impl Delay { //! # impl Delay {
//! # pub fn input(&mut self, value: f64, scheduler: &Scheduler<Self>) { //! # pub fn input(&mut self, value: f64, scheduler: &Scheduler<Self>) {
//! # scheduler.schedule_event_in(Duration::from_secs(1), Self::send, value).unwrap(); //! # scheduler.schedule_event(Duration::from_secs(1), Self::send, value).unwrap();
//! # } //! # }
//! # async fn send(&mut self, value: f64) { // this method can be private //! # async fn send(&mut self, value: f64) { // this method can be private
//! # self.output.send(value).await; //! # self.output.send(value).await;
@ -242,7 +242,7 @@
//! [`Simulation::send_event()`](simulation::Simulation::send_event) or //! [`Simulation::send_event()`](simulation::Simulation::send_event) or
//! [`Simulation::send_query()`](simulation::Simulation::send_query), //! [`Simulation::send_query()`](simulation::Simulation::send_query),
//! 3. by scheduling events, using for instance //! 3. by scheduling events, using for instance
//! [`Simulation::schedule_event_in()`](simulation::Simulation::schedule_event_in). //! [`Simulation::schedule_event()`](simulation::Simulation::schedule_event).
//! //!
//! Simulation outputs can be monitored using //! Simulation outputs can be monitored using
//! [`EventSlot`](simulation::EventSlot)s and //! [`EventSlot`](simulation::EventSlot)s and
@ -275,7 +275,7 @@
//! # } //! # }
//! # impl Delay { //! # impl Delay {
//! # pub fn input(&mut self, value: f64, scheduler: &Scheduler<Self>) { //! # pub fn input(&mut self, value: f64, scheduler: &Scheduler<Self>) {
//! # scheduler.schedule_event_in(Duration::from_secs(1), Self::send, value).unwrap(); //! # scheduler.schedule_event(Duration::from_secs(1), Self::send, value).unwrap();
//! # } //! # }
//! # async fn send(&mut self, value: f64) { // this method can be private //! # async fn send(&mut self, value: f64) { // this method can be private
//! # self.output.send(value).await; //! # self.output.send(value).await;
@ -354,10 +354,10 @@
//! //!
//! The first guarantee (and only the first) also extends to events scheduled //! The first guarantee (and only the first) also extends to events scheduled
//! from a simulation with a //! from a simulation with a
//! [`Simulation::schedule_*()`](simulation::Simulation::schedule_event_at) //! [`Simulation::schedule_*()`](simulation::Simulation::schedule_event) method:
//! method: if the scheduler contains several events to be delivered at the same //! if the scheduler contains several events to be delivered at the same time to
//! time to the same model, these events will always be processed in the order //! the same model, these events will always be processed in the order in which
//! in which they were scheduled. //! they were scheduled.
//! //!
//! [actor_model]: https://en.wikipedia.org/wiki/Actor_model //! [actor_model]: https://en.wikipedia.org/wiki/Actor_model
//! [pony]: https://www.ponylang.io/ //! [pony]: https://www.ponylang.io/

207
asynchronix/src/simulation.rs
View File

@ -137,7 +137,7 @@ use recycle_box::{coerce_box, RecycleBox};
use crate::executor::Executor; use crate::executor::Executor;
use crate::model::{InputFn, Model, ReplierFn}; use crate::model::{InputFn, Model, ReplierFn};
use crate::time::{ use crate::time::{
self, EventKey, MonotonicTime, ScheduledEvent, SchedulerQueue, SchedulingError, self, Deadline, EventKey, MonotonicTime, ScheduledEvent, SchedulerQueue, SchedulingError,
TearableAtomicTime, TearableAtomicTime,
}; };
use crate::util::futures::SeqFuture; use crate::util::futures::SeqFuture;
@ -156,8 +156,8 @@ use crate::util::sync_cell::SyncCell;
/// itself, but also from models via the optional /// itself, but also from models via the optional
/// [`&Scheduler`][time::Scheduler] argument of input and replier port methods. /// [`&Scheduler`][time::Scheduler] argument of input and replier port methods.
/// Likewise, simulation time can be accessed with the [`Simulation::time()`] /// Likewise, simulation time can be accessed with the [`Simulation::time()`]
/// method, or from models with the /// method, or from models with the [`Scheduler::time()`](time::Scheduler::time)
/// [`Scheduler::time()`](time::Scheduler::time) method. /// method.
/// ///
/// Events and queries can be scheduled immediately, *i.e.* for the current /// Events and queries can be scheduled immediately, *i.e.* for the current
/// simulation time, using [`send_event()`](Simulation::send_event) and /// simulation time, using [`send_event()`](Simulation::send_event) and
@ -166,8 +166,8 @@ use crate::util::sync_cell::SyncCell;
/// the case of queries, the response is returned. /// the case of queries, the response is returned.
/// ///
/// Events can also be scheduled at a future simulation time using one of the /// Events can also be scheduled at a future simulation time using one of the
/// [`schedule_*()`](Simulation::schedule_event_at) method. These methods queue /// [`schedule_*()`](Simulation::schedule_event) method. These methods queue an
/// an event without blocking. /// event without blocking.
/// ///
/// Finally, the [`Simulation`] instance manages simulation time. Calling /// Finally, the [`Simulation`] instance manages simulation time. Calling
/// [`step()`](Simulation::step) will increment simulation time until that of /// [`step()`](Simulation::step) will increment simulation time until that of
@ -247,10 +247,10 @@ impl Simulation {
/// Events scheduled for the same time and targeting the same model are /// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order. /// guaranteed to be processed according to the scheduling order.
/// ///
/// See also: [`time::Scheduler::schedule_event_at`]. /// See also: [`time::Scheduler::schedule_event`].
pub fn schedule_event_at<M, F, T, S>( pub fn schedule_event<M, F, T, S>(
&mut self, &mut self,
time: MonotonicTime, deadline: impl Deadline,
func: F, func: F,
arg: T, arg: T,
address: impl Into<Address<M>>, address: impl Into<Address<M>>,
@ -261,7 +261,9 @@ impl Simulation {
T: Send + Clone + 'static, T: Send + Clone + 'static,
S: Send + 'static, S: Send + 'static,
{ {
if self.time.read() >= time { let now = self.time();
let time = deadline.into_time(now);
if now >= time {
return Err(SchedulingError::InvalidScheduledTime); return Err(SchedulingError::InvalidScheduledTime);
} }
time::schedule_event_at_unchecked(time, func, arg, address.into().0, &self.scheduler_queue); time::schedule_event_at_unchecked(time, func, arg, address.into().0, &self.scheduler_queue);
@ -269,37 +271,6 @@ impl Simulation {
Ok(()) Ok(())
} }
/// Schedules an event at the lapse of the specified duration.
///
/// An error is returned if the specified delay is null.
///
/// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order.
///
/// See also: [`time::Scheduler::schedule_event_in`].
pub fn schedule_event_in<M, F, T, S>(
&mut self,
delay: Duration,
func: F,
arg: T,
address: impl Into<Address<M>>,
) -> Result<(), SchedulingError>
where
M: Model,
F: for<'a> InputFn<'a, M, T, S>,
T: Send + Clone + 'static,
S: Send + 'static,
{
if delay.is_zero() {
return Err(SchedulingError::InvalidScheduledTime);
}
let time = self.time.read() + delay;
time::schedule_event_at_unchecked(time, func, arg, address.into().0, &self.scheduler_queue);
Ok(())
}
/// Schedules a cancellable event at a future time and returns an event key. /// Schedules a cancellable event at a future time and returns an event key.
/// ///
/// An error is returned if the specified time is not in the future of the /// An error is returned if the specified time is not in the future of the
@ -308,10 +279,10 @@ impl Simulation {
/// Events scheduled for the same time and targeting the same model are /// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order. /// guaranteed to be processed according to the scheduling order.
/// ///
/// See also: [`time::Scheduler::schedule_keyed_event_at`]. /// See also: [`time::Scheduler::schedule_keyed_event`].
pub fn schedule_keyed_event_at<M, F, T, S>( pub fn schedule_keyed_event<M, F, T, S>(
&mut self, &mut self,
time: MonotonicTime, deadline: impl Deadline,
func: F, func: F,
arg: T, arg: T,
address: impl Into<Address<M>>, address: impl Into<Address<M>>,
@ -322,7 +293,9 @@ impl Simulation {
T: Send + Clone + 'static, T: Send + Clone + 'static,
S: Send + 'static, S: Send + 'static,
{ {
if self.time.read() >= time { let now = self.time();
let time = deadline.into_time(now);
if now >= time {
return Err(SchedulingError::InvalidScheduledTime); return Err(SchedulingError::InvalidScheduledTime);
} }
let event_key = time::schedule_keyed_event_at_unchecked( let event_key = time::schedule_keyed_event_at_unchecked(
@ -336,44 +309,6 @@ impl Simulation {
Ok(event_key) Ok(event_key)
} }
/// Schedules a cancellable event at the lapse of the specified duration and
/// returns an event key.
///
/// An error is returned if the specified delay is null.
///
/// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order.
///
/// See also: [`time::Scheduler::schedule_keyed_event_in`].
pub fn schedule_keyed_event_in<M, F, T, S>(
&mut self,
delay: Duration,
func: F,
arg: T,
address: impl Into<Address<M>>,
) -> Result<EventKey, SchedulingError>
where
M: Model,
F: for<'a> InputFn<'a, M, T, S>,
T: Send + Clone + 'static,
S: Send + 'static,
{
if delay.is_zero() {
return Err(SchedulingError::InvalidScheduledTime);
}
let time = self.time.read() + delay;
let event_key = time::schedule_keyed_event_at_unchecked(
time,
func,
arg,
address.into().0,
&self.scheduler_queue,
);
Ok(event_key)
}
/// Schedules a periodically recurring event at a future time. /// Schedules a periodically recurring event at a future time.
/// ///
/// An error is returned if the specified time is not in the future of the /// An error is returned if the specified time is not in the future of the
@ -382,10 +317,10 @@ impl Simulation {
/// Events scheduled for the same time and targeting the same model are /// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order. /// guaranteed to be processed according to the scheduling order.
/// ///
/// See also: [`time::Scheduler::schedule_periodic_event_at`]. /// See also: [`time::Scheduler::schedule_periodic_event`].
pub fn schedule_periodic_event_at<M, F, T, S>( pub fn schedule_periodic_event<M, F, T, S>(
&mut self, &mut self,
time: MonotonicTime, deadline: impl Deadline,
period: Duration, period: Duration,
func: F, func: F,
arg: T, arg: T,
@ -397,7 +332,9 @@ impl Simulation {
T: Send + Clone + 'static, T: Send + Clone + 'static,
S: Send + 'static, S: Send + 'static,
{ {
if self.time.read() >= time { let now = self.time();
let time = deadline.into_time(now);
if now >= time {
return Err(SchedulingError::InvalidScheduledTime); return Err(SchedulingError::InvalidScheduledTime);
} }
if period.is_zero() { if period.is_zero() {
@ -415,50 +352,6 @@ impl Simulation {
Ok(()) Ok(())
} }
/// Schedules a periodically recurring event at the lapse of the specified
/// duration.
///
/// An error is returned if the specified delay or the specified period are
/// null.
///
/// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order.
///
/// See also: [`time::Scheduler::schedule_periodic_event_in`].
pub fn schedule_periodic_event_in<M, F, T, S>(
&mut self,
delay: Duration,
period: Duration,
func: F,
arg: T,
address: impl Into<Address<M>>,
) -> Result<(), SchedulingError>
where
M: Model,
F: for<'a> InputFn<'a, M, T, S> + Clone,
T: Send + Clone + 'static,
S: Send + 'static,
{
if delay.is_zero() {
return Err(SchedulingError::InvalidScheduledTime);
}
if period.is_zero() {
return Err(SchedulingError::NullRepetitionPeriod);
}
let time = self.time.read() + delay;
time::schedule_periodic_event_at_unchecked(
time,
period,
func,
arg,
address.into().0,
&self.scheduler_queue,
);
Ok(())
}
/// Schedules a cancellable, periodically recurring event at a future time /// Schedules a cancellable, periodically recurring event at a future time
/// and returns an event key. /// and returns an event key.
/// ///
@ -468,10 +361,10 @@ impl Simulation {
/// Events scheduled for the same time and targeting the same model are /// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order. /// guaranteed to be processed according to the scheduling order.
/// ///
/// See also: [`time::Scheduler::schedule_periodic_keyed_event_at`]. /// See also: [`time::Scheduler::schedule_keyed_periodic_event`].
pub fn schedule_periodic_keyed_event_at<M, F, T, S>( pub fn schedule_keyed_periodic_event<M, F, T, S>(
&mut self, &mut self,
time: MonotonicTime, deadline: impl Deadline,
period: Duration, period: Duration,
func: F, func: F,
arg: T, arg: T,
@ -483,7 +376,9 @@ impl Simulation {
T: Send + Clone + 'static, T: Send + Clone + 'static,
S: Send + 'static, S: Send + 'static,
{ {
if self.time.read() >= time { let now = self.time();
let time = deadline.into_time(now);
if now >= time {
return Err(SchedulingError::InvalidScheduledTime); return Err(SchedulingError::InvalidScheduledTime);
} }
if period.is_zero() { if period.is_zero() {
@ -501,50 +396,6 @@ impl Simulation {
Ok(event_key) Ok(event_key)
} }
/// Schedules a cancellable, periodically recurring event at the lapse of
/// the specified duration and returns an event key.
///
/// An error is returned if the specified delay or the specified period are
/// null.
///
/// Events scheduled for the same time and targeting the same model are
/// guaranteed to be processed according to the scheduling order.
///
/// See also: [`time::Scheduler::schedule_periodic_keyed_event_in`].
pub fn schedule_periodic_keyed_event_in<M, F, T, S>(
&mut self,
delay: Duration,
period: Duration,
func: F,
arg: T,
address: impl Into<Address<M>>,
) -> Result<EventKey, SchedulingError>
where
M: Model,
F: for<'a> InputFn<'a, M, T, S> + Clone,
T: Send + Clone + 'static,
S: Send + 'static,
{
if delay.is_zero() {
return Err(SchedulingError::InvalidScheduledTime);
}
if period.is_zero() {
return Err(SchedulingError::NullRepetitionPeriod);
}
let time = self.time.read() + delay;
let event_key = time::schedule_periodic_keyed_event_at_unchecked(
time,
period,
func,
arg,
address.into().0,
&self.scheduler_queue,
);
Ok(event_key)
}
/// Sends and processes an event, blocking until completion. /// Sends and processes an event, blocking until completion.
/// ///
/// Simulation time remains unchanged. /// Simulation time remains unchanged.

4
asynchronix/src/time.rs
View File

@ -31,7 +31,7 @@
//! //!
//! // Sets an alarm [input port]. //! // Sets an alarm [input port].
//! pub fn set(&mut self, setting: MonotonicTime, scheduler: &Scheduler<Self>) { //! pub fn set(&mut self, setting: MonotonicTime, scheduler: &Scheduler<Self>) {
//! if scheduler.schedule_event_at(setting, Self::ring, ()).is_err() { //! if scheduler.schedule_event(setting, Self::ring, ()).is_err() {
//! println!("The alarm clock can only be set for a future time"); //! println!("The alarm clock can only be set for a future time");
//! } //! }
//! } //! }
@ -55,4 +55,4 @@ pub(crate) use scheduler::{
schedule_periodic_event_at_unchecked, schedule_periodic_keyed_event_at_unchecked, schedule_periodic_event_at_unchecked, schedule_periodic_keyed_event_at_unchecked,
ScheduledEvent, SchedulerQueue, ScheduledEvent, SchedulerQueue,
}; };
pub use scheduler::{EventKey, Scheduler, SchedulingError}; pub use scheduler::{Deadline, EventKey, Scheduler, SchedulingError};

271
asynchronix/src/time/scheduler.rs
View File

@ -23,6 +23,30 @@ use crate::util::sync_cell::SyncCellReader;
/// Shorthand for the scheduler queue type. /// Shorthand for the scheduler queue type.
pub(crate) type SchedulerQueue = PriorityQueue<(MonotonicTime, ChannelId), Box<dyn ScheduledEvent>>; pub(crate) type SchedulerQueue = PriorityQueue<(MonotonicTime, ChannelId), Box<dyn ScheduledEvent>>;
/// Trait abstracting over time-absolute and time-relative deadlines.
///
/// This trait is implemented by [`std::time::Duration`] and
/// [`MonotonicTime`].
pub trait Deadline {
/// Make this deadline into an absolute timestamp, using the provided
/// current time as a reference.
fn into_time(self, now: MonotonicTime) -> MonotonicTime;
}
impl Deadline for Duration {
#[inline(always)]
fn into_time(self, now: MonotonicTime) -> MonotonicTime {
now + self
}
}
impl Deadline for MonotonicTime {
#[inline(always)]
fn into_time(self, _: MonotonicTime) -> MonotonicTime {
self
}
}
/// A local scheduler for models. /// A local scheduler for models.
/// ///
/// A `Scheduler` is a handle to the global scheduler associated to a model /// A `Scheduler` is a handle to the global scheduler associated to a model
@ -73,7 +97,7 @@ pub(crate) type SchedulerQueue = PriorityQueue<(MonotonicTime, ChannelId), Box<d
/// if delay.is_zero() { /// if delay.is_zero() {
/// self.msg_out.send(greeting).await; /// self.msg_out.send(greeting).await;
/// } else { /// } else {
/// scheduler.schedule_event_in(delay, Self::send_msg, greeting).unwrap(); /// scheduler.schedule_event(delay, Self::send_msg, greeting).unwrap();
/// } /// }
/// } /// }
/// ///
@ -127,87 +151,39 @@ impl<M: Model> Scheduler<M> {
/// Schedules an event at a future time. /// Schedules an event at a future time.
/// ///
/// An error is returned if the specified time is not in the future of the /// An error is returned if the specified deadline is not in the future of
/// current simulation time. /// the current simulation time.
///
/// # Examples
///
/// ```
/// use asynchronix::model::Model;
/// use asynchronix::time::{MonotonicTime, Scheduler};
///
/// // An alarm clock.
/// pub struct AlarmClock {}
///
/// impl AlarmClock {
/// // Sets an alarm [input port].
/// pub fn set(&mut self, setting: MonotonicTime, scheduler: &Scheduler<Self>) {
/// if scheduler.schedule_event_at(setting, Self::ring, ()).is_err() {
/// println!("The alarm clock can only be set for a future time");
/// }
/// }
///
/// // Rings the alarm [private input port].
/// fn ring(&mut self) {
/// println!("Brringggg");
/// }
/// }
///
/// impl Model for AlarmClock {}
/// ```
pub fn schedule_event_at<F, T, S>(
&self,
time: MonotonicTime,
func: F,
arg: T,
) -> Result<(), SchedulingError>
where
F: for<'a> InputFn<'a, M, T, S>,
T: Send + Clone + 'static,
S: Send + 'static,
{
if self.time() >= time {
return Err(SchedulingError::InvalidScheduledTime);
}
let sender = self.sender.clone();
schedule_event_at_unchecked(time, func, arg, sender, &self.scheduler_queue);
Ok(())
}
/// Schedules an event at the lapse of the specified duration.
///
/// An error is returned if the specified delay is null.
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```
/// use std::time::Duration; /// use std::time::Duration;
/// use std::future::Future; ///
/// use asynchronix::model::Model; /// use asynchronix::model::Model;
/// use asynchronix::time::Scheduler; /// use asynchronix::time::Scheduler;
/// ///
/// // A model that logs the value of a counter every second after being /// // A timer.
/// // triggered the first time. /// pub struct Timer {}
/// pub struct CounterLogger {}
/// ///
/// impl CounterLogger { /// impl Timer {
/// // Triggers the logging of a timestamp every second [input port]. /// // Sets an alarm [input port].
/// pub fn trigger(&mut self, counter: u64, scheduler: &Scheduler<Self>) { /// pub fn set(&mut self, setting: Duration, scheduler: &Scheduler<Self>) {
/// println!("counter: {}", counter); /// if scheduler.schedule_event(setting, Self::ring, ()).is_err() {
/// /// println!("The alarm clock can only be set for a future time");
/// // Schedule this method again in 1s with an incremented counter.
/// scheduler
/// .schedule_event_in(Duration::from_secs(1), Self::trigger, counter + 1)
/// .unwrap();
/// } /// }
/// } /// }
/// ///
/// impl Model for CounterLogger {} /// // Rings [private input port].
/// fn ring(&mut self) {
/// println!("Brringggg");
/// }
/// }
///
/// impl Model for Timer {}
/// ``` /// ```
pub fn schedule_event_in<F, T, S>( pub fn schedule_event<F, T, S>(
&self, &self,
delay: Duration, deadline: impl Deadline,
func: F, func: F,
arg: T, arg: T,
) -> Result<(), SchedulingError> ) -> Result<(), SchedulingError>
@ -216,10 +192,11 @@ impl<M: Model> Scheduler<M> {
T: Send + Clone + 'static, T: Send + Clone + 'static,
S: Send + 'static, S: Send + 'static,
{ {
if delay.is_zero() { let now = self.time();
let time = deadline.into_time(now);
if now >= time {
return Err(SchedulingError::InvalidScheduledTime); return Err(SchedulingError::InvalidScheduledTime);
} }
let time = self.time() + delay;
let sender = self.sender.clone(); let sender = self.sender.clone();
schedule_event_at_unchecked(time, func, arg, sender, &self.scheduler_queue); schedule_event_at_unchecked(time, func, arg, sender, &self.scheduler_queue);
@ -228,8 +205,8 @@ impl<M: Model> Scheduler<M> {
/// Schedules a cancellable event at a future time and returns an event key. /// Schedules a cancellable event at a future time and returns an event key.
/// ///
/// An error is returned if the specified time is not in the future of the /// An error is returned if the specified deadline is not in the future of
/// current simulation time. /// the current simulation time.
/// ///
/// # Examples /// # Examples
/// ///
@ -247,7 +224,7 @@ impl<M: Model> Scheduler<M> {
/// // Sets an alarm [input port]. /// // Sets an alarm [input port].
/// pub fn set(&mut self, setting: MonotonicTime, scheduler: &Scheduler<Self>) { /// pub fn set(&mut self, setting: MonotonicTime, scheduler: &Scheduler<Self>) {
/// self.cancel(); /// self.cancel();
/// match scheduler.schedule_keyed_event_at(setting, Self::ring, ()) { /// match scheduler.schedule_keyed_event(setting, Self::ring, ()) {
/// Ok(event_key) => self.event_key = Some(event_key), /// Ok(event_key) => self.event_key = Some(event_key),
/// Err(_) => println!("The alarm clock can only be set for a future time"), /// Err(_) => println!("The alarm clock can only be set for a future time"),
/// }; /// };
@ -266,9 +243,9 @@ impl<M: Model> Scheduler<M> {
/// ///
/// impl Model for CancellableAlarmClock {} /// impl Model for CancellableAlarmClock {}
/// ``` /// ```
pub fn schedule_keyed_event_at<F, T, S>( pub fn schedule_keyed_event<F, T, S>(
&self, &self,
time: MonotonicTime, deadline: impl Deadline,
func: F, func: F,
arg: T, arg: T,
) -> Result<EventKey, SchedulingError> ) -> Result<EventKey, SchedulingError>
@ -277,7 +254,9 @@ impl<M: Model> Scheduler<M> {
T: Send + Clone + 'static, T: Send + Clone + 'static,
S: Send + 'static, S: Send + 'static,
{ {
if self.time() >= time { let now = self.time();
let time = deadline.into_time(now);
if now >= time {
return Err(SchedulingError::InvalidScheduledTime); return Err(SchedulingError::InvalidScheduledTime);
} }
let sender = self.sender.clone(); let sender = self.sender.clone();
@ -287,40 +266,10 @@ impl<M: Model> Scheduler<M> {
Ok(event_key) Ok(event_key)
} }
/// Schedules a cancellable event at the lapse of the specified duration and
/// returns an event key.
///
/// An error is returned if the specified delay is null.
///
/// See also:
/// [`schedule_keyed_event_at`][Scheduler::schedule_keyed_event_at],
/// [`schedule_event_in`][Scheduler::schedule_event_in].
pub fn schedule_keyed_event_in<F, T, S>(
&self,
delay: Duration,
func: F,
arg: T,
) -> Result<EventKey, SchedulingError>
where
F: for<'a> InputFn<'a, M, T, S>,
T: Send + Clone + 'static,
S: Send + 'static,
{
if delay.is_zero() {
return Err(SchedulingError::InvalidScheduledTime);
}
let time = self.time() + delay;
let sender = self.sender.clone();
let event_key =
schedule_keyed_event_at_unchecked(time, func, arg, sender, &self.scheduler_queue);
Ok(event_key)
}
/// Schedules a periodically recurring event at a future time. /// Schedules a periodically recurring event at a future time.
/// ///
/// An error is returned if the specified time is not in the future of the /// An error is returned if the specified deadline is not in the future of
/// current simulation time or if the specified period is null. /// the current simulation time or if the specified period is null.
/// ///
/// # Examples /// # Examples
/// ///
@ -336,7 +285,7 @@ impl<M: Model> Scheduler<M> {
/// impl BeepingAlarmClock { /// impl BeepingAlarmClock {
/// // Sets an alarm [input port]. /// // Sets an alarm [input port].
/// pub fn set(&mut self, setting: MonotonicTime, scheduler: &Scheduler<Self>) { /// pub fn set(&mut self, setting: MonotonicTime, scheduler: &Scheduler<Self>) {
/// if scheduler.schedule_periodic_event_at( /// if scheduler.schedule_periodic_event(
/// setting, /// setting,
/// Duration::from_secs(1), // 1Hz = 1/1s /// Duration::from_secs(1), // 1Hz = 1/1s
/// Self::beep, /// Self::beep,
@ -354,9 +303,9 @@ impl<M: Model> Scheduler<M> {
/// ///
/// impl Model for BeepingAlarmClock {} /// impl Model for BeepingAlarmClock {}
/// ``` /// ```
pub fn schedule_periodic_event_at<F, T, S>( pub fn schedule_periodic_event<F, T, S>(
&self, &self,
time: MonotonicTime, deadline: impl Deadline,
period: Duration, period: Duration,
func: F, func: F,
arg: T, arg: T,
@ -366,7 +315,9 @@ impl<M: Model> Scheduler<M> {
T: Send + Clone + 'static, T: Send + Clone + 'static,
S: Send + 'static, S: Send + 'static,
{ {
if self.time() >= time { let now = self.time();
let time = deadline.into_time(now);
if now >= time {
return Err(SchedulingError::InvalidScheduledTime); return Err(SchedulingError::InvalidScheduledTime);
} }
if period.is_zero() { if period.is_zero() {
@ -385,52 +336,11 @@ impl<M: Model> Scheduler<M> {
Ok(()) Ok(())
} }
/// Schedules a periodically recurring event at the lapse of the specified
/// duration.
///
/// An error is returned if the specified delay or the specified period are
/// null.
///
/// See also:
/// [`schedule_periodic_event_at`][Scheduler::schedule_periodic_event_at],
/// [`schedule_event_in`][Scheduler::schedule_event_in].
pub fn schedule_periodic_event_in<F, T, S>(
&self,
delay: Duration,
period: Duration,
func: F,
arg: T,
) -> Result<(), SchedulingError>
where
F: for<'a> InputFn<'a, M, T, S> + Clone,
T: Send + Clone + 'static,
S: Send + 'static,
{
if delay.is_zero() {
return Err(SchedulingError::InvalidScheduledTime);
}
if period.is_zero() {
return Err(SchedulingError::NullRepetitionPeriod);
}
let time = self.time() + delay;
let sender = self.sender.clone();
schedule_periodic_event_at_unchecked(
time,
period,
func,
arg,
sender,
&self.scheduler_queue,
);
Ok(())
}
/// Schedules a cancellable, periodically recurring event at a future time /// Schedules a cancellable, periodically recurring event at a future time
/// and returns an event key. /// and returns an event key.
/// ///
/// An error is returned if the specified time is not in the future of the /// An error is returned if the specified deadline is not in the future of
/// current simulation time or if the specified period is null. /// the current simulation time or if the specified period is null.
/// ///
/// # Examples /// # Examples
/// ///
@ -451,7 +361,7 @@ impl<M: Model> Scheduler<M> {
/// // Sets an alarm [input port]. /// // Sets an alarm [input port].
/// pub fn set(&mut self, setting: MonotonicTime, scheduler: &Scheduler<Self>) { /// pub fn set(&mut self, setting: MonotonicTime, scheduler: &Scheduler<Self>) {
/// self.cancel(); /// self.cancel();
/// match scheduler.schedule_periodic_keyed_event_at( /// match scheduler.schedule_keyed_periodic_event(
/// setting, /// setting,
/// Duration::from_secs(1), // 1Hz = 1/1s /// Duration::from_secs(1), // 1Hz = 1/1s
/// Self::beep, /// Self::beep,
@ -475,9 +385,9 @@ impl<M: Model> Scheduler<M> {
/// ///
/// impl Model for CancellableBeepingAlarmClock {} /// impl Model for CancellableBeepingAlarmClock {}
/// ``` /// ```
pub fn schedule_periodic_keyed_event_at<F, T, S>( pub fn schedule_keyed_periodic_event<F, T, S>(
&self, &self,
time: MonotonicTime, deadline: impl Deadline,
period: Duration, period: Duration,
func: F, func: F,
arg: T, arg: T,
@ -487,7 +397,9 @@ impl<M: Model> Scheduler<M> {
T: Send + Clone + 'static, T: Send + Clone + 'static,
S: Send + 'static, S: Send + 'static,
{ {
if self.time() >= time { let now = self.time();
let time = deadline.into_time(now);
if now >= time {
return Err(SchedulingError::InvalidScheduledTime); return Err(SchedulingError::InvalidScheduledTime);
} }
if period.is_zero() { if period.is_zero() {
@ -505,47 +417,6 @@ impl<M: Model> Scheduler<M> {
Ok(event_key) Ok(event_key)
} }
/// Schedules a cancellable, periodically recurring event at the lapse of
/// the specified duration and returns an event key.
///
/// An error is returned if the specified delay or the specified period are
/// null.
///
/// See also:
/// [`schedule_periodic_keyed_event_at`][Scheduler::schedule_periodic_keyed_event_at],
/// [`schedule_event_in`][Scheduler::schedule_event_in].
pub fn schedule_periodic_keyed_event_in<F, T, S>(
&self,
delay: Duration,
period: Duration,
func: F,
arg: T,
) -> Result<EventKey, SchedulingError>
where
F: for<'a> InputFn<'a, M, T, S> + Clone,
T: Send + Clone + 'static,
S: Send + 'static,
{
if delay.is_zero() {
return Err(SchedulingError::InvalidScheduledTime);
}
if period.is_zero() {
return Err(SchedulingError::NullRepetitionPeriod);
}
let time = self.time() + delay;
let sender = self.sender.clone();
let event_key = schedule_periodic_keyed_event_at_unchecked(
time,
period,
func,
arg,
sender,
&self.scheduler_queue,
);
Ok(event_key)
}
} }
impl<M: Model> fmt::Debug for Scheduler<M> { impl<M: Model> fmt::Debug for Scheduler<M> {

22
asynchronix/tests/model_scheduling.rs
View File

@ -15,7 +15,7 @@ fn model_schedule_event() {
impl TestModel { impl TestModel {
fn trigger(&mut self, _: (), scheduler: &Scheduler<Self>) { fn trigger(&mut self, _: (), scheduler: &Scheduler<Self>) {
scheduler scheduler
.schedule_event_at(scheduler.time() + Duration::from_secs(2), Self::action, ()) .schedule_event(scheduler.time() + Duration::from_secs(2), Self::action, ())
.unwrap(); .unwrap();
} }
async fn action(&mut self) { async fn action(&mut self) {
@ -51,14 +51,10 @@ fn model_cancel_future_keyed_event() {
impl TestModel { impl TestModel {
fn trigger(&mut self, _: (), scheduler: &Scheduler<Self>) { fn trigger(&mut self, _: (), scheduler: &Scheduler<Self>) {
scheduler scheduler
.schedule_event_at(scheduler.time() + Duration::from_secs(1), Self::action1, ()) .schedule_event(scheduler.time() + Duration::from_secs(1), Self::action1, ())
.unwrap(); .unwrap();
self.key = scheduler self.key = scheduler
.schedule_keyed_event_at( .schedule_keyed_event(scheduler.time() + Duration::from_secs(2), Self::action2, ())
scheduler.time() + Duration::from_secs(2),
Self::action2,
(),
)
.ok(); .ok();
} }
async fn action1(&mut self) { async fn action1(&mut self) {
@ -100,14 +96,10 @@ fn model_cancel_same_time_keyed_event() {
impl TestModel { impl TestModel {
fn trigger(&mut self, _: (), scheduler: &Scheduler<Self>) { fn trigger(&mut self, _: (), scheduler: &Scheduler<Self>) {
scheduler scheduler
.schedule_event_at(scheduler.time() + Duration::from_secs(2), Self::action1, ()) .schedule_event(scheduler.time() + Duration::from_secs(2), Self::action1, ())
.unwrap(); .unwrap();
self.key = scheduler self.key = scheduler
.schedule_keyed_event_at( .schedule_keyed_event(scheduler.time() + Duration::from_secs(2), Self::action2, ())
scheduler.time() + Duration::from_secs(2),
Self::action2,
(),
)
.ok(); .ok();
} }
async fn action1(&mut self) { async fn action1(&mut self) {
@ -148,7 +140,7 @@ fn model_schedule_periodic_event() {
impl TestModel { impl TestModel {
fn trigger(&mut self, _: (), scheduler: &Scheduler<Self>) { fn trigger(&mut self, _: (), scheduler: &Scheduler<Self>) {
scheduler scheduler
.schedule_periodic_event_at( .schedule_periodic_event(
scheduler.time() + Duration::from_secs(2), scheduler.time() + Duration::from_secs(2),
Duration::from_secs(3), Duration::from_secs(3),
Self::action, Self::action,
@ -195,7 +187,7 @@ fn model_cancel_periodic_event() {
impl TestModel { impl TestModel {
fn trigger(&mut self, _: (), scheduler: &Scheduler<Self>) { fn trigger(&mut self, _: (), scheduler: &Scheduler<Self>) {
self.key = scheduler self.key = scheduler
.schedule_periodic_keyed_event_at( .schedule_keyed_periodic_event(
scheduler.time() + Duration::from_secs(2), scheduler.time() + Duration::from_secs(2),
Duration::from_secs(3), Duration::from_secs(3),
Self::action, Self::action,

20
asynchronix/tests/simulation_scheduling.rs
View File

@ -49,9 +49,9 @@ fn simulation_schedule_events() {
let (mut simu, t0, addr, mut output) = simple_bench(); let (mut simu, t0, addr, mut output) = simple_bench();
// Queue 2 events at t0+3s and t0+2s, in reverse order. // Queue 2 events at t0+3s and t0+2s, in reverse order.
simu.schedule_event_in(Duration::from_secs(3), PassThroughModel::input, (), &addr) simu.schedule_event(Duration::from_secs(3), PassThroughModel::input, (), &addr)
.unwrap(); .unwrap();
simu.schedule_event_at( simu.schedule_event(
t0 + Duration::from_secs(2), t0 + Duration::from_secs(2),
PassThroughModel::input, PassThroughModel::input,
(), (),
@ -65,7 +65,7 @@ fn simulation_schedule_events() {
assert!(output.next().is_some()); assert!(output.next().is_some());
// Schedule another event in 4s (at t0+6s). // Schedule another event in 4s (at t0+6s).
simu.schedule_event_in(Duration::from_secs(4), PassThroughModel::input, (), &addr) simu.schedule_event(Duration::from_secs(4), PassThroughModel::input, (), &addr)
.unwrap(); .unwrap();
// Move to the 2nd event at t0+3s. // Move to the 2nd event at t0+3s.
@ -85,7 +85,7 @@ fn simulation_schedule_keyed_events() {
let (mut simu, t0, addr, mut output) = simple_bench(); let (mut simu, t0, addr, mut output) = simple_bench();
let event_t1 = simu let event_t1 = simu
.schedule_keyed_event_at( .schedule_keyed_event(
t0 + Duration::from_secs(1), t0 + Duration::from_secs(1),
PassThroughModel::input, PassThroughModel::input,
1, 1,
@ -94,10 +94,10 @@ fn simulation_schedule_keyed_events() {
.unwrap(); .unwrap();
let event_t2_1 = simu let event_t2_1 = simu
.schedule_keyed_event_in(Duration::from_secs(2), PassThroughModel::input, 21, &addr) .schedule_keyed_event(Duration::from_secs(2), PassThroughModel::input, 21, &addr)
.unwrap(); .unwrap();
simu.schedule_event_in(Duration::from_secs(2), PassThroughModel::input, 22, &addr) simu.schedule_event(Duration::from_secs(2), PassThroughModel::input, 22, &addr)
.unwrap(); .unwrap();
// Move to the 1st event at t0+1. // Move to the 1st event at t0+1.
@ -123,7 +123,7 @@ fn simulation_schedule_periodic_events() {
let (mut simu, t0, addr, mut output) = simple_bench(); let (mut simu, t0, addr, mut output) = simple_bench();
// Queue 2 periodic events at t0 + 3s + k*2s. // Queue 2 periodic events at t0 + 3s + k*2s.
simu.schedule_periodic_event_in( simu.schedule_periodic_event(
Duration::from_secs(3), Duration::from_secs(3),
Duration::from_secs(2), Duration::from_secs(2),
PassThroughModel::input, PassThroughModel::input,
@ -131,7 +131,7 @@ fn simulation_schedule_periodic_events() {
&addr, &addr,
) )
.unwrap(); .unwrap();
simu.schedule_periodic_event_at( simu.schedule_periodic_event(
t0 + Duration::from_secs(3), t0 + Duration::from_secs(3),
Duration::from_secs(2), Duration::from_secs(2),
PassThroughModel::input, PassThroughModel::input,
@ -158,7 +158,7 @@ fn simulation_schedule_periodic_keyed_events() {
let (mut simu, t0, addr, mut output) = simple_bench(); let (mut simu, t0, addr, mut output) = simple_bench();
// Queue 2 periodic events at t0 + 3s + k*2s. // Queue 2 periodic events at t0 + 3s + k*2s.
simu.schedule_periodic_event_in( simu.schedule_periodic_event(
Duration::from_secs(3), Duration::from_secs(3),
Duration::from_secs(2), Duration::from_secs(2),
PassThroughModel::input, PassThroughModel::input,
@ -167,7 +167,7 @@ fn simulation_schedule_periodic_keyed_events() {
) )
.unwrap(); .unwrap();
let event2_key = simu let event2_key = simu
.schedule_periodic_keyed_event_at( .schedule_keyed_periodic_event(
t0 + Duration::from_secs(3), t0 + Duration::from_secs(3),
Duration::from_secs(2), Duration::from_secs(2),
PassThroughModel::input, PassThroughModel::input,