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

First release candidate for v0.1.0

Browse Source
This commit is contained in:
Serge Barral
2023-01-16 23:05:46 +01:00
parent fe00ee0743
commit 31520d461a
58 changed files with 9731 additions and 1401 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
asynchronix/src/model/markers.rs Normal file
View File

@ -0,0 +1,31 @@
//! Marker types for simulation model methods.
/// Marker type for regular simulation model methods that take a mutable
/// reference to the model, without any other argument.
#[derive(Debug)]
pub struct WithoutArguments {}
/// Marker type for regular simulation model methods that take a mutable
/// reference to the model and a message, without scheduler argument.
#[derive(Debug)]
pub struct WithoutScheduler {}
/// Marker type for regular simulation model methods that take a mutable
/// reference to the model, a message and an explicit scheduler argument.
#[derive(Debug)]
pub struct WithScheduler {}
/// Marker type for asynchronous simulation model methods that take a mutable
/// reference to the model, without any other argument.
#[derive(Debug)]
pub struct AsyncWithoutArguments {}
/// Marker type for asynchronous simulation model methods that take a mutable
/// reference to the model and a message, without scheduler argument.
#[derive(Debug)]
pub struct AsyncWithoutScheduler {}
/// Marker type for asynchronous simulation model methods that take a mutable
/// reference to the model, a message and an explicit scheduler argument.
#[derive(Debug)]
pub struct AsyncWithScheduler {}

185
asynchronix/src/model/model_fn.rs Normal file
View File

