Skip to content

Commit

Permalink
refactor: Move drop glue of AddListener into separate struct
Browse files Browse the repository at this point in the history
  • Loading branch information
@fogti
fogti committed Nov 9, 2022
1 parent 7707b73 commit 9c31ae6
Show file tree
Hide file tree
Showing 2 changed files with 37 additions and 68 deletions.
16 changes: 8 additions & 8 deletions src/lib.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -282,7 +282,7 @@ impl Event {
if let Some(mut lock) = inner.lock() {
lock.notify_unnotified(n);
} else {
inner.push(Node::notify(Notify {
inner.push(Node::Notify(Notify {
count: n,
kind: NotifyKind::Notify,
}));
Expand Down Expand Up @@ -333,7 +333,7 @@ impl Event {
if let Some(mut lock) = inner.lock() {
lock.notify_unnotified(n);
} else {
inner.push(Node::notify(Notify {
inner.push(Node::Notify(Notify {
count: n,
kind: NotifyKind::Notify,
}));
Expand Down Expand Up @@ -383,7 +383,7 @@ impl Event {
if let Some(mut lock) = inner.lock() {
lock.notify_additional(n);
} else {
inner.push(Node::notify(Notify {
inner.push(Node::Notify(Notify {
count: n,
kind: NotifyKind::NotifyAdditional,
}));
Expand Down Expand Up @@ -435,7 +435,7 @@ impl Event {
if let Some(mut lock) = inner.lock() {
lock.notify_additional(n);
} else {
inner.push(Node::notify(Notify {
inner.push(Node::Notify(Notify {
count: n,
kind: NotifyKind::NotifyAdditional,
}));
Expand Down Expand Up @@ -618,7 +618,7 @@ impl EventListener {
None => {
// Wake us up when the lock is free.
let unparker = parker.unparker();
self.inner.push(Node::waiting(Task::Thread(unparker)));
self.inner.push(Node::Waiting(Task::Thread(unparker)));
parker.park()
}
}
Expand Down Expand Up @@ -715,7 +715,7 @@ impl EventListener {
}
} else {
// Let someone else do it for us.
self.inner.push(Node::remove_listener(entry, false));
self.inner.push(Node::RemoveListener { listener: entry, propagate: false });
}
}

Expand Down Expand Up @@ -773,7 +773,7 @@ impl Future for EventListener {
None => {
// Wait for the lock to be available.
self.inner
.push(Node::waiting(Task::Waker(cx.waker().clone())));
.push(Node::Waiting(Task::Waker(cx.waker().clone())));

// If the lock is suddenly available, we need to poll again.
if let Some(list) = self.inner.lock() {
Expand Down Expand Up @@ -835,7 +835,7 @@ impl Drop for EventListener {
}
None => {
// Request that someone else do it.
self.inner.push(Node::remove_listener(entry, true));
self.inner.push(Node::RemoveListener { listener: entry, propagate: true });
}
}
}
Expand Down
89 changes: 29 additions & 60 deletions src/node.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,22 +8,11 @@ use alloc::boxed::Box;
use core::ptr::NonNull;

/// A node in the backup queue.
pub(crate) struct Node {
/// The data associated with the node.
data: Option<NodeData>,
}

impl From<NodeData> for Node {
fn from(data: NodeData) -> Self {
Self { data: Some(data) }
}
}

enum NodeData {
pub(crate) enum Node {
/// This node is requesting to add a listener.
AddListener {
/// The pointer to the listener to add.
listener: NonNull<Entry>,
listener: Option<DistOwnedListener>,
},

/// This node is notifying a listener.
Expand All @@ -42,65 +31,55 @@ enum NodeData {
Waiting(Task),
}

impl Node {
/// Create a new listener submission entry.
pub(crate) fn listener() -> (Self, NonNull<Entry>) {
// Allocate an entry on the heap.
let entry = unsafe { NonNull::new_unchecked(Box::into_raw(Box::new(Entry::new()))) };

(NodeData::AddListener { listener: entry }.into(), entry)
}
pub(crate) struct DistOwnedListener(NonNull<Entry>);

/// Create a new notification entry.
pub(crate) fn notify(notify: Notify) -> Self {
NodeData::Notify(notify).into()
impl DistOwnedListener {
/// extracts the contained entry pointer from the DOL,
/// without calling the DOL Drop handler (such that the returned pointer stays valid)
fn take(self) -> NonNull<Entry> {
(&*core::mem::ManuallyDrop::new(self)).0
}
}

/// Create a new listener removal entry.
pub(crate) fn remove_listener(listener: NonNull<Entry>, propagate: bool) -> Self {
NodeData::RemoveListener {
listener,
propagate,
}
.into()
impl Drop for DistOwnedListener {
fn drop(&mut self) {
drop(unsafe { Box::from_raw(self.0.as_ptr()) });
}
}

/// Create a new waiting entry.
pub(crate) fn waiting(task: Task) -> Self {
NodeData::Waiting(task).into()
impl Node {
pub(crate) fn listener() -> (Self, NonNull<Entry>) {
let entry = Box::into_raw(Box::new(Entry::new()));
let entry = unsafe { NonNull::new_unchecked(entry) };
(Self::AddListener { listener: Some(DistOwnedListener(entry)) }, entry)
}

/// Indicate that this node has been enqueued.
pub(crate) fn enqueue(&self) {
if let Some(NodeData::AddListener { listener }) = &self.data {
unsafe {
listener.as_ref().enqueue();
}
if let Node::AddListener { listener: Some(entry) } = self {
unsafe { entry.0.as_ref() }.enqueue();
}
}

/// Apply the node to the list.
pub(crate) fn apply(mut self, list: &mut List, inner: &Inner) -> Option<Task> {
let data = self.data.take().unwrap();

match data {
NodeData::AddListener { listener } => {
pub(crate) fn apply(self, list: &mut List, inner: &Inner) -> Option<Task> {
match self {
Node::AddListener { mut listener } => {
// Add the listener to the list.
list.insert(listener);
let entry = listener.take().unwrap().take();
list.insert(entry);

// Dequeue the listener.
return unsafe { listener.as_ref().dequeue() };
return unsafe { entry.as_ref().dequeue() };
}
NodeData::Notify(notify) => {
Node::Notify(Notify { count, kind }) => {
// Notify the listener.
let Notify { count, kind } = notify;

match kind {
NotifyKind::Notify => list.notify_unnotified(count),
NotifyKind::NotifyAdditional => list.notify_additional(count),
}
}
NodeData::RemoveListener {
Node::RemoveListener {
listener,
propagate,
} => {
Expand All @@ -112,21 +91,11 @@ impl Node {
list.notify(1, additional);
}
}
NodeData::Waiting(task) => {
Node::Waiting(task) => {
return Some(task);
}
}

None
}
}

impl Drop for Node {
fn drop(&mut self) {
if let Some(NodeData::AddListener { listener }) = self.data.take() {
unsafe {
drop(Box::from_raw(listener.as_ptr()));
}
}
}
}

0 comments on commit 9c31ae6

Please sign in to comment.