Snap for 8564071 from 7a01048e40cb464a42fe8b9dd7bac0cb3dde3bff to mainline-os-statsd-release
Change-Id: Iabe192cda6957f6b4a90baa0b366f9cf991ea8a1
diff --git a/.cargo_vcs_info.json b/.cargo_vcs_info.json
index 08632f0..f15a046 100644
--- a/.cargo_vcs_info.json
+++ b/.cargo_vcs_info.json
@@ -1,5 +1,6 @@
{
"git": {
- "sha1": "d9dfc9e1ffabcb3c01addad14878f16c2795c371"
- }
-}
+ "sha1": "f9cec068fa94bced547a66289cd288dca58c2e83"
+ },
+ "path_in_vcs": "crossbeam-channel"
+}
\ No newline at end of file
diff --git a/Android.bp b/Android.bp
index 97ef73e..a5821d8 100644
--- a/Android.bp
+++ b/Android.bp
@@ -1,4 +1,4 @@
-// This file is generated by cargo2android.py --run --device --dependencies.
+// This file is generated by cargo2android.py --config cargo2android.json.
// Do not modify this file as changes will be overridden on upgrade.
package {
@@ -44,9 +44,10 @@
rust_library {
name: "libcrossbeam_channel",
- // has rustc warnings
host_supported: true,
crate_name: "crossbeam_channel",
+ cargo_env_compat: true,
+ cargo_pkg_version: "0.5.2",
srcs: ["src/lib.rs"],
edition: "2018",
features: [
@@ -59,9 +60,3 @@
"libcrossbeam_utils",
],
}
-
-// dependent_library ["feature_list"]
-// autocfg-1.0.1
-// cfg-if-1.0.0
-// crossbeam-utils-0.8.3 "lazy_static,std"
-// lazy_static-1.4.0
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0a53e8a..6bfd923 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,11 @@
+# Version 0.5.2
+
+- Fix stacked borrows violations. (#763, #764)
+
+# Version 0.5.1
+
+- Fix memory leak in unbounded channel. (#669)
+
# Version 0.5.0
- Bump the minimum supported Rust version to 1.36.
diff --git a/Cargo.lock b/Cargo.lock
index 91cb58a..feb6e75 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1,22 +1,6 @@
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
-[[package]]
-name = "arc-swap"
-version = "0.4.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d25d88fd6b8041580a654f9d0c581a047baee2b3efee13275f2fc392fc75034"
-
-[[package]]
-name = "autocfg"
-version = "1.0.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cdb031dd78e28731d87d56cc8ffef4a8f36ca26c38fe2de700543e627f8a464a"
-
-[[package]]
-name = "cfg-if"
-version = "0.1.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4785bdd1c96b2a846b2bd7cc02e86b6b3dbf14e7e53446c4f54c92a361040822"
+version = 3
[[package]]
name = "cfg-if"
@@ -25,16 +9,10 @@
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
[[package]]
-name = "const_fn"
-version = "0.4.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce90df4c658c62f12d78f7508cf92f9173e5184a539c10bfe54a3107b3ffd0f2"
-
-[[package]]
name = "crossbeam-channel"
-version = "0.5.0"
+version = "0.5.2"
dependencies = [
- "cfg-if 1.0.0",
+ "cfg-if",
"crossbeam-utils",
"num_cpus",
"rand",
@@ -43,32 +21,30 @@
[[package]]
name = "crossbeam-utils"
-version = "0.8.0"
+version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ec91540d98355f690a86367e566ecad2e9e579f230230eb7c21398372be73ea5"
+checksum = "cfcae03edb34f947e64acdb1c33ec169824e20657e9ecb61cef6c8c74dcb8120"
dependencies = [
- "autocfg",
- "cfg-if 1.0.0",
- "const_fn",
+ "cfg-if",
"lazy_static",
]
[[package]]
name = "getrandom"
-version = "0.1.15"
+version = "0.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fc587bc0ec293155d5bfa6b9891ec18a1e330c234f896ea47fbada4cadbe47e6"
+checksum = "7fcd999463524c52659517fe2cea98493cfe485d10565e7b0fb07dbba7ad2753"
dependencies = [
- "cfg-if 0.1.10",
+ "cfg-if",
"libc",
"wasi",
]
[[package]]
name = "hermit-abi"
-version = "0.1.17"
+version = "0.1.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5aca5565f760fb5b220e499d72710ed156fdb74e631659e99377d9ebfbd13ae8"
+checksum = "62b467343b94ba476dcb2500d242dadbb39557df889310ac77c5d99100aaac33"
dependencies = [
"libc",
]
@@ -81,15 +57,15 @@
[[package]]
name = "libc"
-version = "0.2.79"
+version = "0.2.112"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2448f6066e80e3bfc792e9c98bf705b4b0fc6e8ef5b43e5889aff0eaa9c58743"
+checksum = "1b03d17f364a3a042d5e5d46b053bbbf82c92c9430c592dd4c064dc6ee997125"
[[package]]
name = "num_cpus"
-version = "1.13.0"
+version = "1.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "05499f3756671c15885fee9034446956fff3f243d6077b91e5767df161f766b3"
+checksum = "19e64526ebdee182341572e50e9ad03965aa510cd94427a4549448f285e957a1"
dependencies = [
"hermit-abi",
"libc",
@@ -97,17 +73,16 @@
[[package]]
name = "ppv-lite86"
-version = "0.2.9"
+version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c36fa947111f5c62a733b652544dd0016a43ce89619538a8ef92724a6f501a20"
+checksum = "eb9f9e6e233e5c4a35559a617bf40a4ec447db2e84c20b55a6f83167b7e57872"
[[package]]
name = "rand"
-version = "0.7.3"
+version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6a6b1679d49b24bbfe0c803429aa1874472f50d9b363131f0e89fc356b544d03"
+checksum = "2e7573632e6454cf6b99d7aac4ccca54be06da05aca2ef7423d22d27d4d4bcd8"
dependencies = [
- "getrandom",
"libc",
"rand_chacha",
"rand_core",
@@ -116,9 +91,9 @@
[[package]]
name = "rand_chacha"
-version = "0.2.2"
+version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f4c8ed856279c9737206bf725bf36935d8666ead7aa69b52be55af369d193402"
+checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88"
dependencies = [
"ppv-lite86",
"rand_core",
@@ -126,27 +101,27 @@
[[package]]
name = "rand_core"
-version = "0.5.1"
+version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "90bde5296fc891b0cef12a6d03ddccc162ce7b2aff54160af9338f8d40df6d19"
+checksum = "d34f1408f55294453790c48b2f1ebbb1c5b4b7563eb1f418bcfcfdbb06ebb4e7"
dependencies = [
"getrandom",
]
[[package]]
name = "rand_hc"
-version = "0.2.0"
+version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ca3129af7b92a17112d59ad498c6f81eaf463253766b90396d39ea7a39d6613c"
+checksum = "d51e9f596de227fda2ea6c84607f5558e196eeaf43c986b724ba4fb8fdf497e7"
dependencies = [
"rand_core",
]
[[package]]
name = "signal-hook"
-version = "0.1.16"
+version = "0.3.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "604508c1418b99dfe1925ca9224829bb2a8a9a04dda655cc01fcad46f4ab05ed"
+checksum = "647c97df271007dcea485bb74ffdb57f2e683f1306c854f468a0c244badabf2d"
dependencies = [
"libc",
"signal-hook-registry",
@@ -154,16 +129,15 @@
[[package]]
name = "signal-hook-registry"
-version = "1.2.1"
+version = "1.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a3e12110bc539e657a646068aaf5eb5b63af9d0c1f7b29c97113fad80e15f035"
+checksum = "e51e73328dc4ac0c7ccbda3a494dfa03df1de2f46018127f60c693f2648455b0"
dependencies = [
- "arc-swap",
"libc",
]
[[package]]
name = "wasi"
-version = "0.9.0+wasi-snapshot-preview1"
+version = "0.10.2+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cccddf32554fecc6acb585f82a32a72e28b48f8c4c1883ddfeeeaa96f7d8e519"
+checksum = "fd6fbd9a79829dd1ad0cc20627bf1ed606756a7f77edff7b66b7064f9cb327c6"
diff --git a/Cargo.toml b/Cargo.toml
index a5c9964..87a889f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -3,22 +3,19 @@
# When uploading crates to the registry Cargo will automatically
# "normalize" Cargo.toml files for maximal compatibility
# with all versions of Cargo and also rewrite `path` dependencies
-# to registry (e.g., crates.io) dependencies
+# to registry (e.g., crates.io) dependencies.
#
-# If you believe there's an error in this file please file an
-# issue against the rust-lang/cargo repository. If you're
-# editing this file be aware that the upstream Cargo.toml
-# will likely look very different (and much more reasonable)
+# If you are reading this file be aware that the original Cargo.toml
+# will likely look very different (and much more reasonable).
+# See Cargo.toml.orig for the original contents.
[package]
edition = "2018"
+rust-version = "1.36"
name = "crossbeam-channel"
-version = "0.5.0"
-authors = ["The Crossbeam Project Developers"]
+version = "0.5.2"
description = "Multi-producer multi-consumer channels for message passing"
homepage = "https://github.com/crossbeam-rs/crossbeam/tree/master/crossbeam-channel"
-documentation = "https://docs.rs/crossbeam-channel"
-readme = "README.md"
keywords = ["channel", "mpmc", "select", "golang", "message"]
categories = ["algorithms", "concurrency", "data-structures"]
license = "MIT OR Apache-2.0"
@@ -34,10 +31,10 @@
version = "1.13.0"
[dev-dependencies.rand]
-version = "0.7.3"
+version = "0.8"
[dev-dependencies.signal-hook]
-version = "0.1.15"
+version = "0.3"
[features]
default = ["std"]
diff --git a/Cargo.toml.orig b/Cargo.toml.orig
index 2931d21..640a808 100644
--- a/Cargo.toml.orig
+++ b/Cargo.toml.orig
@@ -4,14 +4,12 @@
# - Update CHANGELOG.md
# - Update README.md
# - Create "crossbeam-channel-X.Y.Z" git tag
-version = "0.5.0"
-authors = ["The Crossbeam Project Developers"]
+version = "0.5.2"
edition = "2018"
+rust-version = "1.36"
license = "MIT OR Apache-2.0"
-readme = "README.md"
repository = "https://github.com/crossbeam-rs/crossbeam"
homepage = "https://github.com/crossbeam-rs/crossbeam/tree/master/crossbeam-channel"
-documentation = "https://docs.rs/crossbeam-channel"
description = "Multi-producer multi-consumer channels for message passing"
keywords = ["channel", "mpmc", "select", "golang", "message"]
categories = ["algorithms", "concurrency", "data-structures"]
@@ -21,6 +19,8 @@
# Enable to use APIs that require `std`.
# This is enabled by default.
+#
+# NOTE: Disabling `std` feature is not supported yet.
std = ["crossbeam-utils/std"]
[dependencies]
@@ -34,5 +34,5 @@
[dev-dependencies]
num_cpus = "1.13.0"
-rand = "0.7.3"
-signal-hook = "0.1.15"
+rand = "0.8"
+signal-hook = "0.3"
diff --git a/LICENSE-THIRD-PARTY b/LICENSE-THIRD-PARTY
index d15e32b..ed4df76 100644
--- a/LICENSE-THIRD-PARTY
+++ b/LICENSE-THIRD-PARTY
@@ -1,37 +1,5 @@
===============================================================================
-Bounded MPMC queue
-http://www.1024cores.net/home/code-license
-
-Copyright (c) 2010-2011 Dmitry Vyukov.
-All rights reserved.
-
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions are met:
-
-1. Redistributions of source code must retain the above copyright notice, this
- list of conditions and the following disclaimer.
-2. Redistributions in binary form must reproduce the above copyright notice,
- this list of conditions and the following disclaimer in the documentation
- and/or other materials provided with the distribution.
-
-THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR IMPLIED
-WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
-MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
-EVENT SHALL DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
-INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
-LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
-PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
-LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
-OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
-ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-
-The views and conclusions contained in the software and documentation are those
-of the authors and should not be interpreted as representing official policies,
-either expressed or implied, of Dmitry Vyukov.
-
-===============================================================================
-
matching.go
https://creativecommons.org/licenses/by/3.0/legalcode
diff --git a/METADATA b/METADATA
index 00d9dde..6cbb05f 100644
--- a/METADATA
+++ b/METADATA
@@ -7,13 +7,13 @@
}
url {
type: ARCHIVE
- value: "https://static.crates.io/crates/crossbeam-channel/crossbeam-channel-0.5.0.crate"
+ value: "https://static.crates.io/crates/crossbeam-channel/crossbeam-channel-0.5.2.crate"
}
- version: "0.5.0"
+ version: "0.5.2"
license_type: NOTICE
last_upgrade_date {
- year: 2020
- month: 12
- day: 21
+ year: 2022
+ month: 3
+ day: 1
}
}
diff --git a/README.md b/README.md
index eab623a..f5077c5 100644
--- a/README.md
+++ b/README.md
@@ -2,7 +2,7 @@
[](
https://github.com/crossbeam-rs/crossbeam/actions)
-[](
+[](
https://github.com/crossbeam-rs/crossbeam/tree/master/crossbeam-channel#license)
[](
https://crates.io/crates/crossbeam-channel)
@@ -10,7 +10,7 @@
https://docs.rs/crossbeam-channel)
[](
https://www.rust-lang.org)
-[](https://discord.gg/BBYwKq)
+[](https://discord.com/invite/JXYwgWZ)
This crate provides multi-producer multi-consumer channels for message passing.
It is an alternative to [`std::sync::mpsc`] with more features and better performance.
@@ -41,7 +41,7 @@
```toml
[dependencies]
-crossbeam-channel = "0.4"
+crossbeam-channel = "0.5"
```
## Compatibility
@@ -73,10 +73,6 @@
[matching.go](http://www.nada.kth.se/~snilsson/concurrency/src/matching.go) by Stefan Nilsson,
licensed under Creative Commons Attribution 3.0 Unported License.
-* [src/flavors/array.rs](src/flavors/array.rs) is based on
- [Bounded MPMC queue](http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue)
- by Dmitry Vyukov, licensed under the Simplified BSD License and the Apache License, Version 2.0.
-
* [tests/mpsc.rs](tests/mpsc.rs) includes modifications of code from The Rust Programming Language,
licensed under the MIT License and the Apache License, Version 2.0.
diff --git a/TEST_MAPPING b/TEST_MAPPING
new file mode 100644
index 0000000..3cbd48d
--- /dev/null
+++ b/TEST_MAPPING
@@ -0,0 +1,17 @@
+// Generated by update_crate_tests.py for tests that depend on this crate.
+{
+ "imports": [
+ {
+ "path": "external/rust/crates/base64"
+ },
+ {
+ "path": "external/rust/crates/tinytemplate"
+ },
+ {
+ "path": "external/rust/crates/tinyvec"
+ },
+ {
+ "path": "external/rust/crates/unicode-xid"
+ }
+ ]
+}
diff --git a/benches/crossbeam.rs b/benches/crossbeam.rs
index 9870c98..1c05222 100644
--- a/benches/crossbeam.rs
+++ b/benches/crossbeam.rs
@@ -13,7 +13,7 @@
#[bench]
fn create(b: &mut Bencher) {
- b.iter(|| unbounded::<i32>());
+ b.iter(unbounded::<i32>);
}
#[bench]
diff --git a/cargo2android.json b/cargo2android.json
new file mode 100644
index 0000000..bf78496
--- /dev/null
+++ b/cargo2android.json
@@ -0,0 +1,4 @@
+{
+ "device": true,
+ "run": true
+}
\ No newline at end of file
diff --git a/examples/fibonacci.rs b/examples/fibonacci.rs
index cf22b7a..e6f5e89 100644
--- a/examples/fibonacci.rs
+++ b/examples/fibonacci.rs
@@ -10,7 +10,7 @@
while sender.send(x).is_ok() {
let tmp = x;
x = y;
- y = tmp + y;
+ y += tmp;
}
}
diff --git a/examples/stopwatch.rs b/examples/stopwatch.rs
index 6a67c9e..3a7578e 100644
--- a/examples/stopwatch.rs
+++ b/examples/stopwatch.rs
@@ -12,13 +12,13 @@
use std::time::{Duration, Instant};
use crossbeam_channel::{bounded, select, tick, Receiver};
+ use signal_hook::consts::SIGINT;
use signal_hook::iterator::Signals;
- use signal_hook::SIGINT;
// Creates a channel that gets a message every time `SIGINT` is signalled.
fn sigint_notifier() -> io::Result<Receiver<()>> {
let (s, r) = bounded(100);
- let signals = Signals::new(&[SIGINT])?;
+ let mut signals = Signals::new(&[SIGINT])?;
thread::spawn(move || {
for _ in signals.forever() {
@@ -33,11 +33,7 @@
// Prints the elapsed time.
fn show(dur: Duration) {
- println!(
- "Elapsed: {}.{:03} sec",
- dur.as_secs(),
- dur.subsec_nanos() / 1_000_000
- );
+ println!("Elapsed: {}.{:03} sec", dur.as_secs(), dur.subsec_millis());
}
let start = Instant::now();
diff --git a/src/channel.rs b/src/channel.rs
index ebcd652..8988235 100644
--- a/src/channel.rs
+++ b/src/channel.rs
@@ -257,8 +257,6 @@
/// recv(timeout) -> _ => println!("timed out"),
/// }
/// ```
-///
-/// [`select!`]: macro.select.html
pub fn never<T>() -> Receiver<T> {
Receiver {
flavor: ReceiverFlavor::Never(flavors::never::Channel::new()),
@@ -645,7 +643,7 @@
unsafe {
match &self.flavor {
SenderFlavor::Array(chan) => chan.release(|c| c.disconnect()),
- SenderFlavor::List(chan) => chan.release(|c| c.disconnect()),
+ SenderFlavor::List(chan) => chan.release(|c| c.disconnect_senders()),
SenderFlavor::Zero(chan) => chan.release(|c| c.disconnect()),
}
}
@@ -1137,7 +1135,7 @@
unsafe {
match &self.flavor {
ReceiverFlavor::Array(chan) => chan.release(|c| c.disconnect()),
- ReceiverFlavor::List(chan) => chan.release(|c| c.disconnect()),
+ ReceiverFlavor::List(chan) => chan.release(|c| c.disconnect_receivers()),
ReceiverFlavor::Zero(chan) => chan.release(|c| c.disconnect()),
ReceiverFlavor::At(_) => {}
ReceiverFlavor::Tick(_) => {}
@@ -1485,7 +1483,7 @@
}
/// Writes a message into the channel.
-pub unsafe fn write<T>(s: &Sender<T>, token: &mut Token, msg: T) -> Result<(), T> {
+pub(crate) unsafe fn write<T>(s: &Sender<T>, token: &mut Token, msg: T) -> Result<(), T> {
match &s.flavor {
SenderFlavor::Array(chan) => chan.write(token, msg),
SenderFlavor::List(chan) => chan.write(token, msg),
@@ -1494,7 +1492,7 @@
}
/// Reads a message from the channel.
-pub unsafe fn read<T>(r: &Receiver<T>, token: &mut Token) -> Result<T, ()> {
+pub(crate) unsafe fn read<T>(r: &Receiver<T>, token: &mut Token) -> Result<T, ()> {
match &r.flavor {
ReceiverFlavor::Array(chan) => chan.read(token),
ReceiverFlavor::List(chan) => chan.read(token),
diff --git a/src/context.rs b/src/context.rs
index e2e8480..7467b80 100644
--- a/src/context.rs
+++ b/src/context.rs
@@ -1,7 +1,8 @@
//! Thread-local context used in select.
use std::cell::Cell;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::ptr;
+use std::sync::atomic::{AtomicPtr, AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread::{self, Thread, ThreadId};
use std::time::Instant;
@@ -11,6 +12,7 @@
use crate::select::Selected;
/// Thread-local context used in select.
+// This is a private API that is used by the select macro.
#[derive(Debug, Clone)]
pub struct Context {
inner: Arc<Inner>,
@@ -23,7 +25,7 @@
select: AtomicUsize,
/// A slot into which another thread may store a pointer to its `Packet`.
- packet: AtomicUsize,
+ packet: AtomicPtr<()>,
/// Thread handle.
thread: Thread,
@@ -45,7 +47,7 @@
}
let mut f = Some(f);
- let mut f = move |cx: &Context| -> R {
+ let mut f = |cx: &Context| -> R {
let f = f.take().unwrap();
f(cx)
};
@@ -69,7 +71,7 @@
Context {
inner: Arc::new(Inner {
select: AtomicUsize::new(Selected::Waiting.into()),
- packet: AtomicUsize::new(0),
+ packet: AtomicPtr::new(ptr::null_mut()),
thread: thread::current(),
thread_id: thread::current().id(),
}),
@@ -82,7 +84,7 @@
self.inner
.select
.store(Selected::Waiting.into(), Ordering::Release);
- self.inner.packet.store(0, Ordering::Release);
+ self.inner.packet.store(ptr::null_mut(), Ordering::Release);
}
/// Attempts to select an operation.
@@ -112,19 +114,19 @@
///
/// This method must be called after `try_select` succeeds and there is a packet to provide.
#[inline]
- pub fn store_packet(&self, packet: usize) {
- if packet != 0 {
+ pub fn store_packet(&self, packet: *mut ()) {
+ if !packet.is_null() {
self.inner.packet.store(packet, Ordering::Release);
}
}
/// Waits until a packet is provided and returns it.
#[inline]
- pub fn wait_packet(&self) -> usize {
+ pub fn wait_packet(&self) -> *mut () {
let backoff = Backoff::new();
loop {
let packet = self.inner.packet.load(Ordering::Acquire);
- if packet != 0 {
+ if !packet.is_null() {
return packet;
}
backoff.snooze();
diff --git a/src/counter.rs b/src/counter.rs
index 2eaf067..2c27f7c 100644
--- a/src/counter.rs
+++ b/src/counter.rs
@@ -21,7 +21,7 @@
}
/// Wraps a channel into the reference counter.
-pub fn new<C>(chan: C) -> (Sender<C>, Receiver<C>) {
+pub(crate) fn new<C>(chan: C) -> (Sender<C>, Receiver<C>) {
let counter = Box::into_raw(Box::new(Counter {
senders: AtomicUsize::new(1),
receivers: AtomicUsize::new(1),
@@ -34,7 +34,7 @@
}
/// The sending side.
-pub struct Sender<C> {
+pub(crate) struct Sender<C> {
counter: *mut Counter<C>,
}
@@ -45,7 +45,7 @@
}
/// Acquires another sender reference.
- pub fn acquire(&self) -> Sender<C> {
+ pub(crate) fn acquire(&self) -> Sender<C> {
let count = self.counter().senders.fetch_add(1, Ordering::Relaxed);
// Cloning senders and calling `mem::forget` on the clones could potentially overflow the
@@ -63,7 +63,7 @@
/// Releases the sender reference.
///
/// Function `disconnect` will be called if this is the last sender reference.
- pub unsafe fn release<F: FnOnce(&C) -> bool>(&self, disconnect: F) {
+ pub(crate) unsafe fn release<F: FnOnce(&C) -> bool>(&self, disconnect: F) {
if self.counter().senders.fetch_sub(1, Ordering::AcqRel) == 1 {
disconnect(&self.counter().chan);
@@ -89,7 +89,7 @@
}
/// The receiving side.
-pub struct Receiver<C> {
+pub(crate) struct Receiver<C> {
counter: *mut Counter<C>,
}
@@ -100,7 +100,7 @@
}
/// Acquires another receiver reference.
- pub fn acquire(&self) -> Receiver<C> {
+ pub(crate) fn acquire(&self) -> Receiver<C> {
let count = self.counter().receivers.fetch_add(1, Ordering::Relaxed);
// Cloning receivers and calling `mem::forget` on the clones could potentially overflow the
@@ -118,7 +118,7 @@
/// Releases the receiver reference.
///
/// Function `disconnect` will be called if this is the last receiver reference.
- pub unsafe fn release<F: FnOnce(&C) -> bool>(&self, disconnect: F) {
+ pub(crate) unsafe fn release<F: FnOnce(&C) -> bool>(&self, disconnect: F) {
if self.counter().receivers.fetch_sub(1, Ordering::AcqRel) == 1 {
disconnect(&self.counter().chan);
diff --git a/src/flavors/array.rs b/src/flavors/array.rs
index 323a200..871768c 100644
--- a/src/flavors/array.rs
+++ b/src/flavors/array.rs
@@ -5,17 +5,12 @@
//! The implementation is based on Dmitry Vyukov's bounded MPMC queue.
//!
//! Source:
-//! - http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue
-//! - https://docs.google.com/document/d/1yIAYmbvL3JxOKOjuCyon7JhW4cSv1wy5hC0ApeGMV9s/pub
-//!
-//! Copyright & License:
-//! - Copyright (c) 2010-2011 Dmitry Vyukov
-//! - Simplified BSD License and Apache License, Version 2.0
-//! - http://www.1024cores.net/home/code-license
+//! - <http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue>
+//! - <https://docs.google.com/document/d/1yIAYmbvL3JxOKOjuCyon7JhW4cSv1wy5hC0ApeGMV9s/pub>
use std::cell::UnsafeCell;
use std::marker::PhantomData;
-use std::mem::{self, MaybeUninit};
+use std::mem::MaybeUninit;
use std::ptr;
use std::sync::atomic::{self, AtomicUsize, Ordering};
use std::time::Instant;
@@ -57,7 +52,7 @@
}
/// Bounded channel based on a preallocated array.
-pub struct Channel<T> {
+pub(crate) struct Channel<T> {
/// The head of the channel.
///
/// This value is a "stamp" consisting of an index into the buffer, a mark bit, and a lap, but
@@ -100,7 +95,7 @@
impl<T> Channel<T> {
/// Creates a bounded channel of capacity `cap`.
- pub fn with_capacity(cap: usize) -> Self {
+ pub(crate) fn with_capacity(cap: usize) -> Self {
assert!(cap > 0, "capacity must be positive");
// Compute constants `mark_bit` and `one_lap`.
@@ -115,7 +110,7 @@
// Allocate a buffer of `cap` slots initialized
// with stamps.
let buffer = {
- let mut boxed: Box<[Slot<T>]> = (0..cap)
+ let boxed: Box<[Slot<T>]> = (0..cap)
.map(|i| {
// Set the stamp to `{ lap: 0, mark: 0, index: i }`.
Slot {
@@ -124,9 +119,7 @@
}
})
.collect();
- let ptr = boxed.as_mut_ptr();
- mem::forget(boxed);
- ptr
+ Box::into_raw(boxed) as *mut Slot<T>
};
Channel {
@@ -143,12 +136,12 @@
}
/// Returns a receiver handle to the channel.
- pub fn receiver(&self) -> Receiver<'_, T> {
+ pub(crate) fn receiver(&self) -> Receiver<'_, T> {
Receiver(self)
}
/// Returns a sender handle to the channel.
- pub fn sender(&self) -> Sender<'_, T> {
+ pub(crate) fn sender(&self) -> Sender<'_, T> {
Sender(self)
}
@@ -224,7 +217,7 @@
}
/// Writes a message into the channel.
- pub unsafe fn write(&self, token: &mut Token, msg: T) -> Result<(), T> {
+ pub(crate) unsafe fn write(&self, token: &mut Token, msg: T) -> Result<(), T> {
// If there is no slot, the channel is disconnected.
if token.array.slot.is_null() {
return Err(msg);
@@ -314,7 +307,7 @@
}
/// Reads a message from the channel.
- pub unsafe fn read(&self, token: &mut Token) -> Result<T, ()> {
+ pub(crate) unsafe fn read(&self, token: &mut Token) -> Result<T, ()> {
if token.array.slot.is_null() {
// The channel is disconnected.
return Err(());
@@ -332,7 +325,7 @@
}
/// Attempts to send a message into the channel.
- pub fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
+ pub(crate) fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
let token = &mut Token::default();
if self.start_send(token) {
unsafe { self.write(token, msg).map_err(TrySendError::Disconnected) }
@@ -342,7 +335,11 @@
}
/// Sends a message into the channel.
- pub fn send(&self, msg: T, deadline: Option<Instant>) -> Result<(), SendTimeoutError<T>> {
+ pub(crate) fn send(
+ &self,
+ msg: T,
+ deadline: Option<Instant>,
+ ) -> Result<(), SendTimeoutError<T>> {
let token = &mut Token::default();
loop {
// Try sending a message several times.
@@ -391,7 +388,7 @@
}
/// Attempts to receive a message without blocking.
- pub fn try_recv(&self) -> Result<T, TryRecvError> {
+ pub(crate) fn try_recv(&self) -> Result<T, TryRecvError> {
let token = &mut Token::default();
if self.start_recv(token) {
@@ -402,7 +399,7 @@
}
/// Receives a message from the channel.
- pub fn recv(&self, deadline: Option<Instant>) -> Result<T, RecvTimeoutError> {
+ pub(crate) fn recv(&self, deadline: Option<Instant>) -> Result<T, RecvTimeoutError> {
let token = &mut Token::default();
loop {
// Try receiving a message several times.
@@ -453,7 +450,7 @@
}
/// Returns the current number of messages inside the channel.
- pub fn len(&self) -> usize {
+ pub(crate) fn len(&self) -> usize {
loop {
// Load the tail, then load the head.
let tail = self.tail.load(Ordering::SeqCst);
@@ -478,14 +475,15 @@
}
/// Returns the capacity of the channel.
- pub fn capacity(&self) -> Option<usize> {
+ #[allow(clippy::unnecessary_wraps)] // This is intentional.
+ pub(crate) fn capacity(&self) -> Option<usize> {
Some(self.cap)
}
/// Disconnects the channel and wakes up all blocked senders and receivers.
///
/// Returns `true` if this call disconnected the channel.
- pub fn disconnect(&self) -> bool {
+ pub(crate) fn disconnect(&self) -> bool {
let tail = self.tail.fetch_or(self.mark_bit, Ordering::SeqCst);
if tail & self.mark_bit == 0 {
@@ -498,12 +496,12 @@
}
/// Returns `true` if the channel is disconnected.
- pub fn is_disconnected(&self) -> bool {
+ pub(crate) fn is_disconnected(&self) -> bool {
self.tail.load(Ordering::SeqCst) & self.mark_bit != 0
}
/// Returns `true` if the channel is empty.
- pub fn is_empty(&self) -> bool {
+ pub(crate) fn is_empty(&self) -> bool {
let head = self.head.load(Ordering::SeqCst);
let tail = self.tail.load(Ordering::SeqCst);
@@ -515,7 +513,7 @@
}
/// Returns `true` if the channel is full.
- pub fn is_full(&self) -> bool {
+ pub(crate) fn is_full(&self) -> bool {
let tail = self.tail.load(Ordering::SeqCst);
let head = self.head.load(Ordering::SeqCst);
@@ -563,10 +561,10 @@
}
/// Receiver handle to a channel.
-pub struct Receiver<'a, T>(&'a Channel<T>);
+pub(crate) struct Receiver<'a, T>(&'a Channel<T>);
/// Sender handle to a channel.
-pub struct Sender<'a, T>(&'a Channel<T>);
+pub(crate) struct Sender<'a, T>(&'a Channel<T>);
impl<T> SelectHandle for Receiver<'_, T> {
fn try_select(&self, token: &mut Token) -> bool {
diff --git a/src/flavors/at.rs b/src/flavors/at.rs
index a2b1b57..4581edb 100644
--- a/src/flavors/at.rs
+++ b/src/flavors/at.rs
@@ -12,10 +12,10 @@
use crate::utils;
/// Result of a receive operation.
-pub type AtToken = Option<Instant>;
+pub(crate) type AtToken = Option<Instant>;
/// Channel that delivers a message at a certain moment in time
-pub struct Channel {
+pub(crate) struct Channel {
/// The instant at which the message will be delivered.
delivery_time: Instant,
@@ -26,7 +26,7 @@
impl Channel {
/// Creates a channel that delivers a message at a certain instant in time.
#[inline]
- pub fn new_deadline(when: Instant) -> Self {
+ pub(crate) fn new_deadline(when: Instant) -> Self {
Channel {
delivery_time: when,
received: AtomicBool::new(false),
@@ -34,13 +34,13 @@
}
/// Creates a channel that delivers a message after a certain duration of time.
#[inline]
- pub fn new_timeout(dur: Duration) -> Self {
+ pub(crate) fn new_timeout(dur: Duration) -> Self {
Self::new_deadline(Instant::now() + dur)
}
/// Attempts to receive a message without blocking.
#[inline]
- pub fn try_recv(&self) -> Result<Instant, TryRecvError> {
+ pub(crate) fn try_recv(&self) -> Result<Instant, TryRecvError> {
// We use relaxed ordering because this is just an optional optimistic check.
if self.received.load(Ordering::Relaxed) {
// The message has already been received.
@@ -64,7 +64,7 @@
/// Receives a message from the channel.
#[inline]
- pub fn recv(&self, deadline: Option<Instant>) -> Result<Instant, RecvTimeoutError> {
+ pub(crate) fn recv(&self, deadline: Option<Instant>) -> Result<Instant, RecvTimeoutError> {
// We use relaxed ordering because this is just an optional optimistic check.
if self.received.load(Ordering::Relaxed) {
// The message has already been received.
@@ -103,13 +103,13 @@
/// Reads a message from the channel.
#[inline]
- pub unsafe fn read(&self, token: &mut Token) -> Result<Instant, ()> {
+ pub(crate) unsafe fn read(&self, token: &mut Token) -> Result<Instant, ()> {
token.at.ok_or(())
}
/// Returns `true` if the channel is empty.
#[inline]
- pub fn is_empty(&self) -> bool {
+ pub(crate) fn is_empty(&self) -> bool {
// We use relaxed ordering because this is just an optional optimistic check.
if self.received.load(Ordering::Relaxed) {
return true;
@@ -127,13 +127,13 @@
/// Returns `true` if the channel is full.
#[inline]
- pub fn is_full(&self) -> bool {
+ pub(crate) fn is_full(&self) -> bool {
!self.is_empty()
}
/// Returns the number of messages in the channel.
#[inline]
- pub fn len(&self) -> usize {
+ pub(crate) fn len(&self) -> usize {
if self.is_empty() {
0
} else {
@@ -142,8 +142,9 @@
}
/// Returns the capacity of the channel.
+ #[allow(clippy::unnecessary_wraps)] // This is intentional.
#[inline]
- pub fn capacity(&self) -> Option<usize> {
+ pub(crate) fn capacity(&self) -> Option<usize> {
Some(1)
}
}
diff --git a/src/flavors/list.rs b/src/flavors/list.rs
index 532e8b6..5056aa4 100644
--- a/src/flavors/list.rs
+++ b/src/flavors/list.rs
@@ -151,7 +151,7 @@
///
/// Consecutive messages are grouped into blocks in order to put less pressure on the allocator and
/// improve cache efficiency.
-pub struct Channel<T> {
+pub(crate) struct Channel<T> {
/// The head of the channel.
head: CachePadded<Position<T>>,
@@ -167,7 +167,7 @@
impl<T> Channel<T> {
/// Creates a new unbounded channel.
- pub fn new() -> Self {
+ pub(crate) fn new() -> Self {
Channel {
head: CachePadded::new(Position {
block: AtomicPtr::new(ptr::null_mut()),
@@ -183,12 +183,12 @@
}
/// Returns a receiver handle to the channel.
- pub fn receiver(&self) -> Receiver<'_, T> {
+ pub(crate) fn receiver(&self) -> Receiver<'_, T> {
Receiver(self)
}
/// Returns a sender handle to the channel.
- pub fn sender(&self) -> Sender<'_, T> {
+ pub(crate) fn sender(&self) -> Sender<'_, T> {
Sender(self)
}
@@ -231,8 +231,8 @@
if self
.tail
.block
- .compare_and_swap(block, new, Ordering::Release)
- == block
+ .compare_exchange(block, new, Ordering::Release, Ordering::Relaxed)
+ .is_ok()
{
self.head.block.store(new, Ordering::Release);
block = new;
@@ -276,7 +276,7 @@
}
/// Writes a message into the channel.
- pub unsafe fn write(&self, token: &mut Token, msg: T) -> Result<(), T> {
+ pub(crate) unsafe fn write(&self, token: &mut Token, msg: T) -> Result<(), T> {
// If there is no slot, the channel is disconnected.
if token.list.block.is_null() {
return Err(msg);
@@ -380,7 +380,7 @@
}
/// Reads a message from the channel.
- pub unsafe fn read(&self, token: &mut Token) -> Result<T, ()> {
+ pub(crate) unsafe fn read(&self, token: &mut Token) -> Result<T, ()> {
if token.list.block.is_null() {
// The channel is disconnected.
return Err(());
@@ -405,7 +405,7 @@
}
/// Attempts to send a message into the channel.
- pub fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
+ pub(crate) fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
self.send(msg, None).map_err(|err| match err {
SendTimeoutError::Disconnected(msg) => TrySendError::Disconnected(msg),
SendTimeoutError::Timeout(_) => unreachable!(),
@@ -413,7 +413,11 @@
}
/// Sends a message into the channel.
- pub fn send(&self, msg: T, _deadline: Option<Instant>) -> Result<(), SendTimeoutError<T>> {
+ pub(crate) fn send(
+ &self,
+ msg: T,
+ _deadline: Option<Instant>,
+ ) -> Result<(), SendTimeoutError<T>> {
let token = &mut Token::default();
assert!(self.start_send(token));
unsafe {
@@ -423,7 +427,7 @@
}
/// Attempts to receive a message without blocking.
- pub fn try_recv(&self) -> Result<T, TryRecvError> {
+ pub(crate) fn try_recv(&self) -> Result<T, TryRecvError> {
let token = &mut Token::default();
if self.start_recv(token) {
@@ -434,7 +438,7 @@
}
/// Receives a message from the channel.
- pub fn recv(&self, deadline: Option<Instant>) -> Result<T, RecvTimeoutError> {
+ pub(crate) fn recv(&self, deadline: Option<Instant>) -> Result<T, RecvTimeoutError> {
let token = &mut Token::default();
loop {
// Try receiving a message several times.
@@ -486,7 +490,7 @@
}
/// Returns the current number of messages inside the channel.
- pub fn len(&self) -> usize {
+ pub(crate) fn len(&self) -> usize {
loop {
// Load the tail index, then load the head index.
let mut tail = self.tail.index.load(Ordering::SeqCst);
@@ -522,14 +526,14 @@
}
/// Returns the capacity of the channel.
- pub fn capacity(&self) -> Option<usize> {
+ pub(crate) fn capacity(&self) -> Option<usize> {
None
}
- /// Disconnects the channel and wakes up all blocked receivers.
+ /// Disconnects senders and wakes up all blocked receivers.
///
/// Returns `true` if this call disconnected the channel.
- pub fn disconnect(&self) -> bool {
+ pub(crate) fn disconnect_senders(&self) -> bool {
let tail = self.tail.index.fetch_or(MARK_BIT, Ordering::SeqCst);
if tail & MARK_BIT == 0 {
@@ -540,20 +544,90 @@
}
}
+ /// Disconnects receivers.
+ ///
+ /// Returns `true` if this call disconnected the channel.
+ pub(crate) fn disconnect_receivers(&self) -> bool {
+ let tail = self.tail.index.fetch_or(MARK_BIT, Ordering::SeqCst);
+
+ if tail & MARK_BIT == 0 {
+ // If receivers are dropped first, discard all messages to free
+ // memory eagerly.
+ self.discard_all_messages();
+ true
+ } else {
+ false
+ }
+ }
+
+ /// Discards all messages.
+ ///
+ /// This method should only be called when all receivers are dropped.
+ fn discard_all_messages(&self) {
+ let backoff = Backoff::new();
+ let mut tail = self.tail.index.load(Ordering::Acquire);
+ loop {
+ let offset = (tail >> SHIFT) % LAP;
+ if offset != BLOCK_CAP {
+ break;
+ }
+
+ // New updates to tail will be rejected by MARK_BIT and aborted unless it's
+ // at boundary. We need to wait for the updates take affect otherwise there
+ // can be memory leaks.
+ backoff.snooze();
+ tail = self.tail.index.load(Ordering::Acquire);
+ }
+
+ let mut head = self.head.index.load(Ordering::Acquire);
+ let mut block = self.head.block.load(Ordering::Acquire);
+
+ unsafe {
+ // Drop all messages between head and tail and deallocate the heap-allocated blocks.
+ while head >> SHIFT != tail >> SHIFT {
+ let offset = (head >> SHIFT) % LAP;
+
+ if offset < BLOCK_CAP {
+ // Drop the message in the slot.
+ let slot = (*block).slots.get_unchecked(offset);
+ slot.wait_write();
+ let p = &mut *slot.msg.get();
+ p.as_mut_ptr().drop_in_place();
+ } else {
+ (*block).wait_next();
+ // Deallocate the block and move to the next one.
+ let next = (*block).next.load(Ordering::Acquire);
+ drop(Box::from_raw(block));
+ block = next;
+ }
+
+ head = head.wrapping_add(1 << SHIFT);
+ }
+
+ // Deallocate the last remaining block.
+ if !block.is_null() {
+ drop(Box::from_raw(block));
+ }
+ }
+ head &= !MARK_BIT;
+ self.head.block.store(ptr::null_mut(), Ordering::Release);
+ self.head.index.store(head, Ordering::Release);
+ }
+
/// Returns `true` if the channel is disconnected.
- pub fn is_disconnected(&self) -> bool {
+ pub(crate) fn is_disconnected(&self) -> bool {
self.tail.index.load(Ordering::SeqCst) & MARK_BIT != 0
}
/// Returns `true` if the channel is empty.
- pub fn is_empty(&self) -> bool {
+ pub(crate) fn is_empty(&self) -> bool {
let head = self.head.index.load(Ordering::SeqCst);
let tail = self.tail.index.load(Ordering::SeqCst);
head >> SHIFT == tail >> SHIFT
}
/// Returns `true` if the channel is full.
- pub fn is_full(&self) -> bool {
+ pub(crate) fn is_full(&self) -> bool {
false
}
}
@@ -597,10 +671,10 @@
}
/// Receiver handle to a channel.
-pub struct Receiver<'a, T>(&'a Channel<T>);
+pub(crate) struct Receiver<'a, T>(&'a Channel<T>);
/// Sender handle to a channel.
-pub struct Sender<'a, T>(&'a Channel<T>);
+pub(crate) struct Sender<'a, T>(&'a Channel<T>);
impl<T> SelectHandle for Receiver<'_, T> {
fn try_select(&self, token: &mut Token) -> bool {
diff --git a/src/flavors/mod.rs b/src/flavors/mod.rs
index 299e78f..0314bf4 100644
--- a/src/flavors/mod.rs
+++ b/src/flavors/mod.rs
@@ -9,9 +9,9 @@
//! 5. `tick` - Channel that delivers messages periodically.
//! 6. `zero` - Zero-capacity channel.
-pub mod array;
-pub mod at;
-pub mod list;
-pub mod never;
-pub mod tick;
-pub mod zero;
+pub(crate) mod array;
+pub(crate) mod at;
+pub(crate) mod list;
+pub(crate) mod never;
+pub(crate) mod tick;
+pub(crate) mod zero;
diff --git a/src/flavors/never.rs b/src/flavors/never.rs
index e49d214..1951e96 100644
--- a/src/flavors/never.rs
+++ b/src/flavors/never.rs
@@ -11,17 +11,17 @@
use crate::utils;
/// This flavor doesn't need a token.
-pub type NeverToken = ();
+pub(crate) type NeverToken = ();
/// Channel that never delivers messages.
-pub struct Channel<T> {
+pub(crate) struct Channel<T> {
_marker: PhantomData<T>,
}
impl<T> Channel<T> {
/// Creates a channel that never delivers messages.
#[inline]
- pub fn new() -> Self {
+ pub(crate) fn new() -> Self {
Channel {
_marker: PhantomData,
}
@@ -29,44 +29,45 @@
/// Attempts to receive a message without blocking.
#[inline]
- pub fn try_recv(&self) -> Result<T, TryRecvError> {
+ pub(crate) fn try_recv(&self) -> Result<T, TryRecvError> {
Err(TryRecvError::Empty)
}
/// Receives a message from the channel.
#[inline]
- pub fn recv(&self, deadline: Option<Instant>) -> Result<T, RecvTimeoutError> {
+ pub(crate) fn recv(&self, deadline: Option<Instant>) -> Result<T, RecvTimeoutError> {
utils::sleep_until(deadline);
Err(RecvTimeoutError::Timeout)
}
/// Reads a message from the channel.
#[inline]
- pub unsafe fn read(&self, _token: &mut Token) -> Result<T, ()> {
+ pub(crate) unsafe fn read(&self, _token: &mut Token) -> Result<T, ()> {
Err(())
}
/// Returns `true` if the channel is empty.
#[inline]
- pub fn is_empty(&self) -> bool {
+ pub(crate) fn is_empty(&self) -> bool {
true
}
/// Returns `true` if the channel is full.
#[inline]
- pub fn is_full(&self) -> bool {
+ pub(crate) fn is_full(&self) -> bool {
true
}
/// Returns the number of messages in the channel.
#[inline]
- pub fn len(&self) -> usize {
+ pub(crate) fn len(&self) -> usize {
0
}
/// Returns the capacity of the channel.
+ #[allow(clippy::unnecessary_wraps)] // This is intentional.
#[inline]
- pub fn capacity(&self) -> Option<usize> {
+ pub(crate) fn capacity(&self) -> Option<usize> {
Some(0)
}
}
diff --git a/src/flavors/tick.rs b/src/flavors/tick.rs
index e8e7020..d4b1f6c 100644
--- a/src/flavors/tick.rs
+++ b/src/flavors/tick.rs
@@ -12,10 +12,10 @@
use crate::select::{Operation, SelectHandle, Token};
/// Result of a receive operation.
-pub type TickToken = Option<Instant>;
+pub(crate) type TickToken = Option<Instant>;
/// Channel that delivers messages periodically.
-pub struct Channel {
+pub(crate) struct Channel {
/// The instant at which the next message will be delivered.
delivery_time: AtomicCell<Instant>,
@@ -26,7 +26,7 @@
impl Channel {
/// Creates a channel that delivers messages periodically.
#[inline]
- pub fn new(dur: Duration) -> Self {
+ pub(crate) fn new(dur: Duration) -> Self {
Channel {
delivery_time: AtomicCell::new(Instant::now() + dur),
duration: dur,
@@ -35,7 +35,7 @@
/// Attempts to receive a message without blocking.
#[inline]
- pub fn try_recv(&self) -> Result<Instant, TryRecvError> {
+ pub(crate) fn try_recv(&self) -> Result<Instant, TryRecvError> {
loop {
let now = Instant::now();
let delivery_time = self.delivery_time.load();
@@ -56,7 +56,7 @@
/// Receives a message from the channel.
#[inline]
- pub fn recv(&self, deadline: Option<Instant>) -> Result<Instant, RecvTimeoutError> {
+ pub(crate) fn recv(&self, deadline: Option<Instant>) -> Result<Instant, RecvTimeoutError> {
loop {
let delivery_time = self.delivery_time.load();
let now = Instant::now();
@@ -85,25 +85,25 @@
/// Reads a message from the channel.
#[inline]
- pub unsafe fn read(&self, token: &mut Token) -> Result<Instant, ()> {
+ pub(crate) unsafe fn read(&self, token: &mut Token) -> Result<Instant, ()> {
token.tick.ok_or(())
}
/// Returns `true` if the channel is empty.
#[inline]
- pub fn is_empty(&self) -> bool {
+ pub(crate) fn is_empty(&self) -> bool {
Instant::now() < self.delivery_time.load()
}
/// Returns `true` if the channel is full.
#[inline]
- pub fn is_full(&self) -> bool {
+ pub(crate) fn is_full(&self) -> bool {
!self.is_empty()
}
/// Returns the number of messages in the channel.
#[inline]
- pub fn len(&self) -> usize {
+ pub(crate) fn len(&self) -> usize {
if self.is_empty() {
0
} else {
@@ -112,8 +112,9 @@
}
/// Returns the capacity of the channel.
+ #[allow(clippy::unnecessary_wraps)] // This is intentional.
#[inline]
- pub fn capacity(&self) -> Option<usize> {
+ pub(crate) fn capacity(&self) -> Option<usize> {
Some(1)
}
}
diff --git a/src/flavors/zero.rs b/src/flavors/zero.rs
index be647b5..4afbd8f 100644
--- a/src/flavors/zero.rs
+++ b/src/flavors/zero.rs
@@ -6,6 +6,7 @@
use std::marker::PhantomData;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Instant;
+use std::{fmt, ptr};
use crossbeam_utils::Backoff;
@@ -16,7 +17,19 @@
use crate::waker::Waker;
/// A pointer to a packet.
-pub type ZeroToken = usize;
+pub struct ZeroToken(*mut ());
+
+impl Default for ZeroToken {
+ fn default() -> Self {
+ Self(ptr::null_mut())
+ }
+}
+
+impl fmt::Debug for ZeroToken {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ fmt::Debug::fmt(&(self.0 as usize), f)
+ }
+}
/// A slot for passing one message from a sender to a receiver.
struct Packet<T> {
@@ -80,7 +93,7 @@
}
/// Zero-capacity channel.
-pub struct Channel<T> {
+pub(crate) struct Channel<T> {
/// Inner representation of the channel.
inner: Spinlock<Inner>,
@@ -90,7 +103,7 @@
impl<T> Channel<T> {
/// Constructs a new zero-capacity channel.
- pub fn new() -> Self {
+ pub(crate) fn new() -> Self {
Channel {
inner: Spinlock::new(Inner {
senders: Waker::new(),
@@ -102,12 +115,12 @@
}
/// Returns a receiver handle to the channel.
- pub fn receiver(&self) -> Receiver<'_, T> {
+ pub(crate) fn receiver(&self) -> Receiver<'_, T> {
Receiver(self)
}
/// Returns a sender handle to the channel.
- pub fn sender(&self) -> Sender<'_, T> {
+ pub(crate) fn sender(&self) -> Sender<'_, T> {
Sender(self)
}
@@ -117,10 +130,10 @@
// If there's a waiting receiver, pair up with it.
if let Some(operation) = inner.receivers.try_select() {
- token.zero = operation.packet;
+ token.zero.0 = operation.packet;
true
} else if inner.is_disconnected {
- token.zero = 0;
+ token.zero.0 = ptr::null_mut();
true
} else {
false
@@ -128,13 +141,13 @@
}
/// Writes a message into the packet.
- pub unsafe fn write(&self, token: &mut Token, msg: T) -> Result<(), T> {
+ pub(crate) unsafe fn write(&self, token: &mut Token, msg: T) -> Result<(), T> {
// If there is no packet, the channel is disconnected.
- if token.zero == 0 {
+ if token.zero.0.is_null() {
return Err(msg);
}
- let packet = &*(token.zero as *const Packet<T>);
+ let packet = &*(token.zero.0 as *const Packet<T>);
packet.msg.get().write(Some(msg));
packet.ready.store(true, Ordering::Release);
Ok(())
@@ -146,10 +159,10 @@
// If there's a waiting sender, pair up with it.
if let Some(operation) = inner.senders.try_select() {
- token.zero = operation.packet;
+ token.zero.0 = operation.packet;
true
} else if inner.is_disconnected {
- token.zero = 0;
+ token.zero.0 = ptr::null_mut();
true
} else {
false
@@ -157,13 +170,13 @@
}
/// Reads a message from the packet.
- pub unsafe fn read(&self, token: &mut Token) -> Result<T, ()> {
+ pub(crate) unsafe fn read(&self, token: &mut Token) -> Result<T, ()> {
// If there is no packet, the channel is disconnected.
- if token.zero == 0 {
+ if token.zero.0.is_null() {
return Err(());
}
- let packet = &*(token.zero as *const Packet<T>);
+ let packet = &*(token.zero.0 as *const Packet<T>);
if packet.on_stack {
// The message has been in the packet from the beginning, so there is no need to wait
@@ -177,19 +190,19 @@
// heap-allocated packet.
packet.wait_ready();
let msg = packet.msg.get().replace(None).unwrap();
- drop(Box::from_raw(packet as *const Packet<T> as *mut Packet<T>));
+ drop(Box::from_raw(token.zero.0 as *mut Packet<T>));
Ok(msg)
}
}
/// Attempts to send a message into the channel.
- pub fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
+ pub(crate) fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> {
let token = &mut Token::default();
let mut inner = self.inner.lock();
// If there's a waiting receiver, pair up with it.
if let Some(operation) = inner.receivers.try_select() {
- token.zero = operation.packet;
+ token.zero.0 = operation.packet;
drop(inner);
unsafe {
self.write(token, msg).ok().unwrap();
@@ -203,13 +216,17 @@
}
/// Sends a message into the channel.
- pub fn send(&self, msg: T, deadline: Option<Instant>) -> Result<(), SendTimeoutError<T>> {
+ pub(crate) fn send(
+ &self,
+ msg: T,
+ deadline: Option<Instant>,
+ ) -> Result<(), SendTimeoutError<T>> {
let token = &mut Token::default();
let mut inner = self.inner.lock();
// If there's a waiting receiver, pair up with it.
if let Some(operation) = inner.receivers.try_select() {
- token.zero = operation.packet;
+ token.zero.0 = operation.packet;
drop(inner);
unsafe {
self.write(token, msg).ok().unwrap();
@@ -224,10 +241,10 @@
Context::with(|cx| {
// Prepare for blocking until a receiver wakes us up.
let oper = Operation::hook(token);
- let packet = Packet::<T>::message_on_stack(msg);
+ let mut packet = Packet::<T>::message_on_stack(msg);
inner
.senders
- .register_with_packet(oper, &packet as *const Packet<T> as usize, cx);
+ .register_with_packet(oper, &mut packet as *mut Packet<T> as *mut (), cx);
inner.receivers.notify();
drop(inner);
@@ -256,13 +273,13 @@
}
/// Attempts to receive a message without blocking.
- pub fn try_recv(&self) -> Result<T, TryRecvError> {
+ pub(crate) fn try_recv(&self) -> Result<T, TryRecvError> {
let token = &mut Token::default();
let mut inner = self.inner.lock();
// If there's a waiting sender, pair up with it.
if let Some(operation) = inner.senders.try_select() {
- token.zero = operation.packet;
+ token.zero.0 = operation.packet;
drop(inner);
unsafe { self.read(token).map_err(|_| TryRecvError::Disconnected) }
} else if inner.is_disconnected {
@@ -273,13 +290,13 @@
}
/// Receives a message from the channel.
- pub fn recv(&self, deadline: Option<Instant>) -> Result<T, RecvTimeoutError> {
+ pub(crate) fn recv(&self, deadline: Option<Instant>) -> Result<T, RecvTimeoutError> {
let token = &mut Token::default();
let mut inner = self.inner.lock();
// If there's a waiting sender, pair up with it.
if let Some(operation) = inner.senders.try_select() {
- token.zero = operation.packet;
+ token.zero.0 = operation.packet;
drop(inner);
unsafe {
return self.read(token).map_err(|_| RecvTimeoutError::Disconnected);
@@ -293,10 +310,12 @@
Context::with(|cx| {
// Prepare for blocking until a sender wakes us up.
let oper = Operation::hook(token);
- let packet = Packet::<T>::empty_on_stack();
- inner
- .receivers
- .register_with_packet(oper, &packet as *const Packet<T> as usize, cx);
+ let mut packet = Packet::<T>::empty_on_stack();
+ inner.receivers.register_with_packet(
+ oper,
+ &mut packet as *mut Packet<T> as *mut (),
+ cx,
+ );
inner.senders.notify();
drop(inner);
@@ -325,7 +344,7 @@
/// Disconnects the channel and wakes up all blocked senders and receivers.
///
/// Returns `true` if this call disconnected the channel.
- pub fn disconnect(&self) -> bool {
+ pub(crate) fn disconnect(&self) -> bool {
let mut inner = self.inner.lock();
if !inner.is_disconnected {
@@ -339,31 +358,32 @@
}
/// Returns the current number of messages inside the channel.
- pub fn len(&self) -> usize {
+ pub(crate) fn len(&self) -> usize {
0
}
/// Returns the capacity of the channel.
- pub fn capacity(&self) -> Option<usize> {
+ #[allow(clippy::unnecessary_wraps)] // This is intentional.
+ pub(crate) fn capacity(&self) -> Option<usize> {
Some(0)
}
/// Returns `true` if the channel is empty.
- pub fn is_empty(&self) -> bool {
+ pub(crate) fn is_empty(&self) -> bool {
true
}
/// Returns `true` if the channel is full.
- pub fn is_full(&self) -> bool {
+ pub(crate) fn is_full(&self) -> bool {
true
}
}
/// Receiver handle to a channel.
-pub struct Receiver<'a, T>(&'a Channel<T>);
+pub(crate) struct Receiver<'a, T>(&'a Channel<T>);
/// Sender handle to a channel.
-pub struct Sender<'a, T>(&'a Channel<T>);
+pub(crate) struct Sender<'a, T>(&'a Channel<T>);
impl<T> SelectHandle for Receiver<'_, T> {
fn try_select(&self, token: &mut Token) -> bool {
@@ -380,7 +400,7 @@
let mut inner = self.0.inner.lock();
inner
.receivers
- .register_with_packet(oper, packet as usize, cx);
+ .register_with_packet(oper, packet as *mut (), cx);
inner.senders.notify();
inner.senders.can_select() || inner.is_disconnected
}
@@ -394,7 +414,7 @@
}
fn accept(&self, token: &mut Token, cx: &Context) -> bool {
- token.zero = cx.wait_packet();
+ token.zero.0 = cx.wait_packet();
true
}
@@ -430,7 +450,7 @@
let mut inner = self.0.inner.lock();
inner
.senders
- .register_with_packet(oper, packet as usize, cx);
+ .register_with_packet(oper, packet as *mut (), cx);
inner.receivers.notify();
inner.receivers.can_select() || inner.is_disconnected
}
@@ -444,7 +464,7 @@
}
fn accept(&self, token: &mut Token, cx: &Context) -> bool {
- token.zero = cx.wait_packet();
+ token.zero.0 = cx.wait_packet();
true
}
diff --git a/src/lib.rs b/src/lib.rs
index e08ac08..cc1ef11 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -294,7 +294,7 @@
//!
//! * [`after`] creates a channel that delivers a single message after a certain duration of time.
//! * [`tick`] creates a channel that delivers messages periodically.
-//! * [`never`] creates a channel that never delivers messages.
+//! * [`never`](never()) creates a channel that never delivers messages.
//!
//! These channels are very efficient because messages get lazily generated on receive operations.
//!
@@ -328,10 +328,13 @@
allow(dead_code, unused_assignments, unused_variables)
)
))]
-#![warn(missing_docs, missing_debug_implementations, rust_2018_idioms)]
+#![warn(
+ missing_docs,
+ missing_debug_implementations,
+ rust_2018_idioms,
+ unreachable_pub
+)]
#![cfg_attr(not(feature = "std"), no_std)]
-// matches! requires Rust 1.42
-#![allow(clippy::match_like_matches_macro)]
use cfg_if::cfg_if;
diff --git a/src/select.rs b/src/select.rs
index 1488f80..6103ef4 100644
--- a/src/select.rs
+++ b/src/select.rs
@@ -19,6 +19,7 @@
/// `read` or `write`.
///
/// Each field contains data associated with a specific channel flavor.
+// This is a private API that is used by the select macro.
#[derive(Debug, Default)]
pub struct Token {
pub at: flavors::at::AtToken,
@@ -93,6 +94,7 @@
///
/// This is a handle that assists select in executing an operation, registration, deciding on the
/// appropriate deadline for blocking, etc.
+// This is a private API (exposed inside crossbeam_channel::internal module) that is used by the select macro.
pub trait SelectHandle {
/// Attempts to select an operation and returns `true` on success.
fn try_select(&self, token: &mut Token) -> bool;
@@ -442,6 +444,7 @@
}
/// Attempts to select one of the operations without blocking.
+// This is a private API (exposed inside crossbeam_channel::internal module) that is used by the select macro.
#[inline]
pub fn try_select<'a>(
handles: &mut [(&'a dyn SelectHandle, usize, *const u8)],
@@ -458,6 +461,7 @@
}
/// Blocks until one of the operations becomes ready and selects it.
+// This is a private API (exposed inside crossbeam_channel::internal module) that is used by the select macro.
#[inline]
pub fn select<'a>(
handles: &mut [(&'a dyn SelectHandle, usize, *const u8)],
@@ -476,6 +480,7 @@
}
/// Blocks for a limited time until one of the operations becomes ready and selects it.
+// This is a private API (exposed inside crossbeam_channel::internal module) that is used by the select macro.
#[inline]
pub fn select_timeout<'a>(
handles: &mut [(&'a dyn SelectHandle, usize, *const u8)],
@@ -486,7 +491,7 @@
/// Blocks until a given deadline, or until one of the operations becomes ready and selects it.
#[inline]
-pub fn select_deadline<'a>(
+pub(crate) fn select_deadline<'a>(
handles: &mut [(&'a dyn SelectHandle, usize, *const u8)],
deadline: Instant,
) -> Result<SelectedOperation<'a>, SelectTimeoutError> {
@@ -864,9 +869,6 @@
/// The selected operation must be completed with [`SelectedOperation::send`]
/// or [`SelectedOperation::recv`].
///
- /// [`SelectedOperation::send`]: struct.SelectedOperation.html#method.send
- /// [`SelectedOperation::recv`]: struct.SelectedOperation.html#method.recv
- ///
/// # Examples
///
/// ```
diff --git a/src/utils.rs b/src/utils.rs
index 3fe171b..557b6a0 100644
--- a/src/utils.rs
+++ b/src/utils.rs
@@ -10,7 +10,7 @@
use crossbeam_utils::Backoff;
/// Randomly shuffles a slice.
-pub fn shuffle<T>(v: &mut [T]) {
+pub(crate) fn shuffle<T>(v: &mut [T]) {
let len = v.len();
if len <= 1 {
return;
@@ -46,7 +46,7 @@
}
/// Sleeps until the deadline, or forever if the deadline isn't specified.
-pub fn sleep_until(deadline: Option<Instant>) {
+pub(crate) fn sleep_until(deadline: Option<Instant>) {
loop {
match deadline {
None => thread::sleep(Duration::from_secs(1000)),
@@ -62,14 +62,14 @@
}
/// A simple spinlock.
-pub struct Spinlock<T> {
+pub(crate) struct Spinlock<T> {
flag: AtomicBool,
value: UnsafeCell<T>,
}
impl<T> Spinlock<T> {
/// Returns a new spinlock initialized with `value`.
- pub fn new(value: T) -> Spinlock<T> {
+ pub(crate) fn new(value: T) -> Spinlock<T> {
Spinlock {
flag: AtomicBool::new(false),
value: UnsafeCell::new(value),
@@ -77,7 +77,7 @@
}
/// Locks the spinlock.
- pub fn lock(&self) -> SpinlockGuard<'_, T> {
+ pub(crate) fn lock(&self) -> SpinlockGuard<'_, T> {
let backoff = Backoff::new();
while self.flag.swap(true, Ordering::Acquire) {
backoff.snooze();
@@ -87,7 +87,7 @@
}
/// A guard holding a spinlock locked.
-pub struct SpinlockGuard<'a, T> {
+pub(crate) struct SpinlockGuard<'a, T> {
parent: &'a Spinlock<T>,
}
diff --git a/src/waker.rs b/src/waker.rs
index 3d0af26..dec73a9 100644
--- a/src/waker.rs
+++ b/src/waker.rs
@@ -1,5 +1,6 @@
//! Waking mechanism for threads blocked on channel operations.
+use std::ptr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread::{self, ThreadId};
@@ -8,22 +9,22 @@
use crate::utils::Spinlock;
/// Represents a thread blocked on a specific channel operation.
-pub struct Entry {
+pub(crate) struct Entry {
/// The operation.
- pub oper: Operation,
+ pub(crate) oper: Operation,
/// Optional packet.
- pub packet: usize,
+ pub(crate) packet: *mut (),
/// Context associated with the thread owning this operation.
- pub cx: Context,
+ pub(crate) cx: Context,
}
/// A queue of threads blocked on channel operations.
///
/// This data structure is used by threads to register blocking operations and get woken up once
/// an operation becomes ready.
-pub struct Waker {
+pub(crate) struct Waker {
/// A list of select operations.
selectors: Vec<Entry>,
@@ -34,7 +35,7 @@
impl Waker {
/// Creates a new `Waker`.
#[inline]
- pub fn new() -> Self {
+ pub(crate) fn new() -> Self {
Waker {
selectors: Vec::new(),
observers: Vec::new(),
@@ -43,13 +44,13 @@
/// Registers a select operation.
#[inline]
- pub fn register(&mut self, oper: Operation, cx: &Context) {
- self.register_with_packet(oper, 0, cx);
+ pub(crate) fn register(&mut self, oper: Operation, cx: &Context) {
+ self.register_with_packet(oper, ptr::null_mut(), cx);
}
/// Registers a select operation and a packet.
#[inline]
- pub fn register_with_packet(&mut self, oper: Operation, packet: usize, cx: &Context) {
+ pub(crate) fn register_with_packet(&mut self, oper: Operation, packet: *mut (), cx: &Context) {
self.selectors.push(Entry {
oper,
packet,
@@ -59,7 +60,7 @@
/// Unregisters a select operation.
#[inline]
- pub fn unregister(&mut self, oper: Operation) -> Option<Entry> {
+ pub(crate) fn unregister(&mut self, oper: Operation) -> Option<Entry> {
if let Some((i, _)) = self
.selectors
.iter()
@@ -75,40 +76,32 @@
/// Attempts to find another thread's entry, select the operation, and wake it up.
#[inline]
- pub fn try_select(&mut self) -> Option<Entry> {
- let mut entry = None;
-
- if !self.selectors.is_empty() {
- let thread_id = current_thread_id();
-
- for i in 0..self.selectors.len() {
+ pub(crate) fn try_select(&mut self) -> Option<Entry> {
+ self.selectors
+ .iter()
+ .position(|selector| {
// Does the entry belong to a different thread?
- if self.selectors[i].cx.thread_id() != thread_id {
- // Try selecting this operation.
- let sel = Selected::Operation(self.selectors[i].oper);
- let res = self.selectors[i].cx.try_select(sel);
-
- if res.is_ok() {
+ selector.cx.thread_id() != current_thread_id()
+ && selector // Try selecting this operation.
+ .cx
+ .try_select(Selected::Operation(selector.oper))
+ .is_ok()
+ && {
// Provide the packet.
- self.selectors[i].cx.store_packet(self.selectors[i].packet);
+ selector.cx.store_packet(selector.packet);
// Wake the thread up.
- self.selectors[i].cx.unpark();
-
- // Remove the entry from the queue to keep it clean and improve
- // performance.
- entry = Some(self.selectors.remove(i));
- break;
+ selector.cx.unpark();
+ true
}
- }
- }
- }
-
- entry
+ })
+ // Remove the entry from the queue to keep it clean and improve
+ // performance.
+ .map(|pos| self.selectors.remove(pos))
}
/// Returns `true` if there is an entry which can be selected by the current thread.
#[inline]
- pub fn can_select(&self) -> bool {
+ pub(crate) fn can_select(&self) -> bool {
if self.selectors.is_empty() {
false
} else {
@@ -122,23 +115,23 @@
/// Registers an operation waiting to be ready.
#[inline]
- pub fn watch(&mut self, oper: Operation, cx: &Context) {
+ pub(crate) fn watch(&mut self, oper: Operation, cx: &Context) {
self.observers.push(Entry {
oper,
- packet: 0,
+ packet: ptr::null_mut(),
cx: cx.clone(),
});
}
/// Unregisters an operation waiting to be ready.
#[inline]
- pub fn unwatch(&mut self, oper: Operation) {
+ pub(crate) fn unwatch(&mut self, oper: Operation) {
self.observers.retain(|e| e.oper != oper);
}
/// Notifies all operations waiting to be ready.
#[inline]
- pub fn notify(&mut self) {
+ pub(crate) fn notify(&mut self) {
for entry in self.observers.drain(..) {
if entry.cx.try_select(Selected::Operation(entry.oper)).is_ok() {
entry.cx.unpark();
@@ -148,7 +141,7 @@
/// Notifies all registered operations that the channel is disconnected.
#[inline]
- pub fn disconnect(&mut self) {
+ pub(crate) fn disconnect(&mut self) {
for entry in self.selectors.iter() {
if entry.cx.try_select(Selected::Disconnected).is_ok() {
// Wake the thread up.
@@ -175,7 +168,7 @@
/// A waker that can be shared among threads without locking.
///
/// This is a simple wrapper around `Waker` that internally uses a mutex for synchronization.
-pub struct SyncWaker {
+pub(crate) struct SyncWaker {
/// The inner `Waker`.
inner: Spinlock<Waker>,
@@ -186,7 +179,7 @@
impl SyncWaker {
/// Creates a new `SyncWaker`.
#[inline]
- pub fn new() -> Self {
+ pub(crate) fn new() -> Self {
SyncWaker {
inner: Spinlock::new(Waker::new()),
is_empty: AtomicBool::new(true),
@@ -195,7 +188,7 @@
/// Registers the current thread with an operation.
#[inline]
- pub fn register(&self, oper: Operation, cx: &Context) {
+ pub(crate) fn register(&self, oper: Operation, cx: &Context) {
let mut inner = self.inner.lock();
inner.register(oper, cx);
self.is_empty.store(
@@ -206,7 +199,7 @@
/// Unregisters an operation previously registered by the current thread.
#[inline]
- pub fn unregister(&self, oper: Operation) -> Option<Entry> {
+ pub(crate) fn unregister(&self, oper: Operation) -> Option<Entry> {
let mut inner = self.inner.lock();
let entry = inner.unregister(oper);
self.is_empty.store(
@@ -218,7 +211,7 @@
/// Attempts to find one thread (not the current one), select its operation, and wake it up.
#[inline]
- pub fn notify(&self) {
+ pub(crate) fn notify(&self) {
if !self.is_empty.load(Ordering::SeqCst) {
let mut inner = self.inner.lock();
if !self.is_empty.load(Ordering::SeqCst) {
@@ -234,7 +227,7 @@
/// Registers an operation waiting to be ready.
#[inline]
- pub fn watch(&self, oper: Operation, cx: &Context) {
+ pub(crate) fn watch(&self, oper: Operation, cx: &Context) {
let mut inner = self.inner.lock();
inner.watch(oper, cx);
self.is_empty.store(
@@ -245,7 +238,7 @@
/// Unregisters an operation waiting to be ready.
#[inline]
- pub fn unwatch(&self, oper: Operation) {
+ pub(crate) fn unwatch(&self, oper: Operation) {
let mut inner = self.inner.lock();
inner.unwatch(oper);
self.is_empty.store(
@@ -256,7 +249,7 @@
/// Notifies all threads that the channel is disconnected.
#[inline]
- pub fn disconnect(&self) {
+ pub(crate) fn disconnect(&self) {
let mut inner = self.inner.lock();
inner.disconnect();
self.is_empty.store(
@@ -269,7 +262,7 @@
impl Drop for SyncWaker {
#[inline]
fn drop(&mut self) {
- debug_assert_eq!(self.is_empty.load(Ordering::SeqCst), true);
+ debug_assert!(self.is_empty.load(Ordering::SeqCst));
}
}
diff --git a/tests/after.rs b/tests/after.rs
index 20670dc..678a8c6 100644
--- a/tests/after.rs
+++ b/tests/after.rs
@@ -1,5 +1,7 @@
//! Tests for the after channel flavor.
+#![cfg(not(miri))] // TODO: many assertions failed due to Miri is slow
+
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::thread;
@@ -56,20 +58,20 @@
let r = after(ms(50));
assert_eq!(r.len(), 0);
- assert_eq!(r.is_empty(), true);
- assert_eq!(r.is_full(), false);
+ assert!(r.is_empty());
+ assert!(!r.is_full());
thread::sleep(ms(100));
assert_eq!(r.len(), 1);
- assert_eq!(r.is_empty(), false);
- assert_eq!(r.is_full(), true);
+ assert!(!r.is_empty());
+ assert!(r.is_full());
r.try_recv().unwrap();
assert_eq!(r.len(), 0);
- assert_eq!(r.is_empty(), true);
- assert_eq!(r.is_full(), false);
+ assert!(r.is_empty());
+ assert!(!r.is_full());
}
#[test]
@@ -211,7 +213,7 @@
break;
}
i => {
- oper.recv(&v[i]).unwrap();
+ oper.recv(v[i]).unwrap();
hits.fetch_add(1, Ordering::SeqCst);
}
}
diff --git a/tests/array.rs b/tests/array.rs
index a7ae323..bb2cebe 100644
--- a/tests/array.rs
+++ b/tests/array.rs
@@ -1,5 +1,7 @@
//! Tests for the array channel flavor.
+#![cfg(not(miri))] // TODO: many assertions failed due to Miri is slow
+
use std::any::Any;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
@@ -43,38 +45,38 @@
let (s, r) = bounded(2);
assert_eq!(s.len(), 0);
- assert_eq!(s.is_empty(), true);
- assert_eq!(s.is_full(), false);
+ assert!(s.is_empty());
+ assert!(!s.is_full());
assert_eq!(r.len(), 0);
- assert_eq!(r.is_empty(), true);
- assert_eq!(r.is_full(), false);
+ assert!(r.is_empty());
+ assert!(!r.is_full());
s.send(()).unwrap();
assert_eq!(s.len(), 1);
- assert_eq!(s.is_empty(), false);
- assert_eq!(s.is_full(), false);
+ assert!(!s.is_empty());
+ assert!(!s.is_full());
assert_eq!(r.len(), 1);
- assert_eq!(r.is_empty(), false);
- assert_eq!(r.is_full(), false);
+ assert!(!r.is_empty());
+ assert!(!r.is_full());
s.send(()).unwrap();
assert_eq!(s.len(), 2);
- assert_eq!(s.is_empty(), false);
- assert_eq!(s.is_full(), true);
+ assert!(!s.is_empty());
+ assert!(s.is_full());
assert_eq!(r.len(), 2);
- assert_eq!(r.is_empty(), false);
- assert_eq!(r.is_full(), true);
+ assert!(!r.is_empty());
+ assert!(r.is_full());
r.recv().unwrap();
assert_eq!(s.len(), 1);
- assert_eq!(s.is_empty(), false);
- assert_eq!(s.is_full(), false);
+ assert!(!s.is_empty());
+ assert!(!s.is_full());
assert_eq!(r.len(), 1);
- assert_eq!(r.is_empty(), false);
- assert_eq!(r.is_full(), false);
+ assert!(!r.is_empty());
+ assert!(!r.is_full());
}
#[test]
@@ -497,8 +499,8 @@
let mut rng = thread_rng();
for _ in 0..RUNS {
- let steps = rng.gen_range(0, 10_000);
- let additional = rng.gen_range(0, 50);
+ let steps = rng.gen_range(0..10_000);
+ let additional = rng.gen_range(0..50);
DROPS.store(0, Ordering::SeqCst);
let (s, r) = bounded::<DropCounter>(50);
diff --git a/tests/golang.rs b/tests/golang.rs
index 69a9315..05d67f6 100644
--- a/tests/golang.rs
+++ b/tests/golang.rs
@@ -9,6 +9,8 @@
//! - https://golang.org/LICENSE
//! - https://golang.org/PATENTS
+#![allow(clippy::mutex_atomic, clippy::redundant_clone)]
+
use std::alloc::{GlobalAlloc, Layout, System};
use std::any::Any;
use std::cell::Cell;
@@ -176,7 +178,7 @@
if !ret.is_null() {
ALLOCATED.fetch_add(layout.size(), SeqCst);
}
- return ret;
+ ret
}
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
@@ -232,6 +234,9 @@
mod doubleselect {
use super::*;
+ #[cfg(miri)]
+ const ITERATIONS: i32 = 100;
+ #[cfg(not(miri))]
const ITERATIONS: i32 = 10_000;
fn sender(n: i32, c1: Chan<i32>, c2: Chan<i32>, c3: Chan<i32>, c4: Chan<i32>) {
@@ -315,7 +320,7 @@
fn chain(ch: Chan<i32>, val: i32, inp: Chan<i32>, out: Chan<i32>) {
inp.recv();
if ch.recv() != Some(val) {
- panic!(val);
+ panic!("{}", val);
}
out.send(1);
}
@@ -691,6 +696,11 @@
mod select2 {
use super::*;
+ #[cfg(miri)]
+ const N: i32 = 1000;
+ #[cfg(not(miri))]
+ const N: i32 = 100000;
+
#[test]
fn main() {
fn sender(c: &Chan<i32>, n: i32) {
@@ -702,9 +712,7 @@
fn receiver(c: &Chan<i32>, dummy: &Chan<i32>, n: i32) {
for _ in 0..n {
select! {
- recv(c.rx()) -> _ => {
- ()
- }
+ recv(c.rx()) -> _ => {}
recv(dummy.rx()) -> _ => {
panic!("dummy");
}
@@ -717,15 +725,18 @@
ALLOCATED.store(0, SeqCst);
- go!(c, sender(&c, 100000));
- receiver(&c, &dummy, 100000);
+ go!(c, sender(&c, N));
+ receiver(&c, &dummy, N);
let alloc = ALLOCATED.load(SeqCst);
- go!(c, sender(&c, 100000));
- receiver(&c, &dummy, 100000);
+ go!(c, sender(&c, N));
+ receiver(&c, &dummy, N);
- assert!(!(ALLOCATED.load(SeqCst) > alloc && (ALLOCATED.load(SeqCst) - alloc) > 110000))
+ assert!(
+ !(ALLOCATED.load(SeqCst) > alloc
+ && (ALLOCATED.load(SeqCst) - alloc) > (N as usize + 10000))
+ )
}
}
@@ -913,6 +924,9 @@
#[test]
fn test_chan() {
+ #[cfg(miri)]
+ const N: i32 = 20;
+ #[cfg(not(miri))]
const N: i32 = 200;
for cap in 0..N {
@@ -1052,6 +1066,9 @@
#[test]
fn test_nonblock_recv_race() {
+ #[cfg(miri)]
+ const N: usize = 100;
+ #[cfg(not(miri))]
const N: usize = 1000;
for _ in 0..N {
@@ -1073,6 +1090,9 @@
#[test]
fn test_nonblock_select_race() {
+ #[cfg(miri)]
+ const N: usize = 100;
+ #[cfg(not(miri))]
const N: usize = 1000;
let done = make::<bool>(1);
@@ -1106,6 +1126,9 @@
#[test]
fn test_nonblock_select_race2() {
+ #[cfg(miri)]
+ const N: usize = 100;
+ #[cfg(not(miri))]
const N: usize = 1000;
let done = make::<bool>(1);
@@ -1142,6 +1165,11 @@
// Ensure that send/recv on the same chan in select
// does not crash nor deadlock.
+ #[cfg(miri)]
+ const N: usize = 100;
+ #[cfg(not(miri))]
+ const N: usize = 1000;
+
for &cap in &[0, 10] {
let wg = WaitGroup::new();
wg.add(2);
@@ -1151,7 +1179,7 @@
let p = p;
go!(wg, p, c, {
defer! { wg.done() }
- for i in 0..1000 {
+ for i in 0..N {
if p == 0 || i % 2 == 0 {
select! {
send(c.tx(), p) -> _ => {}
@@ -1180,6 +1208,11 @@
#[test]
fn test_select_stress() {
+ #[cfg(miri)]
+ const N: usize = 100;
+ #[cfg(not(miri))]
+ const N: usize = 10000;
+
let c = vec![
make::<i32>(0),
make::<i32>(0),
@@ -1187,8 +1220,6 @@
make::<i32>(3),
];
- const N: usize = 10000;
-
// There are 4 goroutines that send N values on each of the chans,
// + 4 goroutines that receive N values on each of the chans,
// + 1 goroutine that sends N values on each of the chans in a single select,
@@ -1286,6 +1317,9 @@
#[test]
fn test_select_fairness() {
+ #[cfg(miri)]
+ const TRIALS: usize = 100;
+ #[cfg(not(miri))]
const TRIALS: usize = 10000;
let c1 = make::<u8>(TRIALS + 1);
@@ -1369,6 +1403,9 @@
#[test]
fn test_pseudo_random_send() {
+ #[cfg(miri)]
+ const N: usize = 20;
+ #[cfg(not(miri))]
const N: usize = 100;
for cap in 0..N {
@@ -1412,6 +1449,9 @@
#[test]
fn test_multi_consumer() {
const NWORK: usize = 23;
+ #[cfg(miri)]
+ const NITER: usize = 100;
+ #[cfg(not(miri))]
const NITER: usize = 271828;
let pn = [2, 3, 7, 11, 13, 17, 19, 23, 27, 31];
@@ -1507,5 +1547,61 @@
// https://github.com/golang/go/blob/master/test/ken/chan1.go
mod chan1 {
- // TODO
+ use super::*;
+
+ // sent messages
+ #[cfg(miri)]
+ const N: usize = 100;
+ #[cfg(not(miri))]
+ const N: usize = 1000;
+ // receiving "goroutines"
+ const M: usize = 10;
+ // channel buffering
+ const W: usize = 2;
+
+ fn r(c: Chan<usize>, m: usize, h: Arc<Mutex<[usize; N]>>) {
+ loop {
+ select! {
+ recv(c.rx()) -> rr => {
+ let r = rr.unwrap();
+ let mut data = h.lock().unwrap();
+ if data[r] != 1 {
+ println!("r\nm={}\nr={}\nh={}\n", m, r, data[r]);
+ panic!("fail")
+ }
+ data[r] = 2;
+ }
+ }
+ }
+ }
+
+ fn s(c: Chan<usize>, h: Arc<Mutex<[usize; N]>>) {
+ for n in 0..N {
+ let r = n;
+ let mut data = h.lock().unwrap();
+ if data[r] != 0 {
+ println!("s");
+ panic!("fail");
+ }
+ data[r] = 1;
+ // https://github.com/crossbeam-rs/crossbeam/pull/615#discussion_r550281094
+ drop(data);
+ c.send(r);
+ }
+ }
+
+ #[test]
+ fn main() {
+ let h = Arc::new(Mutex::new([0usize; N]));
+ let c = make::<usize>(W);
+ for m in 0..M {
+ go!(c, h, {
+ r(c, m, h);
+ });
+ thread::yield_now();
+ }
+ thread::yield_now();
+ thread::yield_now();
+ s(c, h);
+ }
}
diff --git a/tests/iter.rs b/tests/iter.rs
index 38bcac2..463f3b0 100644
--- a/tests/iter.rs
+++ b/tests/iter.rs
@@ -93,7 +93,7 @@
assert_eq!(iter.next().unwrap(), 1);
assert_eq!(iter.next().unwrap(), 2);
- assert_eq!(iter.next().is_none(), true);
+ assert!(iter.next().is_none());
}
#[test]
@@ -106,5 +106,5 @@
let mut iter = (&r).into_iter();
assert_eq!(iter.next().unwrap(), 1);
assert_eq!(iter.next().unwrap(), 2);
- assert_eq!(iter.next().is_none(), true);
+ assert!(iter.next().is_none());
}
diff --git a/tests/list.rs b/tests/list.rs
index 8b84105..619e1fc 100644
--- a/tests/list.rs
+++ b/tests/list.rs
@@ -41,29 +41,29 @@
let (s, r) = unbounded();
assert_eq!(s.len(), 0);
- assert_eq!(s.is_empty(), true);
- assert_eq!(s.is_full(), false);
+ assert!(s.is_empty());
+ assert!(!s.is_full());
assert_eq!(r.len(), 0);
- assert_eq!(r.is_empty(), true);
- assert_eq!(r.is_full(), false);
+ assert!(r.is_empty());
+ assert!(!r.is_full());
s.send(()).unwrap();
assert_eq!(s.len(), 1);
- assert_eq!(s.is_empty(), false);
- assert_eq!(s.is_full(), false);
+ assert!(!s.is_empty());
+ assert!(!s.is_full());
assert_eq!(r.len(), 1);
- assert_eq!(r.is_empty(), false);
- assert_eq!(r.is_full(), false);
+ assert!(!r.is_empty());
+ assert!(!r.is_full());
r.recv().unwrap();
assert_eq!(s.len(), 0);
- assert_eq!(s.is_empty(), true);
- assert_eq!(s.is_full(), false);
+ assert!(s.is_empty());
+ assert!(!s.is_full());
assert_eq!(r.len(), 0);
- assert_eq!(r.is_empty(), true);
- assert_eq!(r.is_full(), false);
+ assert!(r.is_empty());
+ assert!(!r.is_full());
}
#[test]
@@ -239,6 +239,9 @@
#[test]
fn spsc() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 100_000;
let (s, r) = unbounded();
@@ -261,6 +264,9 @@
#[test]
fn mpmc() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 25_000;
const THREADS: usize = 4;
@@ -295,6 +301,9 @@
#[test]
fn stress_oneshot() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
for _ in 0..COUNT {
@@ -310,6 +319,9 @@
#[test]
fn stress_iter() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 100_000;
let (request_s, request_r) = unbounded();
@@ -371,8 +383,11 @@
.unwrap();
}
+#[cfg_attr(miri, ignore)] // Miri is too slow
#[test]
fn drops() {
+ const RUNS: usize = 100;
+
static DROPS: AtomicUsize = AtomicUsize::new(0);
#[derive(Debug, PartialEq)]
@@ -386,9 +401,9 @@
let mut rng = thread_rng();
- for _ in 0..100 {
- let steps = rng.gen_range(0, 10_000);
- let additional = rng.gen_range(0, 1000);
+ for _ in 0..RUNS {
+ let steps = rng.gen_range(0..10_000);
+ let additional = rng.gen_range(0..1000);
DROPS.store(0, Ordering::SeqCst);
let (s, r) = unbounded::<DropCounter>();
@@ -421,6 +436,9 @@
#[test]
fn linearizable() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 25_000;
const THREADS: usize = 4;
@@ -441,6 +459,9 @@
#[test]
fn fairness() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = unbounded::<()>();
@@ -463,6 +484,9 @@
#[test]
fn fairness_duplicates() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s, r) = unbounded();
@@ -496,6 +520,9 @@
#[test]
fn channel_through_channel() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 1000;
type T = Box<dyn Any + Send>;
diff --git a/tests/mpsc.rs b/tests/mpsc.rs
index 2a0786a..4d6e179 100644
--- a/tests/mpsc.rs
+++ b/tests/mpsc.rs
@@ -20,6 +20,12 @@
//! - https://github.com/rust-lang/rust/blob/master/COPYRIGHT
//! - https://www.rust-lang.org/en-US/legal.html
+#![allow(
+ clippy::drop_copy,
+ clippy::match_single_binding,
+ clippy::redundant_clone
+)]
+
use std::sync::mpsc::{RecvError, RecvTimeoutError, TryRecvError};
use std::sync::mpsc::{SendError, TrySendError};
use std::thread::JoinHandle;
@@ -176,7 +182,7 @@
) => ({
cc::crossbeam_channel_internal! {
$(
- recv(($rx).inner) -> res => {
+ $meth(($rx).inner) -> res => {
let $name = res.map_err(|_| ::std::sync::mpsc::RecvError);
$code
}
@@ -314,13 +320,18 @@
#[test]
fn stress() {
+ #[cfg(miri)]
+ const COUNT: usize = 500;
+ #[cfg(not(miri))]
+ const COUNT: usize = 10000;
+
let (tx, rx) = channel::<i32>();
let t = thread::spawn(move || {
- for _ in 0..10000 {
+ for _ in 0..COUNT {
tx.send(1).unwrap();
}
});
- for _ in 0..10000 {
+ for _ in 0..COUNT {
assert_eq!(rx.recv().unwrap(), 1);
}
t.join().ok().unwrap();
@@ -328,6 +339,9 @@
#[test]
fn stress_shared() {
+ #[cfg(miri)]
+ const AMT: u32 = 500;
+ #[cfg(not(miri))]
const AMT: u32 = 10000;
const NTHREADS: u32 = 8;
let (tx, rx) = channel::<i32>();
@@ -336,10 +350,7 @@
for _ in 0..AMT * NTHREADS {
assert_eq!(rx.recv().unwrap(), 1);
}
- match rx.try_recv() {
- Ok(..) => panic!(),
- _ => {}
- }
+ assert!(rx.try_recv().is_err());
});
let mut ts = Vec::with_capacity(NTHREADS as usize);
@@ -735,12 +746,17 @@
#[test]
fn recv_a_lot() {
+ #[cfg(miri)]
+ const N: usize = 100;
+ #[cfg(not(miri))]
+ const N: usize = 10000;
+
// Regression test that we don't run out of stack in scheduler context
let (tx, rx) = channel();
- for _ in 0..10000 {
+ for _ in 0..N {
tx.send(()).unwrap();
}
- for _ in 0..10000 {
+ for _ in 0..N {
rx.recv().unwrap();
}
}
@@ -880,7 +896,7 @@
};
assert_eq!(iter.next().unwrap(), 1);
assert_eq!(iter.next().unwrap(), 2);
- assert_eq!(iter.next().is_none(), true);
+ assert!(iter.next().is_none());
}
#[test]
@@ -892,7 +908,7 @@
let mut iter = (&rx).into_iter();
assert_eq!(iter.next().unwrap(), 1);
assert_eq!(iter.next().unwrap(), 2);
- assert_eq!(iter.next().is_none(), true);
+ assert!(iter.next().is_none());
}
#[test]
@@ -1079,13 +1095,18 @@
#[test]
fn stress() {
+ #[cfg(miri)]
+ const N: usize = 100;
+ #[cfg(not(miri))]
+ const N: usize = 10000;
+
let (tx, rx) = sync_channel::<i32>(0);
let t = thread::spawn(move || {
- for _ in 0..10000 {
+ for _ in 0..N {
tx.send(1).unwrap();
}
});
- for _ in 0..10000 {
+ for _ in 0..N {
assert_eq!(rx.recv().unwrap(), 1);
}
t.join().unwrap();
@@ -1093,10 +1114,15 @@
#[test]
fn stress_recv_timeout_two_threads() {
+ #[cfg(miri)]
+ const N: usize = 100;
+ #[cfg(not(miri))]
+ const N: usize = 10000;
+
let (tx, rx) = sync_channel::<i32>(0);
let t = thread::spawn(move || {
- for _ in 0..10000 {
+ for _ in 0..N {
tx.send(1).unwrap();
}
});
@@ -1113,12 +1139,15 @@
}
}
- assert_eq!(recv_count, 10000);
+ assert_eq!(recv_count, N);
t.join().unwrap();
}
#[test]
fn stress_recv_timeout_shared() {
+ #[cfg(miri)]
+ const AMT: u32 = 100;
+ #[cfg(not(miri))]
const AMT: u32 = 1000;
const NTHREADS: u32 = 8;
let (tx, rx) = sync_channel::<i32>(0);
@@ -1165,6 +1194,9 @@
#[test]
fn stress_shared() {
+ #[cfg(miri)]
+ const AMT: u32 = 100;
+ #[cfg(not(miri))]
const AMT: u32 = 1000;
const NTHREADS: u32 = 8;
let (tx, rx) = sync_channel::<i32>(0);
@@ -1174,10 +1206,7 @@
for _ in 0..AMT * NTHREADS {
assert_eq!(rx.recv().unwrap(), 1);
}
- match rx.try_recv() {
- Ok(..) => panic!(),
- _ => {}
- }
+ assert!(rx.try_recv().is_err());
dtx.send(()).unwrap();
});
@@ -1449,12 +1478,17 @@
#[test]
fn recv_a_lot() {
+ #[cfg(miri)]
+ const N: usize = 100;
+ #[cfg(not(miri))]
+ const N: usize = 10000;
+
// Regression test that we don't run out of stack in scheduler context
- let (tx, rx) = sync_channel(10000);
- for _ in 0..10000 {
+ let (tx, rx) = sync_channel(N);
+ for _ in 0..N {
tx.send(()).unwrap();
}
- for _ in 0..10000 {
+ for _ in 0..N {
rx.recv().unwrap();
}
}
@@ -1792,7 +1826,11 @@
#[test]
fn stress() {
+ #[cfg(miri)]
+ const AMT: i32 = 100;
+ #[cfg(not(miri))]
const AMT: i32 = 10000;
+
let (tx1, rx1) = channel::<i32>();
let (tx2, rx2) = channel::<i32>();
let (tx3, rx3) = channel::<()>();
diff --git a/tests/never.rs b/tests/never.rs
index 31cebf6..f275126 100644
--- a/tests/never.rs
+++ b/tests/never.rs
@@ -65,8 +65,8 @@
fn len_empty_full() {
let r = never::<i32>();
assert_eq!(r.len(), 0);
- assert_eq!(r.is_empty(), true);
- assert_eq!(r.is_full(), true);
+ assert!(r.is_empty());
+ assert!(r.is_full());
}
#[test]
diff --git a/tests/ready.rs b/tests/ready.rs
index 700f487..d8dd6ce 100644
--- a/tests/ready.rs
+++ b/tests/ready.rs
@@ -1,5 +1,7 @@
//! Tests for channel readiness using the `Select` struct.
+#![allow(clippy::drop_copy)]
+
use std::any::Any;
use std::cell::Cell;
use std::thread;
@@ -490,6 +492,9 @@
#[test]
fn stress_recv() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = unbounded();
@@ -527,6 +532,9 @@
#[test]
fn stress_send() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded(0);
@@ -561,6 +569,9 @@
#[test]
fn stress_mixed() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded(0);
@@ -606,8 +617,7 @@
thread::sleep(ms(500));
}
- let done = false;
- while !done {
+ loop {
let mut sel = Select::new();
sel.send(&s);
match sel.ready_timeout(ms(100)) {
@@ -628,15 +638,14 @@
thread::sleep(ms(500));
}
- let mut done = false;
- while !done {
+ loop {
let mut sel = Select::new();
sel.recv(&r);
match sel.ready_timeout(ms(100)) {
Err(_) => {}
Ok(0) => {
assert_eq!(r.try_recv(), Ok(i));
- done = true;
+ break;
}
Ok(_) => panic!(),
}
@@ -668,6 +677,9 @@
#[test]
fn channel_through_channel() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 1000;
type T = Box<dyn Any + Send>;
@@ -724,6 +736,9 @@
#[test]
fn fairness1() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded::<()>(COUNT);
@@ -769,6 +784,9 @@
#[test]
fn fairness2() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 100_000;
let (s1, r1) = unbounded::<()>();
diff --git a/tests/select.rs b/tests/select.rs
index 4cf08b6..f24aed8 100644
--- a/tests/select.rs
+++ b/tests/select.rs
@@ -1,5 +1,7 @@
//! Tests for channel selection using the `Select` struct.
+#![allow(clippy::drop_copy)]
+
use std::any::Any;
use std::cell::Cell;
use std::thread;
@@ -406,6 +408,7 @@
.unwrap();
}
+#[cfg_attr(miri, ignore)] // Miri is too slow
#[test]
fn loop_try() {
const RUNS: usize = 20;
@@ -690,6 +693,9 @@
#[test]
fn stress_recv() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = unbounded();
@@ -728,6 +734,9 @@
#[test]
fn stress_send() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded(0);
@@ -763,6 +772,9 @@
#[test]
fn stress_mixed() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded(0);
@@ -809,8 +821,7 @@
thread::sleep(ms(500));
}
- let done = false;
- while !done {
+ loop {
let mut sel = Select::new();
let oper1 = sel.send(&s);
let oper = sel.select_timeout(ms(100));
@@ -834,8 +845,7 @@
thread::sleep(ms(500));
}
- let mut done = false;
- while !done {
+ loop {
let mut sel = Select::new();
let oper1 = sel.recv(&r);
let oper = sel.select_timeout(ms(100));
@@ -844,7 +854,7 @@
Ok(oper) => match oper.index() {
ix if ix == oper1 => {
assert_eq!(oper.recv(&r), Ok(i));
- done = true;
+ break;
}
_ => unreachable!(),
},
@@ -897,12 +907,12 @@
for i in 0..THREADS {
scope.spawn(move |_| {
let mut sel = Select::new();
- let oper1 = sel.recv(&r);
- let oper2 = sel.send(&s);
+ let oper1 = sel.recv(r);
+ let oper2 = sel.send(s);
let oper = sel.select();
match oper.index() {
- ix if ix == oper1 => assert_ne!(oper.recv(&r), Ok(i)),
- ix if ix == oper2 => assert!(oper.send(&s, i).is_ok()),
+ ix if ix == oper1 => assert_ne!(oper.recv(r), Ok(i)),
+ ix if ix == oper2 => assert!(oper.send(s, i).is_ok()),
_ => unreachable!(),
}
});
@@ -923,12 +933,12 @@
for i in 0..THREADS {
scope.spawn(move |_| {
let mut sel = Select::new();
- let oper1 = sel.recv(&r);
- let oper2 = sel.send(&s);
+ let oper1 = sel.recv(r);
+ let oper2 = sel.send(s);
let oper = sel.select();
match oper.index() {
- ix if ix == oper1 => assert_ne!(oper.recv(&r), Ok(i)),
- ix if ix == oper2 => assert!(oper.send(&s, i).is_ok()),
+ ix if ix == oper1 => assert_ne!(oper.recv(r), Ok(i)),
+ ix if ix == oper2 => assert!(oper.send(s, i).is_ok()),
_ => unreachable!(),
}
});
@@ -942,6 +952,9 @@
#[test]
fn channel_through_channel() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 1000;
type T = Box<dyn Any + Send>;
@@ -1000,6 +1013,9 @@
#[test]
fn linearizable_try() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 100_000;
for step in 0..2 {
@@ -1052,6 +1068,9 @@
#[test]
fn linearizable_timeout() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 100_000;
for step in 0..2 {
@@ -1104,6 +1123,9 @@
#[test]
fn fairness1() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded::<()>(COUNT);
@@ -1150,6 +1172,9 @@
#[test]
fn fairness2() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = unbounded::<()>();
@@ -1214,8 +1239,8 @@
let (s, r) = &bounded::<usize>(0);
let mut sel = Select::new();
- let oper1 = sel.recv(&r);
- let oper2 = sel.send(&s);
+ let oper1 = sel.recv(r);
+ let oper2 = sel.send(s);
let sel = &sel;
scope(|scope| {
@@ -1224,8 +1249,8 @@
let mut sel = sel.clone();
let oper = sel.select();
match oper.index() {
- ix if ix == oper1 => assert_ne!(oper.recv(&r), Ok(i)),
- ix if ix == oper2 => assert!(oper.send(&s, i).is_ok()),
+ ix if ix == oper1 => assert_ne!(oper.recv(r), Ok(i)),
+ ix if ix == oper2 => assert!(oper.send(s, i).is_ok()),
_ => unreachable!(),
}
});
@@ -1243,8 +1268,8 @@
let (s, r) = &bounded::<usize>(0);
let mut sel = Select::new();
- let oper1 = sel.recv(&r);
- let oper2 = sel.send(&s);
+ let oper1 = sel.recv(r);
+ let oper2 = sel.send(s);
scope(|scope| {
for i in 0..THREADS {
@@ -1252,8 +1277,8 @@
scope.spawn(move |_| {
let oper = sel.select();
match oper.index() {
- ix if ix == oper1 => assert_ne!(oper.recv(&r), Ok(i)),
- ix if ix == oper2 => assert!(oper.send(&s, i).is_ok()),
+ ix if ix == oper1 => assert_ne!(oper.recv(r), Ok(i)),
+ ix if ix == oper2 => assert!(oper.send(s, i).is_ok()),
_ => unreachable!(),
}
});
@@ -1266,6 +1291,9 @@
#[test]
fn reuse() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded(0);
diff --git a/tests/select_macro.rs b/tests/select_macro.rs
index c05f7a0..0b9a21a 100644
--- a/tests/select_macro.rs
+++ b/tests/select_macro.rs
@@ -1,6 +1,7 @@
//! Tests for the `select!` macro.
#![forbid(unsafe_code)] // select! is safe.
+#![allow(clippy::drop_copy, clippy::match_single_binding)]
use std::any::Any;
use std::cell::Cell;
@@ -283,6 +284,7 @@
.unwrap();
}
+#[cfg_attr(miri, ignore)] // Miri is too slow
#[test]
fn loop_try() {
const RUNS: usize = 20;
@@ -485,6 +487,9 @@
#[test]
fn stress_recv() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = unbounded();
@@ -518,6 +523,9 @@
#[test]
fn stress_send() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded(0);
@@ -548,6 +556,9 @@
#[test]
fn stress_mixed() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded(0);
@@ -681,6 +692,9 @@
#[test]
fn channel_through_channel() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 1000;
type T = Box<dyn Any + Send>;
@@ -726,6 +740,9 @@
#[test]
fn linearizable_default() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 100_000;
for step in 0..2 {
@@ -770,6 +787,9 @@
#[test]
fn linearizable_timeout() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 100_000;
for step in 0..2 {
@@ -814,6 +834,9 @@
#[test]
fn fairness1() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded::<()>(COUNT);
@@ -838,6 +861,9 @@
#[test]
fn fairness2() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = unbounded::<()>();
@@ -875,6 +901,9 @@
#[test]
fn fairness_recv() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded::<()>(COUNT);
@@ -897,6 +926,9 @@
#[test]
fn fairness_send() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, _r1) = bounded::<()>(COUNT);
@@ -912,6 +944,7 @@
assert!(hits.iter().all(|x| *x >= COUNT / 4));
}
+#[allow(clippy::or_fun_call)] // This is intentional.
#[test]
fn references() {
let (s, r) = unbounded::<i32>();
@@ -958,6 +991,7 @@
drop(s);
}
+#[allow(clippy::redundant_closure_call)] // This is intentional.
#[test]
fn move_handles() {
let (s, r) = unbounded::<i32>();
diff --git a/tests/thread_locals.rs b/tests/thread_locals.rs
index 9e27146..effb6a1 100644
--- a/tests/thread_locals.rs
+++ b/tests/thread_locals.rs
@@ -1,5 +1,7 @@
//! Tests that make sure accessing thread-locals while exiting the thread doesn't cause panics.
+#![cfg(not(miri))] // error: abnormal termination: the evaluated program aborted execution
+
use std::thread;
use std::time::Duration;
diff --git a/tests/tick.rs b/tests/tick.rs
index 5dc8730..23bbb1f 100644
--- a/tests/tick.rs
+++ b/tests/tick.rs
@@ -1,5 +1,7 @@
//! Tests for the tick channel flavor.
+#![cfg(not(miri))] // TODO: many assertions failed due to Miri is slow
+
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::thread;
@@ -78,20 +80,20 @@
let r = tick(ms(50));
assert_eq!(r.len(), 0);
- assert_eq!(r.is_empty(), true);
- assert_eq!(r.is_full(), false);
+ assert!(r.is_empty());
+ assert!(!r.is_full());
thread::sleep(ms(100));
assert_eq!(r.len(), 1);
- assert_eq!(r.is_empty(), false);
- assert_eq!(r.is_full(), true);
+ assert!(!r.is_empty());
+ assert!(r.is_full());
r.try_recv().unwrap();
assert_eq!(r.len(), 0);
- assert_eq!(r.is_empty(), true);
- assert_eq!(r.is_full(), false);
+ assert!(r.is_empty());
+ assert!(!r.is_full());
}
#[test]
@@ -127,6 +129,7 @@
assert_eq!(r.try_recv(), Err(TryRecvError::Empty));
}
+#[cfg(not(crossbeam_sanitize))] // TODO: assertions failed due to tsan is slow
#[test]
fn recv_timeout() {
let start = Instant::now();
@@ -251,6 +254,7 @@
assert_eq!(hits.load(Ordering::SeqCst), 8);
}
+#[cfg(not(crossbeam_sanitize))] // TODO: assertions failed due to tsan is slow
#[test]
fn ready() {
const THREADS: usize = 4;
diff --git a/tests/zero.rs b/tests/zero.rs
index 66dcc1e..ba41b1a 100644
--- a/tests/zero.rs
+++ b/tests/zero.rs
@@ -35,11 +35,11 @@
let (s, r) = bounded(0);
assert_eq!(s.len(), 0);
- assert_eq!(s.is_empty(), true);
- assert_eq!(s.is_full(), true);
+ assert!(s.is_empty());
+ assert!(s.is_full());
assert_eq!(r.len(), 0);
- assert_eq!(r.is_empty(), true);
- assert_eq!(r.is_full(), true);
+ assert!(r.is_empty());
+ assert!(r.is_full());
scope(|scope| {
scope.spawn(|_| s.send(0).unwrap());
@@ -48,11 +48,11 @@
.unwrap();
assert_eq!(s.len(), 0);
- assert_eq!(s.is_empty(), true);
- assert_eq!(s.is_full(), true);
+ assert!(s.is_empty());
+ assert!(s.is_full());
assert_eq!(r.len(), 0);
- assert_eq!(r.is_empty(), true);
- assert_eq!(r.is_full(), true);
+ assert!(r.is_empty());
+ assert!(r.is_full());
}
#[test]
@@ -187,6 +187,9 @@
#[test]
fn len() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 25_000;
let (s, r) = bounded(0);
@@ -249,6 +252,9 @@
#[test]
fn spsc() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 100_000;
let (s, r) = bounded(0);
@@ -271,6 +277,9 @@
#[test]
fn mpmc() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 25_000;
const THREADS: usize = 4;
@@ -303,6 +312,9 @@
#[test]
fn stress_oneshot() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
for _ in 0..COUNT {
@@ -316,6 +328,7 @@
}
}
+#[cfg_attr(miri, ignore)] // Miri is too slow
#[test]
fn stress_iter() {
const COUNT: usize = 1000;
@@ -383,8 +396,11 @@
.unwrap();
}
+#[cfg_attr(miri, ignore)] // Miri is too slow
#[test]
fn drops() {
+ const RUNS: usize = 100;
+
static DROPS: AtomicUsize = AtomicUsize::new(0);
#[derive(Debug, PartialEq)]
@@ -398,8 +414,8 @@
let mut rng = thread_rng();
- for _ in 0..100 {
- let steps = rng.gen_range(0, 3_000);
+ for _ in 0..RUNS {
+ let steps = rng.gen_range(0..3_000);
DROPS.store(0, Ordering::SeqCst);
let (s, r) = bounded::<DropCounter>(0);
@@ -428,6 +444,9 @@
#[test]
fn fairness() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s1, r1) = bounded::<()>(0);
@@ -459,6 +478,9 @@
#[test]
fn fairness_duplicates() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 10_000;
let (s, r) = bounded::<()>(0);
@@ -517,6 +539,9 @@
#[test]
fn channel_through_channel() {
+ #[cfg(miri)]
+ const COUNT: usize = 100;
+ #[cfg(not(miri))]
const COUNT: usize = 1000;
type T = Box<dyn Any + Send>;