@ -0,0 +1,185 @@
//! Trait for model input and replier ports.
use std::future::{ready, Future, Ready};
use crate::model::{markers, Model};
use crate::time::Scheduler;
/// A function, method or closures that can be used as an *input port*.
///
/// This trait is in particular implemented for any function or method with the
/// following signature, where it is implicitly assumed that the function
/// implements `Send + 'static`:
///
/// ```ignore
/// FnOnce(&mut M, T)
/// FnOnce(&mut M, T, &Scheduler<M>)
/// async fn(&mut M, T)
/// async fn(&mut M, T, &Scheduler<M>)
/// where
/// M: Model
/// ```
///
/// It is also implemented for the following signatures when `T=()`:
///
/// ```ignore
/// FnOnce(&mut M)
/// async fn(&mut M)
/// where
/// M: Model
/// ```
pub trait InputFn<'a, M: Model, T, S>: Send + 'static {
/// The `Future` returned by the asynchronous method.
type Future: Future<Output = ()> + Send + 'a;
/// Calls the method.
fn call(self, model: &'a mut M, arg: T, scheduler: &'a Scheduler<M>) -> Self::Future;
}
impl<'a, M, F> InputFn<'a, M, (), markers::WithoutArguments> for F
where
M: Model,
F: FnOnce(&'a mut M) + Send + 'static,
{
type Future = Ready<()>;
fn call(self, model: &'a mut M, _arg: (), _scheduler: &'a Scheduler<M>) -> Self::Future {
self(model);
ready(())
}
}
impl<'a, M, T, F> InputFn<'a, M, T, markers::WithoutScheduler> for F
where
M: Model,
F: FnOnce(&'a mut M, T) + Send + 'static,
{
type Future = Ready<()>;
fn call(self, model: &'a mut M, arg: T, _scheduler: &'a Scheduler<M>) -> Self::Future {
self(model, arg);
ready(())
}
}
impl<'a, M, T, F> InputFn<'a, M, T, markers::WithScheduler> for F
where
M: Model,
F: FnOnce(&'a mut M, T, &'a Scheduler<M>) + Send + 'static,
{
type Future = Ready<()>;
fn call(self, model: &'a mut M, arg: T, scheduler: &'a Scheduler<M>) -> Self::Future {
self(model, arg, scheduler);
ready(())
}
}
impl<'a, M, Fut, F> InputFn<'a, M, (), markers::AsyncWithoutArguments> for F
where
M: Model,
Fut: Future<Output = ()> + Send + 'a,
F: FnOnce(&'a mut M) -> Fut + Send + 'static,
{
type Future = Fut;
fn call(self, model: &'a mut M, _arg: (), _scheduler: &'a Scheduler<M>) -> Self::Future {
self(model)
}
}
impl<'a, M, T, Fut, F> InputFn<'a, M, T, markers::AsyncWithoutScheduler> for F
where
M: Model,
Fut: Future<Output = ()> + Send + 'a,
F: FnOnce(&'a mut M, T) -> Fut + Send + 'static,
{
type Future = Fut;
fn call(self, model: &'a mut M, arg: T, _scheduler: &'a Scheduler<M>) -> Self::Future {
self(model, arg)
}
}
impl<'a, M, T, Fut, F> InputFn<'a, M, T, markers::AsyncWithScheduler> for F
where
M: Model,
Fut: Future<Output = ()> + Send + 'a,
F: FnOnce(&'a mut M, T, &'a Scheduler<M>) -> Fut + Send + 'static,
{
type Future = Fut;
fn call(self, model: &'a mut M, arg: T, scheduler: &'a Scheduler<M>) -> Self::Future {
self(model, arg, scheduler)
}
}
/// A function, method or closure that can be used as a *replier port*.
///
/// This trait is in particular implemented for any function or method with the
/// following signature, where it is implicitly assumed that the function
/// implements `Send + 'static`:
///
/// ```ignore
/// async fn(&mut M, T) -> R
/// async fn(&mut M, T, &Scheduler<M>) -> R
/// where
/// M: Model
/// ```
///
/// It is also implemented for the following signatures when `T=()`:
///
/// ```ignore
/// async fn(&mut M) -> R
/// where
/// M: Model
/// ```
pub trait ReplierFn<'a, M: Model, T, R, S>: Send + 'static {
/// The `Future` returned by the asynchronous method.
type Future: Future<Output = R> + Send + 'a;
/// Calls the method.
fn call(self, model: &'a mut M, arg: T, scheduler: &'a Scheduler<M>) -> Self::Future;
}
impl<'a, M, R, Fut, F> ReplierFn<'a, M, (), R, markers::AsyncWithoutArguments> for F
where
M: Model,
Fut: Future<Output = R> + Send + 'a,
F: FnOnce(&'a mut M) -> Fut + Send + 'static,
{
type Future = Fut;
fn call(self, model: &'a mut M, _arg: (), _scheduler: &'a Scheduler<M>) -> Self::Future {
self(model)
}
}
impl<'a, M, T, R, Fut, F> ReplierFn<'a, M, T, R, markers::AsyncWithoutScheduler> for F
where
M: Model,
Fut: Future<Output = R> + Send + 'a,
F: FnOnce(&'a mut M, T) -> Fut + Send + 'static,
{
type Future = Fut;
fn call(self, model: &'a mut M, arg: T, _scheduler: &'a Scheduler<M>) -> Self::Future {
self(model, arg)
}
}
impl<'a, M, T, R, Fut, F> ReplierFn<'a, M, T, R, markers::AsyncWithScheduler> for F
where
M: Model,
Fut: Future<Output = R> + Send + 'a,
F: FnOnce(&'a mut M, T, &'a Scheduler<M>) -> Fut + Send + 'static,
{
type Future = Fut;
fn call(self, model: &'a mut M, arg: T, scheduler: &'a Scheduler<M>) -> Self::Future {
self(model, arg, scheduler)
}
}

218
asynchronix/src/model/ports.rs Normal file
View File

@ -0,0 +1,218 @@
//! Model ports for event and query broadcasting.
//!
//! Models typically contain [`Output`] and/or [`Requestor`] ports, exposed as
//! public member variables. Output ports broadcast events to all connected
//! input ports, while requestor ports broadcast queries to, and retrieve
//! replies from, all connected replier ports.
//!
//! On the surface, output and requestor ports only differ in that sending a
//! query from a requestor port also returns an iterator over the replies from
//! all connected ports. Sending a query is more costly, however, because of the
//! need to wait until all connected models have processed the query. In
//! contrast, since events are buffered in the mailbox of the target model,
//! sending an event is a fire-and-forget operation. For this reason, output
//! ports should generally be preferred over requestor ports when possible.
use std::fmt;
use std::sync::{Arc, Mutex};
mod broadcaster;
mod sender;
use crate::model::{InputFn, Model, ReplierFn};
use crate::simulation::{Address, EventSlot, EventStream};
use crate::util::spsc_queue;
use broadcaster::Broadcaster;
use self::sender::{EventSender, EventSlotSender, EventStreamSender, QuerySender};
#[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
/// Unique identifier for a connection between two ports.
pub struct LineId(u64);
/// An output port.
///
/// `Output` ports can be connected to input ports, i.e. to asynchronous model
/// methods that return no value. They broadcast events to all connected input
/// ports.
pub struct Output<T: Clone + Send + 'static> {
broadcaster: Broadcaster<T, ()>,
next_line_id: u64,
}
impl<T: Clone + Send + 'static> Output<T> {
/// Creates a new, disconnected `Output` port.
pub fn new() -> Self {
Self::default()
}
/// Adds a connection to an input port of the model specified by the
/// address.
///
/// The input port must be an asynchronous method of a model of type `M`
/// taking as argument a value of type `T` plus, optionally, a scheduler
/// reference.
pub fn connect<M, F, S>(&mut self, input: F, address: impl Into<Address<M>>) -> LineId
where
M: Model,
F: for<'a> InputFn<'a, M, T, S> + Copy,
S: Send + 'static,
{
assert!(self.next_line_id != u64::MAX);
let line_id = LineId(self.next_line_id);
self.next_line_id += 1;
let sender = Box::new(EventSender::new(input, address.into().0));
self.broadcaster.add(sender, line_id);
line_id
}
/// Adds a connection to an event stream iterator.
pub fn connect_stream(&mut self) -> (EventStream<T>, LineId) {
assert!(self.next_line_id != u64::MAX);
let line_id = LineId(self.next_line_id);
self.next_line_id += 1;
let (producer, consumer) = spsc_queue::spsc_queue();
let sender = Box::new(EventStreamSender::new(producer));
let event_stream = EventStream::new(consumer);
self.broadcaster.add(sender, line_id);
(event_stream, line_id)
}
/// Adds a connection to an event slot.
pub fn connect_slot(&mut self) -> (EventSlot<T>, LineId) {
assert!(self.next_line_id != u64::MAX);
let line_id = LineId(self.next_line_id);
self.next_line_id += 1;
let slot = Arc::new(Mutex::new(None));
let sender = Box::new(EventSlotSender::new(slot.clone()));
let event_slot = EventSlot::new(slot);
self.broadcaster.add(sender, line_id);
(event_slot, line_id)
}
/// Removes the connection specified by the `LineId` parameter.
///
/// It is a logic error to specify a line identifier from another [`Output`]
/// or [`Requestor`] instance and may result in the disconnection of an
/// arbitrary endpoint.
pub fn disconnect(&mut self, line_id: LineId) -> Result<(), LineError> {
if self.broadcaster.remove(line_id) {
Ok(())
} else {
Err(LineError {})
}
}
/// Removes all connections.
pub fn disconnect_all(&mut self) {
self.broadcaster.clear();
}
/// Broadcasts an event to all connected input ports.
pub async fn send(&mut self, arg: T) {
self.broadcaster.broadcast_event(arg).await.unwrap();
}
}
impl<T: Clone + Send + 'static> Default for Output<T> {
fn default() -> Self {
Self {
broadcaster: Broadcaster::default(),
next_line_id: 0,
}
}
}
impl<T: Clone + Send + 'static> fmt::Debug for Output<T> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "Output ({} connected ports)", self.broadcaster.len())
}
}
/// A requestor port.
///
/// `Requestor` ports can be connected to replier ports, i.e. to asynchronous
/// model methods that return a value. They broadcast queries to all connected
/// replier ports.
pub struct Requestor<T: Clone + Send + 'static, R: Send + 'static> {
broadcaster: Broadcaster<T, R>,
next_line_id: u64,
}
impl<T: Clone + Send + 'static, R: Send + 'static> Requestor<T, R> {
/// Creates a new, disconnected `Requestor` port.
pub fn new() -> Self {
Self::default()
}
/// Adds a connection to a replier port of the model specified by the
/// address.
///
/// The replier port must be an asynchronous method of a model of type `M`
/// returning a value of type `R` and taking as argument a value of type `T`
/// plus, optionally, a scheduler reference.
pub fn connect<M, F, S>(&mut self, replier: F, address: impl Into<Address<M>>) -> LineId
where
M: Model,
F: for<'a> ReplierFn<'a, M, T, R, S> + Copy,
S: Send + 'static,
{
assert!(self.next_line_id != u64::MAX);
let line_id = LineId(self.next_line_id);
self.next_line_id += 1;
let sender = Box::new(QuerySender::new(replier, address.into().0));
self.broadcaster.add(sender, line_id);
line_id
}
/// Removes the connection specified by the `LineId` parameter.
///
/// It is a logic error to specify a line identifier from another [`Output`]
/// or [`Requestor`] instance and may result in the disconnection of an
/// arbitrary endpoint.
pub fn disconnect(&mut self, line_id: LineId) -> Result<(), LineError> {
if self.broadcaster.remove(line_id) {
Ok(())
} else {
Err(LineError {})
}
}
/// Removes all connections.
pub fn disconnect_all(&mut self) {
self.broadcaster.clear();
}
/// Broadcasts a query to all connected replier ports.
pub async fn send(&mut self, arg: T) -> impl Iterator<Item = R> + '_ {
self.broadcaster.broadcast_query(arg).await.unwrap()
}
}
impl<T: Clone + Send + 'static, R: Send + 'static> Default for Requestor<T, R> {
fn default() -> Self {
Self {
broadcaster: Broadcaster::default(),
next_line_id: 0,
}
}
}
impl<T: Clone + Send + 'static, R: Send + 'static> fmt::Debug for Requestor<T, R> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "Requestor ({} connected ports)", self.broadcaster.len())
}
}
/// Error raised when the specified line cannot be found.
#[derive(Copy, Clone, Debug)]
pub struct LineError {}

