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

Replace async-channel with flume #83

Merged
merged 2 commits into from
Apr 8, 2024
Merged

Replace async-channel with flume #83

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
7fb3b9b
tests: Replace async-channel with flume
notgull Apr 8, 2024
68b7182
tests: ignore barrier::smoke on MIRI as well
notgull Apr 8, 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
2 changes: 1 addition & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,8 @@ default = ["std"]
std = ["event-listener/std", "event-listener-strategy/std"]

[dev-dependencies]
async-channel = "2.2.0"
fastrand = "2.0.0"
flume = "0.11.0"
futures-lite = "2.0.0"
waker-fn = "1.1.0"

Expand Down
2 changes: 2 additions & 0 deletions src/barrier.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,8 @@ impl Barrier {
/// println!("after wait");
/// });
/// }
/// # // Wait for threads to stop.
/// # std::thread::sleep(std::time::Duration::from_secs(1));
/// ```
#[cfg(all(feature = "std", not(target_family = "wasm")))]
pub fn wait_blocking(&self) -> BarrierWaitResult {
Expand Down
13 changes: 7 additions & 6 deletions tests/barrier.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,15 @@ use async_lock::Barrier;
use futures_lite::future;

#[test]
#[cfg_attr(miri, ignore)]
fn smoke() {
future::block_on(async move {
const N: usize = 10;

let barrier = Arc::new(Barrier::new(N));

for _ in 0..10 {
let (tx, rx) = async_channel::unbounded();
let (tx, rx) = flume::unbounded();

for _ in 0..N - 1 {
let c = barrier.clone();
Expand All @@ -21,7 +22,7 @@ fn smoke() {
thread::spawn(move || {
future::block_on(async move {
let res = c.wait().await;
tx.send(res.is_leader()).await.unwrap();
tx.send_async(res.is_leader()).await.unwrap();
})
});
}
Expand All @@ -35,7 +36,7 @@ fn smoke() {

// Now, the barrier is cleared and we should get data.
for _ in 0..N - 1 {
if rx.recv().await.unwrap() {
if rx.recv_async().await.unwrap() {
assert!(!leader_found);
leader_found = true;
}
Expand All @@ -55,15 +56,15 @@ fn smoke_blocking() {
let barrier = Arc::new(Barrier::new(N));

for _ in 0..10 {
let (tx, rx) = async_channel::unbounded();
let (tx, rx) = flume::unbounded();

for _ in 0..N - 1 {
let c = barrier.clone();
let tx = tx.clone();

thread::spawn(move || {
let res = c.wait_blocking();
tx.send_blocking(res.is_leader()).unwrap();
tx.send(res.is_leader()).unwrap();
});
}

Expand All @@ -76,7 +77,7 @@ fn smoke_blocking() {

// Now, the barrier is cleared and we should get data.
for _ in 0..N - 1 {
if rx.recv().await.unwrap() {
if rx.recv_async().await.unwrap() {
assert!(!leader_found);
leader_found = true;
}
Expand Down
6 changes: 3 additions & 3 deletions tests/mutex.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ fn get_mut() {
#[test]
fn contention() {
future::block_on(async {
let (tx, rx) = async_channel::unbounded();
let (tx, rx) = flume::unbounded();

let tx = Arc::new(tx);
let mutex = Arc::new(Mutex::new(0i32));
Expand All @@ -77,14 +77,14 @@ fn contention() {
future::block_on(async move {
let mut lock = mutex.lock().await;
*lock += 1;
tx.send(()).await.unwrap();
tx.send_async(()).await.unwrap();
drop(lock);
})
});
}

for _ in 0..num_tasks {
rx.recv().await.unwrap();
rx.recv_async().await.unwrap();
}

let lock = mutex.lock().await;
Expand Down
37 changes: 19 additions & 18 deletions tests/rwlock.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,13 +25,13 @@ wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);

#[cfg(not(target_family = "wasm"))]
fn spawn<T: Send + 'static>(f: impl Future<Output = T> + Send + 'static) -> future::Boxed<T> {
let (s, r) = async_channel::bounded(1);
let (s, r) = flume::bounded(1);
thread::spawn(move || {
future::block_on(async {
let _ = s.send(f.await).await;
let _ = s.send_async(f.await).await;
})
});
async move { r.recv().await.unwrap() }.boxed()
async move { r.recv_async().await.unwrap() }.boxed()
}

#[test]
Expand Down Expand Up @@ -119,13 +119,14 @@ fn get_mut() {
}

// Miri bug; this works when async is replaced with blocking
#[cfg(not(any(target_family = "wasm", miri)))]
#[cfg(not(target_family = "wasm"))]
#[test]
#[cfg_attr(miri, ignore)]
fn contention() {
const N: u32 = 10;
const M: usize = if cfg!(miri) { 100 } else { 1000 };

let (tx, rx) = async_channel::unbounded();
let (tx, rx) = flume::unbounded();
let tx = Arc::new(tx);
let rw = Arc::new(RwLock::new(()));

Expand All @@ -142,25 +143,25 @@ fn contention() {
drop(rw.read().await);
}
}
tx.send(()).await.unwrap();
tx.send_async(()).await.unwrap();
});
}

future::block_on(async move {
for _ in 0..N {
rx.recv().await.unwrap();
rx.recv_async().await.unwrap();
}
});
}

// Miri bug; this works when async is replaced with blocking
#[cfg(not(any(target_family = "wasm", miri)))]
#[cfg(not(target_family = "wasm"))]
#[test]
#[cfg_attr(miri, ignore)]
fn contention_arc() {
const N: u32 = 10;
const M: usize = if cfg!(miri) { 100 } else { 1000 };

let (tx, rx) = async_channel::unbounded();
let (tx, rx) = flume::unbounded();
let tx = Arc::new(tx);
let rw = Arc::new(RwLock::new(()));

Expand All @@ -177,13 +178,13 @@ fn contention_arc() {
drop(rw.read_arc().await);
}
}
tx.send(()).await.unwrap();
tx.send_async(()).await.unwrap();
});
}

future::block_on(async move {
for _ in 0..N {
rx.recv().await.unwrap();
rx.recv_async().await.unwrap();
}
});
}
Expand All @@ -192,7 +193,7 @@ fn contention_arc() {
#[test]
fn writer_and_readers() {
let lock = Arc::new(RwLock::new(0i32));
let (tx, rx) = async_channel::unbounded();
let (tx, rx) = flume::unbounded();

// Spawn a writer task.
let _spawned = spawn({
Expand All @@ -205,7 +206,7 @@ fn writer_and_readers() {
future::yield_now().await;
*lock = tmp + 1;
}
tx.send(()).await.unwrap();
tx.send_async(()).await.unwrap();
}
});

Expand All @@ -228,7 +229,7 @@ fn writer_and_readers() {
}

// Wait for writer to finish.
rx.recv().await.unwrap();
rx.recv_async().await.unwrap();
let lock = lock.read().await;
assert_eq!(*lock, 1000);
});
Expand All @@ -238,7 +239,7 @@ fn writer_and_readers() {
#[test]
fn writer_and_readers_arc() {
let lock = Arc::new(RwLock::new(0i32));
let (tx, rx) = async_channel::unbounded();
let (tx, rx) = flume::unbounded();

// Spawn a writer task.
let _spawned = spawn({
Expand All @@ -251,7 +252,7 @@ fn writer_and_readers_arc() {
future::yield_now().await;
*lock = tmp + 1;
}
tx.send(()).await.unwrap();
tx.send_async(()).await.unwrap();
}
});

Expand All @@ -274,7 +275,7 @@ fn writer_and_readers_arc() {
}

// Wait for writer to finish.
rx.recv().await.unwrap();
rx.recv_async().await.unwrap();
let lock = lock.read_arc().await;
assert_eq!(*lock, 1000);
});
Expand Down
Loading