use crate::notify::{GenericNotify, Internal, Notification};
use crate::sync::atomic::Ordering;
use crate::sync::cell::{Cell, UnsafeCell};
use crate::sync::{Mutex, MutexGuard};
use crate::{RegisterResult, State, TaskRef};
use core::marker::PhantomPinned;
use core::mem;
use core::ops::{Deref, DerefMut};
use core::pin::Pin;
use core::ptr::NonNull;
pub(super) struct List<T>(Mutex<Inner<T>>);
struct Inner<T> {
head: Option<NonNull<Link<T>>>,
tail: Option<NonNull<Link<T>>>,
next: Option<NonNull<Link<T>>>,
len: usize,
notified: usize,
}
impl<T> List<T> {
pub(super) fn new() -> Self {
Self(Mutex::new(Inner {
head: None,
tail: None,
next: None,
len: 0,
notified: 0,
}))
}
pub fn total_listeners(&self) -> Result<usize, &str> {
match self.0.try_lock() {
Ok(mutex) => {
let len = mutex.len;
Ok(len)
}
Err(_) => Err("<locked>"),
}
}
}
impl<T> crate::Inner<T> {
fn lock(&self) -> ListLock<'_, '_, T> {
ListLock {
inner: self,
lock: self.list.0.lock().unwrap_or_else(|e| e.into_inner()),
}
}
pub(crate) fn needs_notification(&self, limit: usize) -> bool {
self.notified.load(Ordering::Acquire) < limit
}
pub(crate) fn insert(&self, mut listener: Pin<&mut Option<Listener<T>>>) {
let mut inner = self.lock();
let entry = unsafe {
listener.as_mut().set(Some(Listener {
link: UnsafeCell::new(Link {
state: Cell::new(State::Created),
prev: Cell::new(inner.tail),
next: Cell::new(None),
}),
_pin: PhantomPinned,
}));
let listener = listener.as_pin_mut().unwrap();
&*listener.link.get()
};
match mem::replace(&mut inner.tail, Some(entry.into())) {
None => inner.head = Some(entry.into()),
Some(t) => unsafe { t.as_ref().next.set(Some(entry.into())) },
};
if inner.next.is_none() {
inner.next = inner.tail;
}
inner.len += 1;
}
pub(crate) fn remove(
&self,
listener: Pin<&mut Option<Listener<T>>>,
propagate: bool,
) -> Option<State<T>> {
self.lock().remove(listener, propagate)
}
#[cold]
pub(crate) fn notify(&self, notify: impl Notification<Tag = T>) -> usize {
self.lock().notify(notify)
}
pub(crate) fn register(
&self,
mut listener: Pin<&mut Option<Listener<T>>>,
task: TaskRef<'_>,
) -> RegisterResult<T> {
let mut inner = self.lock();
let entry = unsafe {
let listener = match listener.as_mut().as_pin_mut() {
Some(listener) => listener,
None => return RegisterResult::NeverInserted,
};
&*listener.link.get()
};
match entry.state.replace(State::NotifiedTaken) {
State::Notified { tag, .. } => {
inner.remove(listener, false);
RegisterResult::Notified(tag)
}
State::Task(other_task) => {
entry.state.set(State::Task({
if !task.will_wake(other_task.as_task_ref()) {
task.into_task()
} else {
other_task
}
}));
RegisterResult::Registered
}
_ => {
entry.state.set(State::Task(task.into_task()));
RegisterResult::Registered
}
}
}
}
impl<T> Inner<T> {
fn remove(
&mut self,
mut listener: Pin<&mut Option<Listener<T>>>,
propagate: bool,
) -> Option<State<T>> {
let entry = unsafe {
let listener = listener.as_mut().as_pin_mut()?;
&*listener.link.get()
};
let prev = entry.prev.get();
let next = entry.next.get();
match prev {
None => self.head = next,
Some(p) => unsafe {
p.as_ref().next.set(next);
},
}
match next {
None => self.tail = prev,
Some(n) => unsafe {
n.as_ref().prev.set(prev);
},
}
if self.next == Some(entry.into()) {
self.next = next;
}
let entry = unsafe {
listener
.get_unchecked_mut()
.take()
.unwrap()
.link
.into_inner()
};
let mut state = entry.state.into_inner();
if state.is_notified() {
self.notified -= 1;
if propagate {
let state = mem::replace(&mut state, State::NotifiedTaken);
if let State::Notified { additional, tag } = state {
let tags = {
let mut tag = Some(tag);
move || tag.take().expect("tag already taken")
};
self.notify(GenericNotify::new(1, additional, tags));
}
}
}
self.len -= 1;
Some(state)
}
#[cold]
fn notify(&mut self, mut notify: impl Notification<Tag = T>) -> usize {
let mut n = notify.count(Internal::new());
let is_additional = notify.is_additional(Internal::new());
if !is_additional {
if n < self.notified {
return 0;
}
n -= self.notified;
}
let original_count = n;
while n > 0 {
n -= 1;
match self.next {
None => return original_count - n - 1,
Some(e) => {
let entry = unsafe { e.as_ref() };
self.next = entry.next.get();
let tag = notify.next_tag(Internal::new());
if let State::Task(task) = entry.state.replace(State::Notified {
additional: is_additional,
tag,
}) {
task.wake();
}
self.notified += 1;
}
}
}
original_count - n
}
}
struct ListLock<'a, 'b, T> {
lock: MutexGuard<'a, Inner<T>>,
inner: &'b crate::Inner<T>,
}
impl<T> Deref for ListLock<'_, '_, T> {
type Target = Inner<T>;
fn deref(&self) -> &Self::Target {
&self.lock
}
}
impl<T> DerefMut for ListLock<'_, '_, T> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.lock
}
}
impl<T> Drop for ListLock<'_, '_, T> {
fn drop(&mut self) {
let list = &mut **self;
let notified = if list.notified < list.len {
list.notified
} else {
core::usize::MAX
};
self.inner.notified.store(notified, Ordering::Release);
}
}
pub(crate) struct Listener<T> {
link: UnsafeCell<Link<T>>,
_pin: PhantomPinned,
}
struct Link<T> {
state: Cell<State<T>>,
prev: Cell<Option<NonNull<Link<T>>>>,
next: Cell<Option<NonNull<Link<T>>>>,
}
#[cfg(test)]
mod tests {
use super::*;
use futures_lite::pin;
#[cfg(target_family = "wasm")]
use wasm_bindgen_test::wasm_bindgen_test as test;
macro_rules! make_listeners {
($($id:ident),*) => {
$(
let $id = Option::<Listener<()>>::None;
pin!($id);
)*
};
}
#[test]
fn insert() {
let inner = crate::Inner::new();
make_listeners!(listen1, listen2, listen3);
inner.insert(listen1.as_mut());
inner.insert(listen2.as_mut());
inner.insert(listen3.as_mut());
assert_eq!(inner.lock().len, 3);
assert_eq!(inner.remove(listen2, false), Some(State::Created));
assert_eq!(inner.lock().len, 2);
assert_eq!(inner.remove(listen1, false), Some(State::Created));
assert_eq!(inner.lock().len, 1);
}
#[test]
fn drop_non_notified() {
let inner = crate::Inner::new();
make_listeners!(listen1, listen2, listen3);
inner.insert(listen1.as_mut());
inner.insert(listen2.as_mut());
inner.insert(listen3.as_mut());
inner.notify(GenericNotify::new(1, false, || ()));
inner.remove(listen3, true);
inner.remove(listen1, true);
inner.remove(listen2, true);
}
}