746
asynchronix/src/model/ports/broadcaster.rs Normal file
View File

@ -0,0 +1,746 @@
use std::future::Future;
use std::mem::ManuallyDrop;
use std::pin::Pin;
use std::task::{Context, Poll};
use diatomic_waker::WakeSink;
use recycle_box::{coerce_box, RecycleBox};
use super::sender::{SendError, Sender};
use super::LineId;
use task_set::TaskSet;
mod task_set;
/// An object that can efficiently broadcast messages to several addresses.
///
/// This object maintains a list of senders associated to each target address.
/// When a message is broadcasted, the sender futures are awaited in parallel.
/// This is somewhat similar to what `FuturesOrdered` in the `futures` crate
/// does, but with some key differences:
///
/// - tasks and future storage are reusable to avoid repeated allocation, so
/// allocation occurs only after a new sender is added,
/// - the outputs of all sender futures are returned all at once rather than
/// with an asynchronous iterator (a.k.a. async stream); the implementation
/// exploits this behavior by waking the main broadcast future only when all
/// sender futures have been awaken, which strongly reduces overhead since
/// waking a sender task does not actually schedule it on the executor.
pub(super) struct Broadcaster<T: Clone + 'static, R: 'static> {
/// The list of senders with their associated line identifier.
senders: Vec<(LineId, Box<dyn Sender<T, R>>)>,
/// Fields explicitly borrowed by the `BroadcastFuture`.
shared: Shared<R>,
}
impl<T: Clone + 'static> Broadcaster<T, ()> {
/// Broadcasts an event to all addresses.
pub(super) async fn broadcast_event(&mut self, arg: T) -> Result<(), BroadcastError> {
match self.senders.as_mut_slice() {
// No sender.
[] => Ok(()),
// One sender.
[sender] => sender.1.send(arg).await.map_err(|_| BroadcastError {}),
// Multiple senders.
_ => self.broadcast(arg).await,
}
}
}
impl<T: Clone + 'static, R> Broadcaster<T, R> {
/// Adds a new sender associated to the specified identifier.
///
/// # Panics
///
/// This method will panic if the total count of senders would reach
/// `u32::MAX - 1`.
pub(super) fn add(&mut self, sender: Box<dyn Sender<T, R>>, id: LineId) {
self.senders.push((id, sender));
self.shared.futures_env.push(FutureEnv {
storage: None,
output: None,
});
self.shared.task_set.resize(self.senders.len());
}
/// Removes the first sender with the specified identifier, if any.
///
/// Returns `true` if there was indeed a sender associated to the specified
/// identifier.
pub(super) fn remove(&mut self, id: LineId) -> bool {
if let Some(pos) = self.senders.iter().position(|s| s.0 == id) {
self.senders.swap_remove(pos);
self.shared.futures_env.swap_remove(pos);
self.shared.task_set.resize(self.senders.len());
return true;
}
false
}
/// Removes all senders.
pub(super) fn clear(&mut self) {
self.senders.clear();
self.shared.futures_env.clear();
self.shared.task_set.resize(0);
}
/// Returns the number of connected senders.
pub(super) fn len(&self) -> usize {
self.senders.len()
}
/// Broadcasts a query to all addresses and collect all responses.
pub(super) async fn broadcast_query(
&mut self,
arg: T,
) -> Result<impl Iterator<Item = R> + '_, BroadcastError> {
match self.senders.as_mut_slice() {
// No sender.
[] => {}
// One sender.
[sender] => {
let output = sender.1.send(arg).await.map_err(|_| BroadcastError {})?;
self.shared.futures_env[0].output = Some(output);
}
// Multiple senders.
_ => self.broadcast(arg).await?,
};
// At this point all outputs should be available so `unwrap` can be
// called on the output of each future.
let outputs = self
.shared
.futures_env
.iter_mut()
.map(|t| t.output.take().unwrap());
Ok(outputs)
}
/// Efficiently broadcasts a message or a query to multiple addresses.
///
/// This method does not collect the responses from queries.
fn broadcast(&mut self, arg: T) -> BroadcastFuture<'_, R> {
let futures_count = self.senders.len();
let mut futures = recycle_vec(self.shared.storage.take().unwrap_or_default());
// Broadcast the message and collect all futures.
for (i, (sender, futures_env)) in self
.senders
.iter_mut()
.zip(self.shared.futures_env.iter_mut())
.enumerate()
{
let future_cache = futures_env
.storage
.take()
.unwrap_or_else(|| RecycleBox::new(()));
// Move the argument rather than clone it for the last future.
if i + 1 == futures_count {
let future: RecycleBox<dyn Future<Output = Result<R, SendError>> + Send + '_> =
coerce_box!(RecycleBox::recycle(future_cache, sender.1.send(arg)));
futures.push(RecycleBox::into_pin(future));
break;
}
let future: RecycleBox<dyn Future<Output = Result<R, SendError>> + Send + '_> = coerce_box!(
RecycleBox::recycle(future_cache, sender.1.send(arg.clone()))
);
futures.push(RecycleBox::into_pin(future));
}
// Generate the global future.
BroadcastFuture::new(&mut self.shared, futures)
}
}
impl<T: Clone + 'static, R> Default for Broadcaster<T, R> {
/// Creates an empty `Broadcaster` object.
fn default() -> Self {
let wake_sink = WakeSink::new();
let wake_src = wake_sink.source();
Self {
senders: Vec::new(),
shared: Shared {
wake_sink,
task_set: TaskSet::new(wake_src),
futures_env: Vec::new(),
storage: None,
},
}
}
}
/// Data related to a sender future.
struct FutureEnv<R> {
/// Cached storage for the future.
storage: Option<RecycleBox<()>>,
/// Output of the associated future.
output: Option<R>,
}
/// A type-erased `Send` future wrapped in a `RecycleBox`.
type RecycleBoxFuture<'a, R> = RecycleBox<dyn Future<Output = Result<R, SendError>> + Send + 'a>;
/// Fields of `Broadcaster` that are explicitly borrowed by a `BroadcastFuture`.
struct Shared<R> {
/// Thread-safe waker handle.
wake_sink: WakeSink,
/// Tasks associated to the sender futures.
task_set: TaskSet,
/// Data related to the sender futures.
futures_env: Vec<FutureEnv<R>>,
/// Cached storage for the sender futures.
///
/// When it exists, the cached storage is always an empty vector but it
/// typically has a non-zero capacity. Its purpose is to reuse the
/// previously allocated capacity when creating new sender futures.
storage: Option<Vec<Pin<RecycleBoxFuture<'static, R>>>>,
}
/// A future aggregating the outputs of a collection of sender futures.
///
/// The idea is to join all sender futures as efficiently as possible, meaning:
///
/// - the sender futures are polled simultaneously rather than waiting for their
/// completion in a sequential manner,
/// - this future is never woken if it can be proven that at least one of the
/// individual sender task will still be awaken,
/// - the storage allocated for the sender futures is always returned to the
/// `Broadcast` object so it can be reused by the next future,
/// - the happy path (all futures immediately ready) is very fast.
pub(super) struct BroadcastFuture<'a, R> {
/// Reference to the shared fields of the `Broadcast` object.
shared: &'a mut Shared<R>,
/// List of all send futures.
futures: ManuallyDrop<Vec<Pin<RecycleBoxFuture<'a, R>>>>,
/// The total count of futures that have not yet been polled to completion.
pending_futures_count: usize,
/// State of completion of the future.
state: FutureState,
}
impl<'a, R> BroadcastFuture<'a, R> {
/// Creates a new `BroadcastFuture`.
fn new(shared: &'a mut Shared<R>, futures: Vec<Pin<RecycleBoxFuture<'a, R>>>) -> Self {
let futures_count = futures.len();
assert!(shared.futures_env.len() == futures_count);
for futures_env in shared.futures_env.iter_mut() {
// Drop the previous output if necessary.
futures_env.output.take();
}
BroadcastFuture {
shared,
futures: ManuallyDrop::new(futures),
state: FutureState::Uninit,
pending_futures_count: futures_count,
}
}
}
impl<'a, R> Drop for BroadcastFuture<'a, R> {
fn drop(&mut self) {
// Safety: this is safe since `self.futures` is never accessed after it
// is moved out.
let mut futures = unsafe { ManuallyDrop::take(&mut self.futures) };
// Recycle the future-containing boxes.
for (future, futures_env) in futures.drain(..).zip(self.shared.futures_env.iter_mut()) {
futures_env.storage = Some(RecycleBox::vacate_pinned(future));
}
// Recycle the vector that contained the futures.
self.shared.storage = Some(recycle_vec(futures));
}
}
impl<'a, R> Future for BroadcastFuture<'a, R> {
type Output = Result<(), BroadcastError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = &mut *self;
assert_ne!(this.state, FutureState::Completed);
// Poll all sender futures once if this is the first time the broadcast
// future is polled.
if this.state == FutureState::Uninit {
// Prevent spurious wake-ups.
this.shared.task_set.discard_scheduled();
for task_idx in 0..this.futures.len() {
let future_env = &mut this.shared.futures_env[task_idx];
let future = &mut this.futures[task_idx];
let task_waker_ref = this.shared.task_set.waker_of(task_idx);
let task_cx_ref = &mut Context::from_waker(&task_waker_ref);
match future.as_mut().poll(task_cx_ref) {
Poll::Ready(Ok(output)) => {
future_env.output = Some(output);
this.pending_futures_count -= 1;
}
Poll::Ready(Err(_)) => {
this.state = FutureState::Completed;
return Poll::Ready(Err(BroadcastError {}));
}
Poll::Pending => {}
}
}
if this.pending_futures_count == 0 {
this.state = FutureState::Completed;
return Poll::Ready(Ok(()));
}
this.state = FutureState::Pending;
}
// Repeatedly poll the futures of all scheduled tasks until there are no
// more scheduled tasks.
loop {
// Only register the waker if it is probable that we won't find any
// scheduled task.
if !this.shared.task_set.has_scheduled() {
this.shared.wake_sink.register(cx.waker());
}
// Retrieve the indices of the scheduled tasks if any. If there are
// no scheduled tasks, `Poll::Pending` is returned and this future
// will be awaken again when enough tasks have been scheduled.
let scheduled_tasks = match this
.shared
.task_set
.steal_scheduled(this.pending_futures_count)
{
Some(st) => st,
None => return Poll::Pending,
};
for task_idx in scheduled_tasks {
let future_env = &mut this.shared.futures_env[task_idx];
// Do not poll completed futures.
if future_env.output.is_some() {
continue;
}
let future = &mut this.futures[task_idx];
let task_waker_ref = this.shared.task_set.waker_of(task_idx);
let task_cx_ref = &mut Context::from_waker(&task_waker_ref);
match future.as_mut().poll(task_cx_ref) {
Poll::Ready(Ok(output)) => {
future_env.output = Some(output);
this.pending_futures_count -= 1;
}
Poll::Ready(Err(_)) => {
this.state = FutureState::Completed;
return Poll::Ready(Err(BroadcastError {}));
}
Poll::Pending => {}
}
}
if this.pending_futures_count == 0 {
this.state = FutureState::Completed;
return Poll::Ready(Ok(()));
}
}
}
}
/// Error returned when a message could not be delivered.
#[derive(Debug)]
pub(super) struct BroadcastError {}
#[derive(Debug, PartialEq)]
enum FutureState {
Uninit,
Pending,
Completed,
}
/// Drops all items in a vector and returns an empty vector of another type,
/// preserving the allocation and capacity of the original vector provided that
/// the layouts of `T` and `U` are compatible.
///
/// # Panics
///
/// This will panic in debug mode if the layouts are incompatible.
fn recycle_vec<T, U>(mut v: Vec<T>) -> Vec<U> {
debug_assert_eq!(
std::alloc::Layout::new::<T>(),
std::alloc::Layout::new::<U>()
);
let cap = v.capacity();
// No unsafe here: this just relies on an optimization in the `collect`
// method.
v.clear();
let v_out: Vec<U> = v.into_iter().map(|_| unreachable!()).collect();
debug_assert_eq!(v_out.capacity(), cap);
v_out
}
#[cfg(all(test, not(asynchronix_loom)))]
mod tests {
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use futures_executor::block_on;
use super::super::sender::QuerySender;
use crate::channel::Receiver;
use crate::model::Model;
use crate::time::Scheduler;
use crate::time::{MonotonicTime, TearableAtomicTime};
use crate::util::priority_queue::PriorityQueue;
use crate::util::sync_cell::SyncCell;
use super::super::*;
use super::*;
struct Counter {
inner: Arc<AtomicUsize>,
}
impl Counter {
fn new(counter: Arc<AtomicUsize>) -> Self {
Self { inner: counter }
}
async fn inc(&mut self, by: usize) {
self.inner.fetch_add(by, Ordering::Relaxed);
}
async fn fetch_inc(&mut self, by: usize) -> usize {
let res = self.inner.fetch_add(by, Ordering::Relaxed);
res
}
}
impl Model for Counter {}
#[test]
fn broadcast_event_smoke() {
const N_RECV: usize = 4;
let mut mailboxes = Vec::new();
let mut broadcaster = Broadcaster::default();
for id in 0..N_RECV {
let mailbox = Receiver::new(10);
let address = mailbox.sender();
let sender = Box::new(EventSender::new(Counter::inc, address));
broadcaster.add(sender, LineId(id as u64));
mailboxes.push(mailbox);
}
let th_broadcast = thread::spawn(move || {
block_on(broadcaster.broadcast_event(1)).unwrap();
});
let counter = Arc::new(AtomicUsize::new(0));
let th_recv: Vec<_> = mailboxes
.into_iter()
.map(|mut mailbox| {
thread::spawn({
let mut counter = Counter::new(counter.clone());
move || {
let dummy_address = Receiver::new(1).sender();
let dummy_priority_queue = Arc::new(Mutex::new(PriorityQueue::new()));
let dummy_time =
SyncCell::new(TearableAtomicTime::new(MonotonicTime::EPOCH)).reader();
let dummy_scheduler =
Scheduler::new(dummy_address, dummy_priority_queue, dummy_time);
block_on(mailbox.recv(&mut counter, &dummy_scheduler)).unwrap();
}
})
})
.collect();
th_broadcast.join().unwrap();
for th in th_recv {
th.join().unwrap();
}
assert_eq!(counter.load(Ordering::Relaxed), N_RECV);
}
#[test]
fn broadcast_query_smoke() {
const N_RECV: usize = 4;
let mut mailboxes = Vec::new();
let mut broadcaster = Broadcaster::default();
for id in 0..N_RECV {
let mailbox = Receiver::new(10);
let address = mailbox.sender();
let sender = Box::new(QuerySender::new(Counter::fetch_inc, address));
broadcaster.add(sender, LineId(id as u64));
mailboxes.push(mailbox);
}
let th_broadcast = thread::spawn(move || {
let iter = block_on(broadcaster.broadcast_query(1)).unwrap();
let sum = iter.fold(0, |acc, val| acc + val);
assert_eq!(sum, N_RECV * (N_RECV - 1) / 2); // sum of {0, 1, 2, ..., (N_RECV - 1)}
});
let counter = Arc::new(AtomicUsize::new(0));
let th_recv: Vec<_> = mailboxes
.into_iter()
.map(|mut mailbox| {
thread::spawn({
let mut counter = Counter::new(counter.clone());
move || {
let dummy_address = Receiver::new(1).sender();
let dummy_priority_queue = Arc::new(Mutex::new(PriorityQueue::new()));
let dummy_time =
SyncCell::new(TearableAtomicTime::new(MonotonicTime::EPOCH)).reader();
let dummy_scheduler =
Scheduler::new(dummy_address, dummy_priority_queue, dummy_time);
block_on(mailbox.recv(&mut counter, &dummy_scheduler)).unwrap();
thread::sleep(std::time::Duration::from_millis(100));
}
})
})
.collect();
th_broadcast.join().unwrap();
for th in th_recv {
th.join().unwrap();
}
assert_eq!(counter.load(Ordering::Relaxed), N_RECV);
}
}
#[cfg(all(test, asynchronix_loom))]
mod tests {
use futures_channel::mpsc;
use futures_util::StreamExt;
use loom::model::Builder;
use loom::sync::atomic::{AtomicBool, Ordering};
use loom::thread;
use waker_fn::waker_fn;
use super::super::sender::RecycledFuture;
use super::*;
// An event that may be waken spuriously.
struct TestEvent<R> {
receiver: mpsc::UnboundedReceiver<Option<R>>,
fut_storage: Option<RecycleBox<()>>,
}
impl<R: Send> Sender<(), R> for TestEvent<R> {
fn send(&mut self, _arg: ()) -> RecycledFuture<'_, Result<R, SendError>> {
let fut_storage = &mut self.fut_storage;
let receiver = &mut self.receiver;
RecycledFuture::new(fut_storage, async {
let mut stream = Box::pin(receiver.filter_map(|item| async { item }));
Ok(stream.next().await.unwrap())
})
}
}
// An object that can wake a `TestEvent`.
#[derive(Clone)]
struct TestEventWaker<R> {
sender: mpsc::UnboundedSender<Option<R>>,
}
impl<R> TestEventWaker<R> {
fn wake_spurious(&self) {
let _ = self.sender.unbounded_send(None);
}
fn wake_final(&self, value: R) {
let _ = self.sender.unbounded_send(Some(value));
}
}
fn test_event<R>() -> (TestEvent<R>, TestEventWaker<R>) {
let (sender, receiver) = mpsc::unbounded();
(
TestEvent {
receiver,
fut_storage: None,
},
TestEventWaker { sender },
)
}
#[test]
fn loom_broadcast_basic() {
const DEFAULT_PREEMPTION_BOUND: usize = 3;
let mut builder = Builder::new();
if builder.preemption_bound.is_none() {
builder.preemption_bound = Some(DEFAULT_PREEMPTION_BOUND);
}
builder.check(move || {
let (test_event1, waker1) = test_event::<usize>();
let (test_event2, waker2) = test_event::<usize>();
let (test_event3, waker3) = test_event::<usize>();
let mut broadcaster = Broadcaster::default();
broadcaster.add(Box::new(test_event1), LineId(1));
broadcaster.add(Box::new(test_event2), LineId(2));
broadcaster.add(Box::new(test_event3), LineId(3));
let mut fut = Box::pin(broadcaster.broadcast_query(()));
let is_scheduled = loom::sync::Arc::new(AtomicBool::new(false));
let is_scheduled_waker = is_scheduled.clone();
let waker = waker_fn(move || {
// We use swap rather than a plain store to work around this
// bug: <https://github.com/tokio-rs/loom/issues/254>
is_scheduled_waker.swap(true, Ordering::Release);
});
let mut cx = Context::from_waker(&waker);
let th1 = thread::spawn(move || waker1.wake_final(3));
let th2 = thread::spawn(move || waker2.wake_final(7));
let th3 = thread::spawn(move || waker3.wake_final(42));
let mut schedule_count = 0;
loop {
match fut.as_mut().poll(&mut cx) {
Poll::Ready(Ok(mut res)) => {
assert_eq!(res.next(), Some(3));
assert_eq!(res.next(), Some(7));
assert_eq!(res.next(), Some(42));
assert_eq!(res.next(), None);
return;
}
Poll::Ready(Err(_)) => panic!("sender error"),
Poll::Pending => {}
}
// If the task has not been scheduled, exit the polling loop.
if !is_scheduled.swap(false, Ordering::Acquire) {
break;
}
schedule_count += 1;
assert!(schedule_count <= 1);
}
th1.join().unwrap();
th2.join().unwrap();
th3.join().unwrap();
assert!(is_scheduled.load(Ordering::Acquire));
match fut.as_mut().poll(&mut cx) {
Poll::Ready(Ok(mut res)) => {
assert_eq!(res.next(), Some(3));
assert_eq!(res.next(), Some(7));
assert_eq!(res.next(), Some(42));
assert_eq!(res.next(), None);
}
Poll::Ready(Err(_)) => panic!("sender error"),
Poll::Pending => panic!("the future has not completed"),
};
});
}
#[test]
fn loom_broadcast_spurious() {
const DEFAULT_PREEMPTION_BOUND: usize = 3;
let mut builder = Builder::new();
if builder.preemption_bound.is_none() {
builder.preemption_bound = Some(DEFAULT_PREEMPTION_BOUND);
}
builder.check(move || {
let (test_event1, waker1) = test_event::<usize>();
let (test_event2, waker2) = test_event::<usize>();
let mut broadcaster = Broadcaster::default();
broadcaster.add(Box::new(test_event1), LineId(1));
broadcaster.add(Box::new(test_event2), LineId(2));
let mut fut = Box::pin(broadcaster.broadcast_query(()));
let is_scheduled = loom::sync::Arc::new(AtomicBool::new(false));
let is_scheduled_waker = is_scheduled.clone();
let waker = waker_fn(move || {
// We use swap rather than a plain store to work around this
// bug: <https://github.com/tokio-rs/loom/issues/254>
is_scheduled_waker.swap(true, Ordering::Release);
});
let mut cx = Context::from_waker(&waker);
let spurious_waker = waker1.clone();
let th1 = thread::spawn(move || waker1.wake_final(3));
let th2 = thread::spawn(move || waker2.wake_final(7));
let th_spurious = thread::spawn(move || spurious_waker.wake_spurious());
let mut schedule_count = 0;
loop {
match fut.as_mut().poll(&mut cx) {
Poll::Ready(Ok(mut res)) => {
assert_eq!(res.next(), Some(3));
assert_eq!(res.next(), Some(7));
assert_eq!(res.next(), None);
return;
}
Poll::Ready(Err(_)) => panic!("sender error"),
Poll::Pending => {}
}
// If the task has not been scheduled, exit the polling loop.
if !is_scheduled.swap(false, Ordering::Acquire) {
break;
}
schedule_count += 1;
assert!(schedule_count <= 2);
}
th1.join().unwrap();
th2.join().unwrap();
th_spurious.join().unwrap();
assert!(is_scheduled.load(Ordering::Acquire));
match fut.as_mut().poll(&mut cx) {
Poll::Ready(Ok(mut res)) => {
assert_eq!(res.next(), Some(3));
assert_eq!(res.next(), Some(7));
assert_eq!(res.next(), None);
}
Poll::Ready(Err(_)) => panic!("sender error"),
Poll::Pending => panic!("the future has not completed"),
};
});
}
}

