Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

sync: change chan closed(&mut self) to closed(&self) #2939

Merged
merged 2 commits into from
Oct 12, 2020
Merged

sync: change chan closed(&mut self) to closed(&self) #2939

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f6034fd
sync: change chan `closed(&mut self)` -> closed(&self)
zaharidichev Oct 10, 2020
730082f
add loom test
zaharidichev Oct 10, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
12 changes: 6 additions & 6 deletions tokio/src/sync/mpsc/bounded.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -332,11 +332,11 @@ impl<T> Sender<T> {
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx1, rx) = mpsc::channel::<()>(1);
/// let mut tx2 = tx1.clone();
/// let mut tx3 = tx1.clone();
/// let mut tx4 = tx1.clone();
/// let mut tx5 = tx1.clone();
/// let (tx1, rx) = mpsc::channel::<()>(1);
/// let tx2 = tx1.clone();
/// let tx3 = tx1.clone();
/// let tx4 = tx1.clone();
/// let tx5 = tx1.clone();
/// tokio::spawn(async move {
/// drop(rx);
/// });
Expand All @@ -351,7 +351,7 @@ impl<T> Sender<T> {
//// println!("Receiver dropped");
/// }
/// ```
pub async fn closed(&mut self) {
pub async fn closed(&self) {
self.chan.closed().await
}

Expand Down
2 changes: 1 addition & 1 deletion tokio/src/sync/mpsc/chan.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ impl<T, S: Semaphore> Tx<T, S> {
self.inner.semaphore.is_closed()
}

pub(crate) async fn closed(&mut self) {
pub(crate) async fn closed(&self) {
use std::future::Future;
use std::pin::Pin;
use std::task::Poll;
Comment on lines -150 to 153
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't know enough about the internals of channels to know whether it relies on the exclusivity of a mutable reference. However if so, we do not have any loom tests that would catch it, because such a test would not compile before this change was applied.

I would like to see a loom test that calls send and closed on the same Arc<Sender> from multiple threads concurrently.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Darksonn Good point! I will verify that.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems these are simply awaiting on a Notify so there should not be much of a problem, but still there is the loom test

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would assume the &mut self requirement (if there was one) would come from the self.closed call but that seems to take &self so this should be fine.

Expand Down
12 changes: 6 additions & 6 deletions tokio/src/sync/mpsc/unbounded.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -223,11 +223,11 @@ impl<T> UnboundedSender<T> {
///
/// #[tokio::main]
/// async fn main() {
/// let (mut tx1, rx) = mpsc::unbounded_channel::<()>();
/// let mut tx2 = tx1.clone();
/// let mut tx3 = tx1.clone();
/// let mut tx4 = tx1.clone();
/// let mut tx5 = tx1.clone();
/// let (tx1, rx) = mpsc::unbounded_channel::<()>();
/// let tx2 = tx1.clone();
/// let tx3 = tx1.clone();
/// let tx4 = tx1.clone();
/// let tx5 = tx1.clone();
/// tokio::spawn(async move {
/// drop(rx);
/// });
Expand All @@ -242,7 +242,7 @@ impl<T> UnboundedSender<T> {
//// println!("Receiver dropped");
/// }
/// ```
pub async fn closed(&mut self) {
pub async fn closed(&self) {
self.chan.closed().await
}
/// Checks if the channel has been closed. This happens when the
Expand Down
37 changes: 33 additions & 4 deletions tokio/src/sync/tests/loom_mpsc.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@ use crate::sync::mpsc;

use futures::future::poll_fn;
use loom::future::block_on;
use loom::sync::Arc;
use loom::thread;
use tokio_test::assert_ok;

#[test]
fn closing_tx() {
Expand Down Expand Up @@ -43,8 +45,8 @@ fn closing_unbounded_tx() {
#[test]
fn closing_bounded_rx() {
loom::model(|| {
let (mut tx1, rx) = mpsc::channel::<()>(16);
let mut tx2 = tx1.clone();
let (tx1, rx) = mpsc::channel::<()>(16);
let tx2 = tx1.clone();
thread::spawn(move || {
drop(rx);
});
Expand All @@ -54,11 +56,38 @@ fn closing_bounded_rx() {
});
}

#[test]
fn closing_and_sending() {
loom::model(|| {
let (tx1, mut rx) = mpsc::channel::<()>(16);
let tx1 = Arc::new(tx1);
let tx2 = tx1.clone();

let th1 = thread::spawn(move || {
tx1.try_send(()).unwrap();
});

let th2 = thread::spawn(move || {
block_on(tx2.closed());
});

let th3 = thread::spawn(move || {
let v = block_on(rx.recv());
assert!(v.is_some());
drop(rx);
});

assert_ok!(th1.join());
assert_ok!(th2.join());
assert_ok!(th3.join());
});
}

#[test]
fn closing_unbounded_rx() {
loom::model(|| {
let (mut tx1, rx) = mpsc::unbounded_channel::<()>();
let mut tx2 = tx1.clone();
let (tx1, rx) = mpsc::unbounded_channel::<()>();
let tx2 = tx1.clone();
thread::spawn(move || {
drop(rx);
});
Expand Down