Skip to content

Commit

Permalink
bugfix: Tracking down a deadlock in async-channel
Browse files Browse the repository at this point in the history
There was a deadlock where notifications would not get delivered for
listeners that were stuck in the fallback queue. There is an
optimization where the notification count is checked to see if the
notifications delivered by the user would make any impact on the
listener chain; if it wouldn't, then it isn't done at all. For no_std,
I've removed this optimization, as I doubt that it can be done
consistently due to the nature of the fallback queue.

During testing I also uncovered problems with the current implementation
of the fallback queue itself. I've replaced it with concurrent-queue for
now, as I doubt that a fully functioning queue wouldn't be trivial
enough to justify inlining it into event-listener itself. Unfortunately
it adds three additional dependencies to the tree of event-listener.

With this change, the async-channel tests pass consistently.

Signed-off-by: John Nunley <dev@notgull.net>
  • Loading branch information
notgull committed Aug 19, 2023
1 parent 564b84b commit e408ccd
Show file tree
Hide file tree
Showing 7 changed files with 239 additions and 296 deletions.
3 changes: 2 additions & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,11 @@ exclude = ["/.*"]

[features]
default = ["std"]
std = ["parking"]
std = ["concurrent-queue/std", "parking"]
portable-atomic = ["portable-atomic-util", "portable_atomic_crate"]

[dependencies]
concurrent-queue = { version = "2.2.0", default-features = false }
parking = { version = "2.0.0", optional = true }
pin-project-lite = "0.2.12"
portable-atomic-util = { version = "0.1.2", default-features = false, optional = true, features = ["alloc"] }
Expand Down
18 changes: 16 additions & 2 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -388,7 +388,7 @@ impl<T> Event<T> {

// Notify if there is at least one unnotified listener and the number of notified
// listeners is less than `limit`.
if inner.notified.load(Ordering::Acquire) < limit {
if inner.needs_notification(limit) {
return inner.notify(notify);
}
}
Expand Down Expand Up @@ -993,7 +993,7 @@ impl<T, B: Borrow<Inner<T>> + Unpin> Listener<T, B> {
}

/// The state of a listener.
#[derive(Debug, PartialEq)]
#[derive(PartialEq)]
enum State<T> {
/// The listener was just created.
Created,
Expand All @@ -1016,6 +1016,20 @@ enum State<T> {
NotifiedTaken,
}

impl<T> fmt::Debug for State<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Created => f.write_str("Created"),
Self::Notified { additional, .. } => f
.debug_struct("Notified")
.field("additional", additional)
.finish(),
Self::Task(_) => f.write_str("Task(_)"),
Self::NotifiedTaken => f.write_str("NotifiedTaken"),
}
}
}

impl<T> State<T> {
fn is_notified(&self) -> bool {
matches!(self, Self::Notified { .. } | Self::NotifiedTaken)
Expand Down
Loading

0 comments on commit e408ccd

Please sign in to comment.