390
asynchronix/src/model/ports/broadcaster/task_set.rs Normal file
View File

@ -0,0 +1,390 @@
use std::sync::atomic::Ordering;
use std::sync::Arc;
use diatomic_waker::WakeSource;
use futures_task::{waker_ref, ArcWake, WakerRef};
use crate::loom_exports::sync::atomic::{AtomicU32, AtomicU64};
/// Special value for the `next` field of a task, indicating that the task to
/// which this field belongs is not currently in the list of scheduled tasks.
const SLEEPING: u32 = u32::MAX;
/// Special value for a task index, indicating the absence of task.
const EMPTY: u32 = u32::MAX - 1;
/// Mask for the index of the task pointed to by the head of the list of
/// scheduled tasks.
const INDEX_MASK: u64 = u32::MAX as u64;
/// Mask for the scheduling countdown in the head of the list of scheduled
/// tasks.
const COUNTDOWN_MASK: u64 = !INDEX_MASK;
/// A single increment of the scheduling countdown in the head of the list of
/// scheduled tasks.
const COUNTDOWN_ONE: u64 = 1 << 32;
/// A set of tasks that may be scheduled cheaply and can be requested to wake a
/// parent task only when a given amount of tasks have been scheduled.
///
/// This object maintains both a list of all active tasks and a list of the
/// subset of active tasks currently scheduled. The latter is stored in a
/// Treiber stack which links tasks through indices rather than pointers. Using
/// indices has two advantages: (i) it enables a fully safe implementation and
/// (ii) it makes it possible to use a single CAS to simultaneously move the
/// head and decrement the outstanding amount of tasks to be scheduled before
/// the parent task is notified.
pub(super) struct TaskSet {
/// Set of all active tasks, scheduled or not.
///
/// In some rare cases, the back of the vector can also contain inactive
/// (retired) tasks.
tasks: Vec<Arc<Task>>,
/// Head of the Treiber stack for scheduled tasks.
///
/// The lower bits specify the index of the last scheduled task, if any,
/// whereas the upper bits specify the countdown of tasks still to be
/// scheduled before the parent task is notified.
head: Arc<AtomicU64>,
/// A notifier used to wake the parent task.
notifier: WakeSource,
/// Count of all active tasks, scheduled or not.
task_count: usize,
}
impl TaskSet {
/// Creates an initially empty set of tasks associated to the parent task
/// which notifier is provided.
#[allow(clippy::assertions_on_constants)]
pub(super) fn new(notifier: WakeSource) -> Self {
// Only 32-bit targets and above are supported.
assert!(usize::BITS >= u32::BITS);
Self {
tasks: Vec::new(),
head: Arc::new(AtomicU64::new(EMPTY as u64)),
notifier,
task_count: 0,
}
}
/// Steals scheduled tasks if any and returns an iterator over their
/// indices, otherwise returns `None` and requests a notification to be sent
/// after `notify_count` tasks have been scheduled.
///
/// In all cases, the list of scheduled tasks is guaranteed to be empty
/// after this call.
///
/// If some tasks were stolen, no notification is requested.
///
/// If no tasks were stolen, the notification is guaranteed to be triggered
/// no later than after `notify_count` tasks have been scheduled, though it
/// may in some cases be triggered earlier. If the specified `notify_count`
/// is zero then no notification is requested.
pub(super) fn steal_scheduled(&self, notify_count: usize) -> Option<TaskIterator<'_>> {
let countdown = u32::try_from(notify_count).unwrap();
let mut head = self.head.load(Ordering::Relaxed);
loop {
let new_head = if head & INDEX_MASK == EMPTY as u64 {
(countdown as u64 * COUNTDOWN_ONE) | EMPTY as u64
} else {
EMPTY as u64
};
// Ordering: this Acquire operation synchronizes with all Release
// operations in `Task::wake_by_ref` and ensures that all memory
// operations performed during and before the tasks were scheduled
// become visible.
match self.head.compare_exchange_weak(
head,
new_head,
Ordering::Acquire,
Ordering::Relaxed,
) {
Ok(_) => break,
Err(h) => head = h,
}
}
let index = (head & INDEX_MASK) as u32;
if index == EMPTY {
None
} else {
Some(TaskIterator {
task_list: self,
next_index: index,
})
}
}
/// Discards all scheduled tasks and cancels any request for notification
/// that may be set.
///
/// This method is very cheap if there are no scheduled tasks and if no
/// notification is currently requested.
///
/// All discarded tasks are put in the sleeping (unscheduled) state.
pub(super) fn discard_scheduled(&self) {
if self.head.load(Ordering::Relaxed) != EMPTY as u64 {
// Dropping the iterator ensures that all tasks are put in the
// sleeping state.
let _ = self.steal_scheduled(0);
}
}
/// Modify the number of active tasks.
///
/// Note that this method may discard all scheduled tasks.
///
/// # Panic
///
/// This method will panic if `len` is greater than `u32::MAX - 1`.
pub(super) fn resize(&mut self, len: usize) {
assert!(len <= EMPTY as usize && len <= SLEEPING as usize);
self.task_count = len;
// Add new tasks if necessary.
if len >= self.tasks.len() {
while len > self.tasks.len() {
let idx = self.tasks.len() as u32;
self.tasks.push(Arc::new(Task {
idx,
notifier: self.notifier.clone(),
next: AtomicU32::new(SLEEPING),
head: self.head.clone(),
}));
}
return;
}
// Try to remove inactive tasks.
//
// The main issue when shrinking the set of active tasks is that stale
// wakers may still be around and may at any moment be scheduled and
// insert their index in the list of scheduled tasks. If it cannot be
// guaranteed that this will not happen, then a reference to that task
// must be kept or the iterator for scheduled tasks will panic when
// indexing a stale task.
//
// To prevent an inactive task from being spuriously scheduled, it is
// enough to pretend that the task is already scheduled by setting its
// `next` field to anything else than `SLEEPING`. However, this could
// race if the task has just set its `next` field but has not yet
// updated the head of the list of scheduled tasks, so this can only be
// done reliably if the task is currently sleeping.
// All scheduled tasks are first unscheduled in case some of them are
// now inactive.
self.discard_scheduled();
// The position of tasks in the set must stay consistent with their
// associated index so tasks are popped from the back.
while self.tasks.len() > len {
// There is at least one task since `len()` was non-zero.
let task = self.tasks.last().unwrap();
// Ordering: Relaxed ordering is sufficient since the task is
// effectively discarded.
if task
.next
.compare_exchange(SLEEPING, EMPTY, Ordering::Relaxed, Ordering::Relaxed)
.is_err()
{
// The task could not be removed for now so the set of tasks cannot
// be shrunk further.
break;
}
self.tasks.pop();
}
}
/// Returns `true` if one or more tasks are currently scheduled.
pub(super) fn has_scheduled(&self) -> bool {
// Ordering: the content of the head is only used as an advisory flag so
// Relaxed ordering is sufficient.
self.head.load(Ordering::Relaxed) & INDEX_MASK != EMPTY as u64
}
/// Returns a reference to the waker associated to the active task with the
/// specified index.
///
/// # Panics
///
/// This method will panic if there is no active task with the provided
/// index.
pub(super) fn waker_of(&self, idx: usize) -> WakerRef {
assert!(idx < self.task_count);
waker_ref(&self.tasks[idx])
}
}
/// An asynchronous task associated with the future of a sender.
pub(super) struct Task {
/// Index of this task.
idx: u32,
/// A notifier triggered once a certain number of tasks have been scheduled.
notifier: WakeSource,
/// Index of the next task in the list of scheduled tasks.
next: AtomicU32,
/// Head of the list of scheduled tasks.
head: Arc<AtomicU64>,
}
impl ArcWake for Task {
fn wake(self: Arc<Self>) {
Self::wake_by_ref(&self);
}
fn wake_by_ref(arc_self: &Arc<Self>) {
let mut next = arc_self.next.load(Ordering::Relaxed);
let mut head = loop {
if next == SLEEPING {
// The task appears not to be scheduled yet: prepare its
// insertion in the list of scheduled tasks by setting the next
// task index to the index of the task currently pointed by the
// head.
//
// Ordering: Relaxed ordering is sufficient since the upcoming
// CAS on the head already ensure that all memory operations
// that precede this call to `wake_by_ref` become visible when
// the tasks are stolen.
let head = arc_self.head.load(Ordering::Relaxed);
match arc_self.next.compare_exchange_weak(
SLEEPING,
(head & INDEX_MASK) as u32,
Ordering::Relaxed,
Ordering::Relaxed,
) {
Ok(_) => break head,
Err(n) => next = n,
}
} else {
// The task appears to be already scheduled: confirm this and
// establish proper memory synchronization by performing a no-op
// RMW.
//
// Ordering: the Release ordering synchronizes with the Acquire
// swap operation in `TaskIterator::next` and ensures that all
// memory operations that precede this call to `wake_by_ref`
// will be visible when the task index is yielded.
match arc_self.next.compare_exchange_weak(
next,
next,
Ordering::Release,
Ordering::Relaxed,
) {
Ok(_) => return,
Err(n) => next = n,
}
}
};
// The index to the next task has been set to the index in the head.
// Other concurrent calls to `wake` or `wake_by_ref` will now see the
// task as scheduled so this thread is responsible for moving the head.
loop {
// Attempt a CAS which decrements the countdown if it is not already
// cleared and which sets the head's index to this task's index.
let countdown = head & COUNTDOWN_MASK;
let new_countdown = countdown.wrapping_sub((countdown != 0) as u64 * COUNTDOWN_ONE);
let new_head = new_countdown | arc_self.idx as u64;
// Ordering: this Release operation synchronizes with the Acquire
// operation on the head in `TaskSet::steal_scheduled` and ensures
// that the value of the `next` field as well as all memory
// operations that precede this call to `wake_by_ref` become visible
// when the tasks are stolen.
match arc_self.head.compare_exchange_weak(
head,
new_head,
Ordering::Release,
Ordering::Relaxed,
) {
Ok(_) => {
// If the countdown has just been cleared, it is necessary
// to send a notification.
if countdown == COUNTDOWN_ONE {
arc_self.notifier.notify();
}
return;
}
Err(h) => {
head = h;
// Update the index of the next task to the new value of the
// head.
//
// Why use a swap instead of a simple store? This is to
// maintain a release sequence which includes previous
// atomic operation on this field, and more specifically any
// no-op CAS that could have been performed by a concurrent
// call to wake. This ensures in turn that all memory
// operations that precede a no-op CAS will be visible when
// `next` is Acquired in `TaskIterator::next`.
//
// Ordering: Relaxed ordering is sufficient since
// synchronization is ensured by the upcoming CAS on the
// head.
arc_self
.next
.swap((head & INDEX_MASK) as u32, Ordering::Relaxed);
}
}
}
}
}
/// An iterator over scheduled tasks.
pub(super) struct TaskIterator<'a> {
task_list: &'a TaskSet,
next_index: u32,
}
impl<'a> Iterator for TaskIterator<'a> {
type Item = usize;
fn next(&mut self) -> Option<Self::Item> {
while self.next_index != EMPTY {
let index = self.next_index as usize;
// Ordering: the Acquire ordering synchronizes with any no-op CAS
// that could have been performed in `Task::wake_by_ref`, ensuring
// that all memory operations that precede such call to
// `Task::wake_by_ref` become visible.
self.next_index = self.task_list.tasks[index]
.next
.swap(SLEEPING, Ordering::Acquire);
// Only yield the index if the task is indeed active.
if index < self.task_list.task_count {
return Some(index);
}
}
None
}
}
impl<'a> Drop for TaskIterator<'a> {
fn drop(&mut self) {
// Put all remaining scheduled tasks in the sleeping state.
//
// Ordering: the task is ignored so it is not necessary to ensure that
// memory operations performed before the task was scheduled are
// visible. For the same reason, it is not necessary to synchronize with
// no-op CAS operations in `Task::wake_by_ref`, which is why separate
// load and store operations are used rather than a more expensive swap
// operation.
while self.next_index != EMPTY {
let index = self.next_index as usize;
self.next_index = self.task_list.tasks[index].next.load(Ordering::Relaxed);
self.task_list.tasks[index]
.next
.store(SLEEPING, Ordering::Relaxed);
}
}
}

