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: expose strong and weak counts of mpsc sender handles #6405

Merged
merged 4 commits into from
Mar 22, 2024
Merged

sync: expose strong and weak counts of mpsc sender handles #6405

Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fc3791f
expose strong and weak count for mpsc senders
maminrayej Mar 16, 2024
afeb31c
increment weak count when cloning a weak handle
maminrayej Mar 16, 2024
2ccad6a
split big tests into smaller tests
maminrayej Mar 22, 2024
f57bfc7
Merge branch 'master' into expose-strong-and-weak-counts
maminrayej Mar 22, 2024
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
Prev Previous commit
Next Next commit
split big tests into smaller tests
  • Loading branch information
@maminrayej
maminrayej committed Mar 22, 2024
commit 2ccad6accf6f92ee3ad121885075979f17d90083
191 changes: 87 additions & 104 deletions tokio/tests/sync_mpsc_weak.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,8 @@ use wasm_bindgen_test::wasm_bindgen_test as test;

use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::{Acquire, Release};
use std::sync::Arc;
use tokio::sync::mpsc::{self, channel, unbounded_channel};
use tokio::sync::{oneshot, Notify};
use tokio::sync::oneshot;

#[tokio::test]
async fn weak_sender() {
Expand Down Expand Up @@ -514,159 +513,143 @@ fn test_tx_count_weak_unbounded_sender() {
}

#[tokio::test]
maminrayej marked this conversation as resolved.
Show resolved Hide resolved
async fn strong_and_weak_count() {
async fn sender_strong_count_when_cloned() {
let (tx, _rx) = mpsc::channel::<()>(1);

let first_downgrade = Arc::new(Notify::new());
let second_downgrade = Arc::new(Notify::new());
let tx2 = tx.clone();

let task_handle = {
let tx = tx.clone();
let first_downgrade = first_downgrade.clone();
let second_downgrade = second_downgrade.clone();
assert_eq!(tx.strong_count(), 2);
assert_eq!(tx2.strong_count(), 2);
}

tokio::spawn(async move {
let weak = tx.downgrade();
#[tokio::test]
async fn sender_weak_count_when_downgraded() {
let (tx, _rx) = mpsc::channel::<()>(1);

assert_eq!(tx.weak_count(), 1);
assert_eq!(tx.strong_count(), 2);
assert_eq!(weak.weak_count(), 1);
assert_eq!(weak.strong_count(), 2);
let weak = tx.downgrade();

let weak2 = weak.clone();
assert_eq!(tx.weak_count(), 1);
assert_eq!(weak.weak_count(), 1);
}

assert_eq!(tx.weak_count(), 2);
assert_eq!(tx.strong_count(), 2);
assert_eq!(weak.weak_count(), 2);
assert_eq!(weak.strong_count(), 2);
assert_eq!(weak2.weak_count(), 2);
assert_eq!(weak2.strong_count(), 2);
#[tokio::test]
async fn sender_strong_count_when_dropped() {
let (tx, _rx) = mpsc::channel::<()>(1);

let tx2 = tx.clone();

first_downgrade.notify_one();
drop(tx2);

second_downgrade.notified().await;
assert_eq!(tx.strong_count(), 1);
}

assert_eq!(tx.weak_count(), 3);
assert_eq!(tx.strong_count(), 2);
assert_eq!(weak.weak_count(), 3);
assert_eq!(weak.strong_count(), 2);
assert_eq!(weak2.weak_count(), 3);
assert_eq!(weak2.strong_count(), 2);
#[tokio::test]
async fn sender_weak_count_when_dropped() {
let (tx, _rx) = mpsc::channel::<()>(1);

drop(weak);
assert_eq!(tx.weak_count(), 2);
assert_eq!(tx.strong_count(), 2);
assert_eq!(weak2.weak_count(), 2);
assert_eq!(weak2.strong_count(), 2);
let weak = tx.downgrade();

drop(weak2);
assert_eq!(tx.weak_count(), 1);
assert_eq!(tx.strong_count(), 2);
drop(weak);

drop(tx);
})
};
assert_eq!(tx.weak_count(), 0);
}

first_downgrade.notified().await;
#[tokio::test]
async fn sender_strong_and_weak_conut() {
let (tx, _rx) = mpsc::channel::<()>(1);

assert_eq!(tx.weak_count(), 2);
assert_eq!(tx.strong_count(), 2);
let tx2 = tx.clone();

let weak = tx.downgrade();
let weak2 = tx2.downgrade();

assert_eq!(tx.weak_count(), 3);
assert_eq!(tx.strong_count(), 2);
assert_eq!(weak.weak_count(), 3);
assert_eq!(tx2.strong_count(), 2);
assert_eq!(weak.strong_count(), 2);
assert_eq!(weak2.strong_count(), 2);

second_downgrade.notify_one();
assert_eq!(tx.weak_count(), 2);
assert_eq!(tx2.weak_count(), 2);
assert_eq!(weak.weak_count(), 2);
assert_eq!(weak2.weak_count(), 2);

task_handle.await.unwrap();
drop(tx2);
drop(weak2);

assert_eq!(tx.weak_count(), 1);
assert_eq!(tx.strong_count(), 1);
assert_eq!(weak.strong_count(), 1);

drop(weak);

assert_eq!(tx.weak_count(), 0);
assert_eq!(tx.strong_count(), 1);
assert_eq!(tx.weak_count(), 1);
assert_eq!(weak.weak_count(), 1);
}

#[tokio::test]
async fn unbounded_strong_and_weak_count() {
async fn unbounded_sender_strong_count_when_cloned() {
let (tx, _rx) = mpsc::unbounded_channel::<()>();

let first_downgrade = Arc::new(Notify::new());
let second_downgrade = Arc::new(Notify::new());
let tx2 = tx.clone();

let task_handle = {
let tx = tx.clone();
let first_downgrade = first_downgrade.clone();
let second_downgrade = second_downgrade.clone();
assert_eq!(tx.strong_count(), 2);
assert_eq!(tx2.strong_count(), 2);
}

tokio::spawn(async move {
let weak = tx.downgrade();
#[tokio::test]
async fn unbounded_sender_weak_count_when_downgraded() {
let (tx, _rx) = mpsc::unbounded_channel::<()>();

assert_eq!(tx.weak_count(), 1);
assert_eq!(tx.strong_count(), 2);
assert_eq!(weak.weak_count(), 1);
assert_eq!(weak.strong_count(), 2);
let weak = tx.downgrade();

let weak2 = weak.clone();
assert_eq!(tx.weak_count(), 1);
assert_eq!(weak.weak_count(), 1);
}

assert_eq!(tx.weak_count(), 2);
assert_eq!(tx.strong_count(), 2);
assert_eq!(weak.weak_count(), 2);
assert_eq!(weak.strong_count(), 2);
assert_eq!(weak2.weak_count(), 2);
assert_eq!(weak2.strong_count(), 2);
#[tokio::test]
async fn unbounded_sender_strong_count_when_dropped() {
let (tx, _rx) = mpsc::unbounded_channel::<()>();

let tx2 = tx.clone();

first_downgrade.notify_one();
drop(tx2);

second_downgrade.notified().await;
assert_eq!(tx.strong_count(), 1);
}

assert_eq!(tx.weak_count(), 3);
assert_eq!(tx.strong_count(), 2);
assert_eq!(weak.weak_count(), 3);
assert_eq!(weak.strong_count(), 2);
assert_eq!(weak2.weak_count(), 3);
assert_eq!(weak2.strong_count(), 2);
#[tokio::test]
async fn unbounded_sender_weak_count_when_dropped() {
let (tx, _rx) = mpsc::unbounded_channel::<()>();

drop(weak);
assert_eq!(tx.weak_count(), 2);
assert_eq!(tx.strong_count(), 2);
assert_eq!(weak2.weak_count(), 2);
assert_eq!(weak2.strong_count(), 2);
let weak = tx.downgrade();

drop(weak2);
assert_eq!(tx.weak_count(), 1);
assert_eq!(tx.strong_count(), 2);
drop(weak);

drop(tx);
})
};
assert_eq!(tx.weak_count(), 0);
}

first_downgrade.notified().await;
#[tokio::test]
async fn unbounded_sender_strong_and_weak_conut() {
let (tx, _rx) = mpsc::unbounded_channel::<()>();

assert_eq!(tx.weak_count(), 2);
assert_eq!(tx.strong_count(), 2);
let tx2 = tx.clone();

let weak = tx.downgrade();
let weak2 = tx2.downgrade();

assert_eq!(tx.weak_count(), 3);
assert_eq!(tx.strong_count(), 2);
assert_eq!(weak.weak_count(), 3);
assert_eq!(tx2.strong_count(), 2);
assert_eq!(weak.strong_count(), 2);
assert_eq!(weak2.strong_count(), 2);

second_downgrade.notify_one();
assert_eq!(tx.weak_count(), 2);
assert_eq!(tx2.weak_count(), 2);
assert_eq!(weak.weak_count(), 2);
assert_eq!(weak2.weak_count(), 2);

task_handle.await.unwrap();
drop(tx2);
drop(weak2);

assert_eq!(tx.weak_count(), 1);
assert_eq!(tx.strong_count(), 1);
assert_eq!(weak.strong_count(), 1);

drop(weak);

assert_eq!(tx.weak_count(), 0);
assert_eq!(tx.strong_count(), 1);
assert_eq!(tx.weak_count(), 1);
assert_eq!(weak.weak_count(), 1);
}
Loading