Upgrade rust/crates/crossbeam-channel to 0.5.1
Test: make
Change-Id: If2f1efb29a30eca4b0823966bda2e4684d19b91c
diff --git a/.cargo_vcs_info.json b/.cargo_vcs_info.json
index 08632f0..28a14f3 100644
--- a/.cargo_vcs_info.json
+++ b/.cargo_vcs_info.json
@@ -1,5 +1,5 @@
{
"git": {
- "sha1": "d9dfc9e1ffabcb3c01addad14878f16c2795c371"
+ "sha1": "b13fdeef2ab90bac1482d38959beadbc6b0a08ed"
}
}
diff --git a/Android.bp b/Android.bp
index 97ef73e..710a437 100644
--- a/Android.bp
+++ b/Android.bp
@@ -44,7 +44,6 @@
rust_library {
name: "libcrossbeam_channel",
- // has rustc warnings
host_supported: true,
crate_name: "crossbeam_channel",
srcs: ["src/lib.rs"],
@@ -63,5 +62,5 @@
// dependent_library ["feature_list"]
// autocfg-1.0.1
// cfg-if-1.0.0
-// crossbeam-utils-0.8.3 "lazy_static,std"
+// crossbeam-utils-0.8.4 "lazy_static,std"
// lazy_static-1.4.0
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0a53e8a..98fd9f6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,7 @@
+# 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..1ae0d57 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1,10 +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"
+version = 3
[[package]]
name = "autocfg"
@@ -14,27 +10,15 @@
[[package]]
name = "cfg-if"
-version = "0.1.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4785bdd1c96b2a846b2bd7cc02e86b6b3dbf14e7e53446c4f54c92a361040822"
-
-[[package]]
-name = "cfg-if"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
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.1"
dependencies = [
- "cfg-if 1.0.0",
+ "cfg-if",
"crossbeam-utils",
"num_cpus",
"rand",
@@ -43,32 +27,31 @@
[[package]]
name = "crossbeam-utils"
-version = "0.8.0"
+version = "0.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ec91540d98355f690a86367e566ecad2e9e579f230230eb7c21398372be73ea5"
+checksum = "e7e9d99fa91428effe99c5c6d4634cdeba32b8cf784fc428a2a687f61a952c49"
dependencies = [
"autocfg",
- "cfg-if 1.0.0",
- "const_fn",
+ "cfg-if",
"lazy_static",
]
[[package]]
name = "getrandom"
-version = "0.1.15"
+version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fc587bc0ec293155d5bfa6b9891ec18a1e330c234f896ea47fbada4cadbe47e6"
+checksum = "c9495705279e7140bf035dde1f6e750c162df8b625267cd52cc44e0b156732c8"
dependencies = [
- "cfg-if 0.1.10",
+ "cfg-if",
"libc",
"wasi",
]
[[package]]
name = "hermit-abi"
-version = "0.1.17"
+version = "0.1.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5aca5565f760fb5b220e499d72710ed156fdb74e631659e99377d9ebfbd13ae8"
+checksum = "322f4de77956e22ed0e5032c359a0f1273f1f7f0d79bfa3b8ffbc730d7fbcc5c"
dependencies = [
"libc",
]
@@ -81,9 +64,9 @@
[[package]]
name = "libc"
-version = "0.2.79"
+version = "0.2.93"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2448f6066e80e3bfc792e9c98bf705b4b0fc6e8ef5b43e5889aff0eaa9c58743"
+checksum = "9385f66bf6105b241aa65a61cb923ef20efc665cb9f9bb50ac2f0c4b7f378d41"
[[package]]
name = "num_cpus"
@@ -97,17 +80,16 @@
[[package]]
name = "ppv-lite86"
-version = "0.2.9"
+version = "0.2.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c36fa947111f5c62a733b652544dd0016a43ce89619538a8ef92724a6f501a20"
+checksum = "ac74c624d6b2d21f425f752262f42188365d7b8ff1aff74c82e45136510a4857"
[[package]]
name = "rand"
-version = "0.7.3"
+version = "0.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6a6b1679d49b24bbfe0c803429aa1874472f50d9b363131f0e89fc356b544d03"
+checksum = "0ef9e7e66b4468674bfcb0c81af8b7fa0bb154fa9f28eb840da5c447baeb8d7e"
dependencies = [
- "getrandom",
"libc",
"rand_chacha",
"rand_core",
@@ -116,9 +98,9 @@
[[package]]
name = "rand_chacha"
-version = "0.2.2"
+version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f4c8ed856279c9737206bf725bf36935d8666ead7aa69b52be55af369d193402"
+checksum = "e12735cf05c9e10bf21534da50a147b924d555dc7a547c42e6bb2d5b6017ae0d"
dependencies = [
"ppv-lite86",
"rand_core",
@@ -126,27 +108,27 @@
[[package]]
name = "rand_core"
-version = "0.5.1"
+version = "0.6.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "90bde5296fc891b0cef12a6d03ddccc162ce7b2aff54160af9338f8d40df6d19"
+checksum = "34cf66eb183df1c5876e2dcf6b13d57340741e8dc255b48e40a26de954d06ae7"
dependencies = [
"getrandom",
]
[[package]]
name = "rand_hc"
-version = "0.2.0"
+version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ca3129af7b92a17112d59ad498c6f81eaf463253766b90396d39ea7a39d6613c"
+checksum = "3190ef7066a446f2e7f42e239d161e905420ccab01eb967c9eb27d21b2322a73"
dependencies = [
"rand_core",
]
[[package]]
name = "signal-hook"
-version = "0.1.16"
+version = "0.3.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "604508c1418b99dfe1925ca9224829bb2a8a9a04dda655cc01fcad46f4ab05ed"
+checksum = "ef33d6d0cd06e0840fba9985aab098c147e67e05cee14d412d3345ed14ff30ac"
dependencies = [
"libc",
"signal-hook-registry",
@@ -154,16 +136,15 @@
[[package]]
name = "signal-hook-registry"
-version = "1.2.1"
+version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a3e12110bc539e657a646068aaf5eb5b63af9d0c1f7b29c97113fad80e15f035"
+checksum = "16f1d0fef1604ba8f7a073c7e701f213e056707210e9020af4528e0101ce11a6"
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..abcc485 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -13,12 +13,11 @@
[package]
edition = "2018"
name = "crossbeam-channel"
-version = "0.5.0"
+version = "0.5.1"
authors = ["The Crossbeam Project Developers"]
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 +33,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..0e048a2 100644
--- a/Cargo.toml.orig
+++ b/Cargo.toml.orig
@@ -4,11 +4,10 @@
# - Update CHANGELOG.md
# - Update README.md
# - Create "crossbeam-channel-X.Y.Z" git tag
-version = "0.5.0"
+version = "0.5.1"
authors = ["The Crossbeam Project Developers"]
edition = "2018"
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"
@@ -34,5 +33,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..c1dad0e 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.1.crate"
}
- version: "0.5.0"
+ version: "0.5.1"
license_type: NOTICE
last_upgrade_date {
- year: 2020
- month: 12
- day: 21
+ year: 2021
+ month: 5
+ day: 19
}
}
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/examples/stopwatch.rs b/examples/stopwatch.rs
index 6a67c9e..98895c5 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() {
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/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..c49eef1 100644
--- a/src/flavors/array.rs
+++ b/src/flavors/array.rs
@@ -5,13 +5,8 @@
//! 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;
@@ -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`.
@@ -143,12 +138,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 +219,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 +309,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 +327,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 +337,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 +390,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 +401,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 +452,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 +477,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 +498,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 +515,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 +563,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..9790b77 100644
--- a/src/flavors/zero.rs
+++ b/src/flavors/zero.rs
@@ -16,7 +16,7 @@
use crate::waker::Waker;
/// A pointer to a packet.
-pub type ZeroToken = usize;
+pub(crate) type ZeroToken = usize;
/// A slot for passing one message from a sender to a receiver.
struct Packet<T> {
@@ -80,7 +80,7 @@
}
/// Zero-capacity channel.
-pub struct Channel<T> {
+pub(crate) struct Channel<T> {
/// Inner representation of the channel.
inner: Spinlock<Inner>,
@@ -90,7 +90,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 +102,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)
}
@@ -128,7 +128,7 @@
}
/// 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 {
return Err(msg);
@@ -157,7 +157,7 @@
}
/// 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 {
return Err(());
@@ -183,7 +183,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();
let mut inner = self.inner.lock();
@@ -203,7 +203,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();
let mut inner = self.inner.lock();
@@ -256,7 +260,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();
let mut inner = self.inner.lock();
@@ -273,7 +277,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();
let mut inner = self.inner.lock();
@@ -325,7 +329,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 +343,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 {
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..5259328 100644
--- a/src/select.rs
+++ b/src/select.rs
@@ -486,7 +486,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 +864,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..62defa2 100644
--- a/src/waker.rs
+++ b/src/waker.rs
@@ -8,22 +8,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: usize,
/// 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 +34,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 +43,13 @@
/// Registers a select operation.
#[inline]
- pub fn register(&mut self, oper: Operation, cx: &Context) {
+ pub(crate) fn register(&mut self, oper: Operation, cx: &Context) {
self.register_with_packet(oper, 0, 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: usize, cx: &Context) {
self.selectors.push(Entry {
oper,
packet,
@@ -59,7 +59,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,7 +75,7 @@
/// Attempts to find another thread's entry, select the operation, and wake it up.
#[inline]
- pub fn try_select(&mut self) -> Option<Entry> {
+ pub(crate) fn try_select(&mut self) -> Option<Entry> {
let mut entry = None;
if !self.selectors.is_empty() {
@@ -108,7 +108,7 @@
/// 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,7 +122,7 @@
/// 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,
@@ -132,13 +132,13 @@
/// 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 +148,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 +175,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 +186,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 +195,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 +206,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 +218,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 +234,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 +245,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 +256,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(
diff --git a/tests/array.rs b/tests/array.rs
index a7ae323..15c30a5 100644
--- a/tests/array.rs
+++ b/tests/array.rs
@@ -497,8 +497,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..cd70013 100644
--- a/tests/golang.rs
+++ b/tests/golang.rs
@@ -315,7 +315,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);
}
@@ -1507,5 +1507,58 @@
// https://github.com/golang/go/blob/master/test/ken/chan1.go
mod chan1 {
- // TODO
+ use super::*;
+
+ // sent messages
+ 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/list.rs b/tests/list.rs
index 8b84105..f166a19 100644
--- a/tests/list.rs
+++ b/tests/list.rs
@@ -387,8 +387,8 @@
let mut rng = thread_rng();
for _ in 0..100 {
- let steps = rng.gen_range(0, 10_000);
- let additional = rng.gen_range(0, 1000);
+ 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>();
diff --git a/tests/ready.rs b/tests/ready.rs
index 700f487..6779694 100644
--- a/tests/ready.rs
+++ b/tests/ready.rs
@@ -606,8 +606,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 +627,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!(),
}
diff --git a/tests/select.rs b/tests/select.rs
index 4cf08b6..fcc291e 100644
--- a/tests/select.rs
+++ b/tests/select.rs
@@ -809,8 +809,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 +833,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 +842,7 @@
Ok(oper) => match oper.index() {
ix if ix == oper1 => {
assert_eq!(oper.recv(&r), Ok(i));
- done = true;
+ break;
}
_ => unreachable!(),
},
diff --git a/tests/tick.rs b/tests/tick.rs
index 5dc8730..1273f64 100644
--- a/tests/tick.rs
+++ b/tests/tick.rs
@@ -127,6 +127,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 +252,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..1dd39e1 100644
--- a/tests/zero.rs
+++ b/tests/zero.rs
@@ -399,7 +399,7 @@
let mut rng = thread_rng();
for _ in 0..100 {
- let steps = rng.gen_range(0, 3_000);
+ let steps = rng.gen_range(0..3_000);
DROPS.store(0, Ordering::SeqCst);
let (s, r) = bounded::<DropCounter>(0);