245
asynchronix/src/model/ports/sender.rs Normal file
View File

@ -0,0 +1,245 @@
use std::error::Error;
use std::fmt;
use std::future::Future;
use std::marker::PhantomData;
use std::mem::ManuallyDrop;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use recycle_box::{coerce_box, RecycleBox};
use crate::channel;
use crate::model::{InputFn, Model, ReplierFn};
use crate::util::spsc_queue;
/// Abstraction over `EventSender` and `QuerySender`.
pub(super) trait Sender<T, R>: Send {
fn send(&mut self, arg: T) -> RecycledFuture<'_, Result<R, SendError>>;
}
/// An object that can send a payload to a model.
pub(super) struct EventSender<M: 'static, F, T, S> {
func: F,
sender: channel::Sender<M>,
fut_storage: Option<RecycleBox<()>>,
_phantom_closure: PhantomData<fn(&mut M, T)>,
_phantom_closure_marker: PhantomData<S>,
}
impl<M: Send, F, T, S> EventSender<M, F, T, S>
where
M: Model,
F: for<'a> InputFn<'a, M, T, S>,
T: Send + 'static,
{
pub(super) fn new(func: F, sender: channel::Sender<M>) -> Self {
Self {
func,
sender,
fut_storage: None,
_phantom_closure: PhantomData,
_phantom_closure_marker: PhantomData,
}
}
}
impl<M: Send, F, T, S> Sender<T, ()> for EventSender<M, F, T, S>
where
M: Model,
F: for<'a> InputFn<'a, M, T, S> + Copy,
T: Send + 'static,
S: Send,
{
fn send(&mut self, arg: T) -> RecycledFuture<'_, Result<(), SendError>> {
let func = self.func;
let fut = self.sender.send(move |model, scheduler, recycle_box| {
let fut = func.call(model, arg, scheduler);
coerce_box!(RecycleBox::recycle(recycle_box, fut))
});
RecycledFuture::new(&mut self.fut_storage, async move {
fut.await.map_err(|_| SendError {})
})
}
}
/// An object that can send a payload to a model and retrieve a response.
pub(super) struct QuerySender<M: 'static, F, T, R, S> {
func: F,
sender: channel::Sender<M>,
receiver: multishot::Receiver<R>,
fut_storage: Option<RecycleBox<()>>,
_phantom_closure: PhantomData<fn(&mut M, T) -> R>,
_phantom_closure_marker: PhantomData<S>,
}
impl<M, F, T, R, S> QuerySender<M, F, T, R, S>
where
M: Model,
F: for<'a> ReplierFn<'a, M, T, R, S>,
T: Send + 'static,
R: Send + 'static,
{
pub(super) fn new(func: F, sender: channel::Sender<M>) -> Self {
Self {
func,
sender,
receiver: multishot::Receiver::new(),
fut_storage: None,
_phantom_closure: PhantomData,
_phantom_closure_marker: PhantomData,
}
}
}
impl<M, F, T, R, S> Sender<T, R> for QuerySender<M, F, T, R, S>
where
M: Model,
F: for<'a> ReplierFn<'a, M, T, R, S> + Copy,
T: Send + 'static,
R: Send + 'static,
S: Send,
{
fn send(&mut self, arg: T) -> RecycledFuture<'_, Result<R, SendError>> {
let func = self.func;
let sender = &mut self.sender;
let reply_receiver = &mut self.receiver;
let fut_storage = &mut self.fut_storage;
// The previous future generated by this method should have been polled
// to completion so a new sender should be readily available.
let reply_sender = reply_receiver.sender().unwrap();
let send_fut = sender.send(move |model, scheduler, recycle_box| {
let fut = async move {
let reply = func.call(model, arg, scheduler).await;
reply_sender.send(reply);
};
coerce_box!(RecycleBox::recycle(recycle_box, fut))
});
RecycledFuture::new(fut_storage, async move {
// Send the message.
send_fut.await.map_err(|_| SendError {})?;
// Wait until the message is processed and the reply is sent back.
// If an error is received, it most likely means the mailbox was
// dropped before the message was processed.
reply_receiver.recv().await.map_err(|_| SendError {})
})
}
}
/// An object that can send a payload to an unbounded queue.
pub(super) struct EventStreamSender<T> {
producer: spsc_queue::Producer<T>,
fut_storage: Option<RecycleBox<()>>,
}
impl<T> EventStreamSender<T> {
pub(super) fn new(producer: spsc_queue::Producer<T>) -> Self {
Self {
producer,
fut_storage: None,
}
}
}
impl<T> Sender<T, ()> for EventStreamSender<T>
where
T: Send + 'static,
{
fn send(&mut self, arg: T) -> RecycledFuture<'_, Result<(), SendError>> {
let producer = &mut self.producer;
RecycledFuture::new(&mut self.fut_storage, async move {
producer.push(arg).map_err(|_| SendError {})
})
}
}
/// An object that can send a payload to a mutex-protected slot.
pub(super) struct EventSlotSender<T> {
slot: Arc<Mutex<Option<T>>>,
fut_storage: Option<RecycleBox<()>>,
}
impl<T> EventSlotSender<T> {
pub(super) fn new(slot: Arc<Mutex<Option<T>>>) -> Self {
Self {
slot,
fut_storage: None,
}
}
}
impl<T> Sender<T, ()> for EventSlotSender<T>
where
T: Send + 'static,
{
fn send(&mut self, arg: T) -> RecycledFuture<'_, Result<(), SendError>> {
let slot = &*self.slot;
RecycledFuture::new(&mut self.fut_storage, async move {
let mut slot = slot.lock().unwrap();
*slot = Some(arg);
Ok(())
})
}
}
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
/// Error returned when the mailbox was closed or dropped.
pub(super) struct SendError {}
impl fmt::Display for SendError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "sending message into a closed mailbox")
}
}
impl Error for SendError {}
pub(super) struct RecycledFuture<'a, T> {
fut: ManuallyDrop<Pin<RecycleBox<dyn Future<Output = T> + Send + 'a>>>,
lender_box: &'a mut Option<RecycleBox<()>>,
}
impl<'a, T> RecycledFuture<'a, T> {
pub(super) fn new<F: Future<Output = T> + Send + 'a>(
lender_box: &'a mut Option<RecycleBox<()>>,
fut: F,
) -> Self {
let vacated_box = lender_box.take().unwrap_or_else(|| RecycleBox::new(()));
let fut: RecycleBox<dyn Future<Output = T> + Send + 'a> =
coerce_box!(RecycleBox::recycle(vacated_box, fut));
Self {
fut: ManuallyDrop::new(RecycleBox::into_pin(fut)),
lender_box,
}
}
}
impl<'a, T> Drop for RecycledFuture<'a, T> {
fn drop(&mut self) {
// Return the box to the lender.
//
// Safety: taking the `fut` member is safe since it is never used again.
*self.lender_box = Some(RecycleBox::vacate_pinned(unsafe {
ManuallyDrop::take(&mut self.fut)
}));
}
}
impl<'a, T> Future for RecycledFuture<'a, T> {
type Output = T;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.fut.as_mut().poll(cx)
}
}