Compare commits

...
Author SHA1 Message Date
Alice Ryhl 66f836e61a tests: mark failing taskdump tests as #[ignore] (#8403) 2026-09-03 14:31:54 +00:00
Alice Ryhl 16a6a2390c ci: pin cargo-fuzz to 0.13.1 for check-fuzzing (#8403) 2026-09-03 13:58:02 +00:00
Joel Dice 705989c98b ci: pin Wasmtime version(s) in CI (#8314)
Per https://github.com/bytecodealliance/wasmtime/pull/13558, Wasmtime v46.0.1
was the last release to support `wasm32-wasip1-threads`, so we use that for the
WASIp1 testing.

For WASIp2, we should be able to use any recent version of Wasmtime, but we pin
to a specific version anyway to avoid surprises.

(cherry picked from commit 5760ccdc37)
2026-09-03 13:58:02 +00:00
fly1d 9c880b082b ci: skip workspace semver on release pushes (#8390)
`github.event.pull_request.base.ref` is empty for push events, so the workspace semver check runs on `tokio-1.*.x` branches. Also check `github.ref_name` to preserve the intended release-branch exclusion while leaving pull request behavior unchanged.

Fixes: #8389
2026-08-27 11:14:39 +02:00
Alice Ryhl 11bfc1345b chore: prepare Tokio v1.47.5 (#8122) 2026-05-07 13:55:35 +02:00
Alice Ryhl f085b6211b sync: notify receivers in mpsc OwnedPermit::release() method (#8075) 2026-05-07 09:32:14 +02:00
Alice Ryhl 30d25ccb8b sync: require that an RwLock has max_readers != 0 (#8076) 2026-05-07 09:31:12 +02:00
Alice Ryhl 9fccf5339d sync: return Empty from try_recv() when mpsc is closed with outstanding permits (#8074) 2026-05-07 09:30:50 +02:00
Alice Ryhl ebf61b45b5 sync: fix underflow in mpsc channel len() (#8062) 2026-05-07 09:29:33 +02:00
Alice RyhlandMattia Pitossi 670a907c55 ci: fix CI on tokio-1.47.x (#8101)
Co-authored-by: Mattia Pitossi <[email protected]>
2026-05-02 13:33:40 +02:00
Alice Ryhl aa65d0d0b8 chore: prepare Tokio v1.47.4 (#8002) 2026-04-02 14:12:06 +02:00
LeoniePhiline bf18ed452d sync: fix panic in Chan::recv_many when called with non-empty vector on closed channel (#7991)
`Chan::recv_many` intends to assert that no slots
have been consumed when exiting with `Ready` via
the `rx_closed` code path.

Instead of asserting no items were added to the
buffer, it asserted buffer emptiness, incorrectly
making assumptions about the provided buffer.

When `recv_many` was called on an empty channel
with idle semaphore after the receiver was closed,
the method would panic.

The branch coverage had been previously missing.

This changeset corrects the assertion
and adds tests covering the code path.

Fixes #7990.
2026-04-02 13:16:03 +02:00
17 changed files with 373 additions and 118 deletions
-58
View File
@@ -1,58 +0,0 @@
only_if: $CIRRUS_TAG == '' && ($CIRRUS_PR != '' || $CIRRUS_BRANCH == 'master' || $CIRRUS_BRANCH =~ 'tokio-.*')
auto_cancellation: $CIRRUS_BRANCH != 'master' && $CIRRUS_BRANCH !=~ 'tokio-.*'
freebsd_instance:
image_family: freebsd-14-3
env:
RUST_STABLE: stable
RUST_NIGHTLY: nightly-2025-01-25
RUSTFLAGS: -D warnings
# Test FreeBSD in a full VM on cirrus-ci.com. Test the i686 target too, in the
# same VM. The binary will be built in 32-bit mode, but will execute on a
# 64-bit kernel and in a 64-bit environment. Our tests don't execute any of
# the system's binaries, so the environment shouldn't matter.
task:
name: FreeBSD 64-bit
setup_script:
- pkg install -y bash
- curl https://sh.rustup.rs -sSf --output rustup.sh
- sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
- . $HOME/.cargo/env
- |
echo "~~~~ rustc --version ~~~~"
rustc --version
test_script:
- . $HOME/.cargo/env
- cargo test --all --all-features
task:
name: FreeBSD docs
env:
RUSTFLAGS: --cfg docsrs --cfg tokio_unstable
RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings
setup_script:
- pkg install -y bash
- curl https://sh.rustup.rs -sSf --output rustup.sh
- sh rustup.sh -y --profile minimal --default-toolchain $RUST_NIGHTLY
- . $HOME/.cargo/env
- |
echo "~~~~ rustc --version ~~~~"
rustc --version
test_script:
- . $HOME/.cargo/env
- cargo doc --lib --no-deps --all-features --document-private-items
task:
name: FreeBSD 32-bit
setup_script:
- pkg install -y bash
- curl https://sh.rustup.rs -sSf --output rustup.sh
- sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
- . $HOME/.cargo/env
- rustup target add i686-unknown-freebsd
- |
echo "~~~~ rustc --version ~~~~"
rustc --version
test_script:
- . $HOME/.cargo/env
- cargo test --all --all-features --target i686-unknown-freebsd
+77 -3
View File
@@ -515,7 +515,7 @@ jobs:
package: tokio
release-type: minor
- name: Check semver for rest of the workspace
if: ${{ !startsWith(github.event.pull_request.base.ref, 'tokio-1.') }}
if: ${{ !startsWith(github.event.pull_request.base.ref, 'tokio-1.') && !startsWith(github.ref_name, 'tokio-1.') }}
uses: obi1kenobi/cargo-semver-checks-action@v2
with:
rust-toolchain: ${{ env.rust_stable }}
@@ -1071,7 +1071,10 @@ jobs:
- name: Install cargo-hack, wasmtime
uses: taiki-e/install-action@v2
with:
tool: cargo-hack,wasmtime
# Wasmtime v46.0.1 is the last version to support
# `wasm32-wasip1-threads` (which was an experiment that was never
# standardized):
tool: cargo-hack,[email protected]
- uses: Swatinem/rust-cache@v2
- name: WASI test tokio full
@@ -1152,7 +1155,7 @@ jobs:
toolchain: ${{ env.rust_nightly }}
- uses: Swatinem/rust-cache@v2
- name: Install cargo-fuzz
run: cargo install cargo-fuzz
run: cargo install --locked cargo-fuzz --version 0.13.1
- name: Check /tokio/
run: cargo fuzz check --all-features
working-directory: tokio
@@ -1231,3 +1234,74 @@ jobs:
echo 'Please remove trailing whitespace from these lines.'
exit 1
fi
freebsd-x86_64:
name: FreeBSD x86_64
needs: basics
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6
- name: Test in FreeBSD
uses: vmactions/freebsd-vm@v1
with:
release: '14.4'
envs: "RUSTFLAGS"
prepare: |
pkg install -y curl
curl https://sh.rustup.rs -sSf --output rustup.sh
sh rustup.sh -y --profile minimal --default-toolchain ${{ env.rust_stable }}
run: |
. $HOME/.cargo/env
cargo test --workspace --features full,test-util
# Enable all unstable features except `io_uring` and `taskdump`,
# which are Linux-only features.
RUSTFLAGS="$RUSTFLAGS --cfg tokio_unstable" \
cargo test \
--features full,test-util,tracing
freebsd-docs:
name: FreeBSD docs
needs: basics
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6
- name: Test in FreeBSD
uses: vmactions/freebsd-vm@v1
env:
RUSTFLAGS: --cfg docsrs --cfg tokio_unstable
RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings
with:
release: '14.4'
envs: "RUST_NIGHTLY RUSTDOCFLAGS RUSTFLAGS"
prepare: |
pkg install -y curl
curl https://sh.rustup.rs -sSf --output rustup.sh
sh rustup.sh -y --profile minimal --default-toolchain ${{ env.rust_nightly }}
run: |
. $HOME/.cargo/env
# We use `--features full,test-util,io-uring,tracing` instead of
# `--all-features` to exclude `taskdump` and `io_uring`, which are Linux-only
# features.
cargo doc --lib --no-deps --features full,test-util,tracing \
--document-private-items
freebsd-i686:
name: FreeBSD i686
needs: basics
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6
- name: Test in FreeBSD
uses: vmactions/freebsd-vm@v1
with:
release: '14.4'
envs: "RUSTFLAGS"
prepare: |
pkg install -y curl
curl https://sh.rustup.rs -sSf --output rustup.sh
sh rustup.sh -y --profile minimal --default-toolchain ${{ env.rust_stable }}
run: |
. $HOME/.cargo/env
rustup target add i686-unknown-freebsd
cargo test --workspace --features full,test-util \
--target i686-unknown-freebsd
Generated
+5 -5
View File
@@ -123,9 +123,9 @@ checksum = "46c5e41b57b8bba42a04676d81cb89e9ee8e859a1a66f80a5a72e1cb76b34d43"
[[package]]
name = "bytes"
version = "1.10.1"
version = "1.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a"
checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33"
[[package]]
name = "cast"
@@ -1050,9 +1050,9 @@ checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
[[package]]
name = "rand"
version = "0.9.2"
version = "0.9.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1"
checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea"
dependencies = [
"rand_chacha",
"rand_core",
@@ -1405,7 +1405,7 @@ dependencies = [
[[package]]
name = "tokio"
version = "1.47.2"
version = "1.47.5"
dependencies = [
"async-stream",
"backtrace",
+1 -1
View File
@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
```toml
[dependencies]
tokio = { version = "1.47.3", features = ["full"] }
tokio = { version = "1.47.5", features = ["full"] }
```
Then, on your main.rs:
+23 -1
View File
@@ -1,4 +1,26 @@
# 1.47.3 (Januar 3rd, 2026)
# 1.47.5 (May 7th, 2026)
### Fixed
* sync: fix underflow in mpsc channel `len()` ([#8062])
* sync: notify receivers in mpsc `OwnedPermit::release()` method ([#8075])
* sync: require that an `RwLock` has `max_readers != 0` ([#8076])
* sync: return `Empty` from `try_recv()` when mpsc is closed with outstanding permits ([#8074])
[#8062]: https://github.com/tokio-rs/tokio/pull/8062
[#8074]: https://github.com/tokio-rs/tokio/pull/8074
[#8075]: https://github.com/tokio-rs/tokio/pull/8075
[#8076]: https://github.com/tokio-rs/tokio/pull/8076
# 1.47.4 (April 2nd, 2026)
### Fixed
* sync: fix panic in `Chan::recv_many` when called with non-empty vector on closed channel ([#7991])
[#7991]: https://github.com/tokio-rs/tokio/pull/7991
# 1.47.3 (January 3rd, 2026)
### Fixed
+1 -1
View File
@@ -6,7 +6,7 @@ name = "tokio"
# - README.md
# - Update CHANGELOG.md.
# - Create "v1.x.y" git tag.
version = "1.47.3"
version = "1.47.5"
edition = "2021"
rust-version = "1.70"
authors = ["Tokio Contributors <[email protected]>"]
+1 -1
View File
@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
```toml
[dependencies]
tokio = { version = "1.47.3", features = ["full"] }
tokio = { version = "1.47.5", features = ["full"] }
```
Then, on your main.rs:
-5
View File
@@ -211,11 +211,6 @@ impl<T> Block<T> {
self.header.ready_slots.fetch_or(TX_CLOSED, Release);
}
pub(crate) unsafe fn is_closed(&self) -> bool {
let ready_bits = self.header.ready_slots.load(Acquire);
is_tx_closed(ready_bits)
}
/// Resets the block to a blank state. This enables reusing blocks in the
/// channel.
///
+3 -16
View File
@@ -1844,14 +1844,12 @@ impl<T> OwnedPermit<T> {
///
/// [`Sender`]: Sender
pub fn release(mut self) -> Sender<T> {
use chan::Semaphore;
let chan = self.chan.take().unwrap_or_else(|| {
unreachable!("OwnedPermit channel is only taken when the permit is moved")
});
// Add the permit back to the semaphore
chan.semaphore().add_permit();
drop(Permit { chan: &chan });
Sender { chan }
}
@@ -1910,21 +1908,10 @@ impl<T> OwnedPermit<T> {
impl<T> Drop for OwnedPermit<T> {
fn drop(&mut self) {
use chan::Semaphore;
// Are we still holding onto the sender?
if let Some(chan) = self.chan.take() {
let semaphore = chan.semaphore();
// Add the permit back to the semaphore
semaphore.add_permit();
// If this `OwnedPermit` is holding the last sender for this
// channel, wake the receiver so that it can be notified that the
// channel is closed.
if semaphore.is_closed() && semaphore.is_idle() {
chan.wake_rx();
}
// Reuse Drop impl of non-owned Permit.
drop(Permit { chan: &chan });
}
// Otherwise, do nothing.
+8 -10
View File
@@ -306,13 +306,11 @@ impl<T, S: Semaphore> Rx<T, S> {
return Ready(Some(value));
}
Some(Read::Closed) => {
// TODO: This check may not be required as it most
// likely can only return `true` at this point. A
// channel is closed when all tx handles are
// A channel is closed when all tx handles are
// dropped. Dropping a tx handle releases memory,
// which ensures that if dropping the tx handle is
// visible, then all messages sent are also visible.
assert!(self.inner.semaphore.is_idle());
debug_assert!(self.inner.semaphore.is_idle());
coop.made_progress();
return Ready(None);
}
@@ -380,13 +378,11 @@ impl<T, S: Semaphore> Rx<T, S> {
if number_added > 0 {
self.inner.semaphore.add_permits(number_added);
}
// TODO: This check may not be required as it most
// likely can only return `true` at this point. A
// channel is closed when all tx handles are
// A channel is closed when all tx handles are
// dropped. Dropping a tx handle releases memory,
// which ensures that if dropping the tx handle is
// visible, then all messages sent are also visible.
assert!(self.inner.semaphore.is_idle());
debug_assert!(self.inner.semaphore.is_idle());
coop.made_progress();
return Ready(number_added);
}
@@ -415,7 +411,7 @@ impl<T, S: Semaphore> Rx<T, S> {
try_recv!();
if rx_fields.rx_closed && self.inner.semaphore.is_idle() {
assert!(buffer.is_empty());
debug_assert_eq!(buffer.len(), initial_length);
coop.made_progress();
Ready(0usize)
} else {
@@ -440,7 +436,9 @@ impl<T, S: Semaphore> Rx<T, S> {
}
TryPopResult::Closed => return Err(TryRecvError::Disconnected),
// If close() was called, an empty queue should report Disconnected.
TryPopResult::Empty if rx_fields.rx_closed => {
TryPopResult::Empty
if rx_fields.rx_closed && self.inner.semaphore.is_idle() =>
{
return Err(TryRecvError::Disconnected)
}
TryPopResult::Empty => return Err(TryRecvError::Empty),
+50 -12
View File
@@ -224,15 +224,6 @@ impl<T> Tx<T> {
let _ = Box::from_raw(block.as_ptr());
}
}
pub(crate) fn is_closed(&self) -> bool {
let tail = self.block_tail.load(Acquire);
unsafe {
let tail_block = &*tail;
tail_block.is_closed()
}
}
}
impl<T> fmt::Debug for Tx<T> {
@@ -256,11 +247,58 @@ impl<T> Rx<T> {
self.len(tx) == 0
}
// Guaranteed to return true if `slot_index` is the fake message sent on channel close.
// Guaranteed to return false if `slot_index` is a fully sent message.
//
// For messages that are partially sent, may return either true or false.
fn is_maybe_closed(&self, tx: &Tx<T>, slot_index: usize) -> bool {
let start_index = block::start_index(slot_index);
let tail = tx.block_tail.load(Acquire);
// SAFETY: Only the receiver frees blocks, so since we are the receiver, this will not be
// freed right now.
let tail_ref = unsafe { &*tail };
if tail_ref.is_at_index(start_index) {
return !tail_ref.has_value(slot_index);
}
// This method is optimized for checking whether the last value is present, so most of the
// time it is in `block_tail`. However, this isn't always the case since it's possible
// that the list was grown with an empty block, in which case `block_tail` points one block
// too far. To handle this case, we walk the list from the head.
let mut block_ptr = Some(self.head);
while let Some(block) = block_ptr {
// SAFETY: Only the receiver frees blocks, so since we are the receiver, this will not
// be freed right now.
let block_ref = unsafe { block.as_ref() };
if block_ref.is_at_index(start_index) {
return !block_ref.has_value(slot_index);
}
block_ptr = block_ref.load_next(Acquire);
}
true
}
pub(crate) fn len(&self, tx: &Tx<T>) -> usize {
// When all the senders are dropped, there will be a last block in the tail position,
// but it will be closed
let tail_position = tx.tail_position.load(Acquire);
tail_position - self.index - (tx.is_closed() as usize)
let mut len = tail_position.wrapping_sub(self.index);
debug_assert!(0 <= len as isize);
if len == 0 {
return 0;
}
// There are messages present in the queue. However, it's possible that the last message is
// a fake "closed" message that we do not wish to count. To avoid counting it, we do not
// count the last message if the ready bit is unset.
//
// Note that it is also possible for the ready bit to be unset on a normal message, but
// this happens only if that message is currently being sent *right now* in parallel on
// another thread. That is okay because it is optional to count messages that are currently
// being sent.
if self.is_maybe_closed(tx, tail_position.wrapping_sub(1)) {
len -= 1;
}
len
}
/// Pops the next value off the queue.
+3 -3
View File
@@ -135,10 +135,10 @@ pub mod error;
/// This value must be a power of 2. It also must be smaller than the number of
/// bits in `usize`.
#[cfg(all(target_pointer_width = "64", not(loom)))]
const BLOCK_CAP: usize = 32;
pub(crate) const BLOCK_CAP: usize = 32;
#[cfg(all(not(target_pointer_width = "64"), not(loom)))]
const BLOCK_CAP: usize = 16;
pub(crate) const BLOCK_CAP: usize = 16;
#[cfg(loom)]
const BLOCK_CAP: usize = 2;
pub(crate) const BLOCK_CAP: usize = 2;
+11 -1
View File
@@ -266,12 +266,13 @@ impl<T: ?Sized> RwLock<T> {
///
/// # Panics
///
/// Panics if `max_reads` is more than `u32::MAX >> 3`.
/// Panics if `max_reads` is `0` or is bigger than `u32::MAX >> 3`.
#[track_caller]
pub fn with_max_readers(value: T, max_reads: u32) -> RwLock<T>
where
T: Sized,
{
assert_ne!(max_reads, 0, "a RwLock may not be created with 0 readers");
assert!(
max_reads <= MAX_READS,
"a RwLock may not be created with more than {MAX_READS} readers"
@@ -367,11 +368,16 @@ impl<T: ?Sized> RwLock<T> {
///
/// static LOCK: RwLock<i32> = RwLock::const_with_max_readers(5, 1024);
/// ```
///
/// # Panics
///
/// Panics if `max_reads` is `0` or is bigger than `u32::MAX >> 3`.
#[cfg(not(all(loom, test)))]
pub const fn const_with_max_readers(value: T, max_reads: u32) -> RwLock<T>
where
T: Sized,
{
assert!(max_reads != 0, "a RwLock may not be created with 0 readers");
assert!(max_reads <= MAX_READS);
RwLock {
@@ -771,6 +777,7 @@ impl<T: ?Sized> RwLock<T> {
/// ```
pub async fn write(&self) -> RwLockWriteGuard<'_, T> {
let acquire_fut = async {
debug_assert_ne!(self.mr, 0);
self.s.acquire(self.mr as usize).await.unwrap_or_else(|_| {
// The semaphore was closed. but, we never explicitly close it, and we have a
// handle to it through the Arc, which means that this can never happen.
@@ -906,6 +913,7 @@ impl<T: ?Sized> RwLock<T> {
let resource_span = self.resource_span.clone();
let acquire_fut = async {
debug_assert_ne!(self.mr, 0);
self.s.acquire(self.mr as usize).await.unwrap_or_else(|_| {
// The semaphore was closed. but, we never explicitly close it, and we have a
// handle to it through the Arc, which means that this can never happen.
@@ -970,6 +978,7 @@ impl<T: ?Sized> RwLock<T> {
/// }
/// ```
pub fn try_write(&self) -> Result<RwLockWriteGuard<'_, T>, TryLockError> {
debug_assert_ne!(self.mr, 0);
match self.s.try_acquire(self.mr as usize) {
Ok(permit) => permit,
Err(TryAcquireError::NoPermits) => return Err(TryLockError(())),
@@ -1028,6 +1037,7 @@ impl<T: ?Sized> RwLock<T> {
/// }
/// ```
pub fn try_write_owned(self: Arc<Self>) -> Result<OwnedRwLockWriteGuard<T>, TryLockError> {
debug_assert_ne!(self.mr, 0);
match self.s.try_acquire(self.mr as usize) {
Ok(permit) => permit,
Err(TryAcquireError::NoPermits) => return Err(TryLockError(())),
+54 -1
View File
@@ -1,4 +1,4 @@
use crate::sync::mpsc;
use crate::sync::mpsc::{self, BLOCK_CAP};
use loom::future::block_on;
use loom::sync::Arc;
@@ -222,3 +222,56 @@ fn nonempty_after_send() {
join.join().unwrap();
});
}
#[test]
fn is_empty_during_close() {
loom::model(|| {
let (tx, rx) = mpsc::channel::<()>(1);
let th1 = thread::spawn(move || {
assert!(rx.is_empty());
});
drop(tx);
th1.join().unwrap();
});
}
fn len_during_close_helper(n: usize) {
loom::model(move || {
let (tx, rx) = mpsc::channel::<()>(n + 1);
for _ in 0..n {
tx.try_send(()).unwrap();
}
let th1 = thread::spawn(move || {
assert_eq!(rx.len(), n);
});
drop(tx);
th1.join().unwrap();
});
}
#[test]
fn len_during_close_0() {
len_during_close_helper(0);
}
#[test]
fn len_during_close_1() {
len_during_close_helper(1);
}
#[test]
fn len_during_close_block_cap() {
len_during_close_helper(BLOCK_CAP);
}
#[test]
fn len_during_close_block_cap_plus_1() {
len_during_close_helper(BLOCK_CAP + 1);
}
+2
View File
@@ -26,6 +26,7 @@ async fn c() {
}
#[test]
#[ignore]
fn current_thread() {
let rt = runtime::Builder::new_current_thread()
.enable_all()
@@ -63,6 +64,7 @@ fn current_thread() {
}
#[test]
#[ignore]
fn multi_thread() {
let rt = runtime::Builder::new_multi_thread()
.enable_all()
+122
View File
@@ -358,6 +358,30 @@ async fn send_recv_many_unbounded_capacity() {
assert_eq!(expected, buffer);
}
#[maybe_tokio_test]
async fn recv_many_with_non_empty_buffer_bounded_rx_closed_and_idle() {
let (_tx, mut rx) = mpsc::channel::<i32>(1);
let mut buffer: Vec<i32> = vec![1];
rx.close();
assert_eq!(0, rx.recv_many(&mut buffer, 1).await);
assert_eq!(vec![1], buffer);
}
#[maybe_tokio_test]
async fn recv_many_with_non_empty_buffer_unbounded_rx_closed_and_idle() {
let (_tx, mut rx) = mpsc::unbounded_channel::<i32>();
let mut buffer: Vec<i32> = vec![1];
rx.close();
assert_eq!(0, rx.recv_many(&mut buffer, 1).await);
assert_eq!(vec![1], buffer);
}
#[tokio::test]
#[cfg(feature = "full")]
async fn async_send_recv_unbounded() {
@@ -764,6 +788,91 @@ async fn drop_permit_iterator_releases_permits() {
}
}
#[test]
fn dropping_last_permit_wakes_closed_receiver() {
let (tx, mut rx) = mpsc::channel::<()>(100);
let permit = tx.try_reserve().unwrap();
rx.close();
let mut recv = tokio_test::task::spawn(rx.recv());
assert_pending!(recv.poll());
drop(permit);
assert!(recv.is_woken());
assert_ready!(recv.poll());
}
#[test]
fn dropping_last_owned_permit_wakes_closed_receiver() {
let (tx, mut rx) = mpsc::channel::<()>(100);
let permit = tx.try_reserve_owned().unwrap();
rx.close();
let mut recv = tokio_test::task::spawn(rx.recv());
assert_pending!(recv.poll());
drop(permit);
assert!(recv.is_woken());
assert_ready!(recv.poll());
}
#[test]
fn dropping_last_permit_iterator_wakes_closed_receiver() {
let (tx, mut rx) = mpsc::channel::<()>(100);
let permits = tx.try_reserve_many(1).unwrap();
rx.close();
let mut recv = tokio_test::task::spawn(rx.recv());
assert_pending!(recv.poll());
drop(permits);
assert!(recv.is_woken());
assert_ready!(recv.poll());
}
#[test]
fn sending_last_permit_wakes_closed_receiver() {
let (tx, mut rx) = mpsc::channel::<()>(100);
let permit = tx.try_reserve().unwrap();
rx.close();
let mut recv = tokio_test::task::spawn(rx.recv());
assert_pending!(recv.poll());
permit.send(());
assert!(recv.is_woken());
assert_ready!(recv.poll());
}
#[test]
fn sending_last_owned_permit_wakes_closed_receiver() {
let (tx, mut rx) = mpsc::channel::<()>(100);
let permit = tx.try_reserve_owned().unwrap();
rx.close();
let mut recv = tokio_test::task::spawn(rx.recv());
assert_pending!(recv.poll());
permit.send(());
assert!(recv.is_woken());
assert_ready!(recv.poll());
}
#[test]
fn releasing_last_owned_permit_wakes_closed_receiver() {
let (tx, mut rx) = mpsc::channel::<()>(100);
let permit = tx.try_reserve_owned().unwrap();
rx.close();
let mut recv = tokio_test::task::spawn(rx.recv());
assert_pending!(recv.poll());
let inert_sender = permit.release();
assert!(recv.is_woken());
assert_ready!(recv.poll());
drop(inert_sender);
}
#[maybe_tokio_test]
async fn dropping_rx_closes_channel() {
let (tx, rx) = mpsc::channel(100);
@@ -975,6 +1084,19 @@ fn try_recv_after_receiver_close() {
assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv());
}
#[test]
fn try_recv_after_receiver_close_with_permit() {
let (tx, mut rx) = mpsc::channel::<()>(5);
let permit = tx.try_reserve().unwrap();
assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
rx.close();
assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
drop(permit);
assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv());
}
#[test]
fn try_recv_close_while_empty_bounded() {
let (tx, mut rx) = mpsc::channel::<()>(5);
+12
View File
@@ -78,6 +78,18 @@ fn exhaust_reading() {
let _g1 = assert_ready!(t1.poll());
}
#[test]
#[should_panic(expected = "a RwLock may not be created with 0 readers")]
fn zero_max_readers() {
RwLock::with_max_readers(100, 0);
}
#[test]
#[should_panic(expected = "a RwLock may not be created with 0 readers")]
fn zero_max_readers_const() {
RwLock::const_with_max_readers(100, 0);
}
// When there is an active exclusive owner, subsequent exclusive access should not be possible
#[test]
fn write_exclusive_pending() {