Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion glommio/src/channels/shared_channel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -496,7 +496,7 @@ mod test {
use std::{
sync::{
atomic::{AtomicUsize, Ordering},
Arc,
mpsc, Arc,
},
time::Duration,
};
Expand Down Expand Up @@ -914,14 +914,17 @@ mod test {
#[test]
fn shared_drop_cascade_drop_executor_reverse() {
let (sender, receiver) = new_bounded(100);
let (connected_tx, connected_rx) = mpsc::channel();

let original = Arc::new(AtomicUsize::new(0));
let send_count = original.clone();
let drop_count = original.clone();
let sender_connected = connected_tx.clone();

let ex1 = LocalExecutorBuilder::default()
.spawn(move || async move {
let sender = sender.connect().await;
sender_connected.send(()).unwrap();
for x in 0..50 {
let val = WithDrop(send_count.clone(), x);
drop_count.fetch_add(1, Ordering::SeqCst);
Expand All @@ -933,13 +936,19 @@ mod test {
let ex2 = LocalExecutorBuilder::default()
.spawn(move || async move {
let receiver = receiver.connect().await;
connected_tx.send(()).unwrap();
for x in 0..50 {
let resp = receiver.recv().await.unwrap();
assert_eq!(x, resp.1);
}
})
.unwrap();

// Both endpoints must be connected before destroying the sender's
// executor. Otherwise the receiver can observe its peer ID after the
// sender's reactor has already removed the corresponding notifier.
connected_rx.recv().unwrap();
connected_rx.recv().unwrap();
drop(ex1);
ex2.join().unwrap();

Expand Down
Loading