diff options
author | Joel Galenson <jgalenson@google.com> | 2021-05-20 13:43:21 +0000 |
---|---|---|
committer | Automerger Merge Worker <android-build-automerger-merge-worker@system.gserviceaccount.com> | 2021-05-20 13:43:21 +0000 |
commit | 9348c4f80fd0ad668f75520d513b3643d4f61083 (patch) | |
tree | db946445fe46f98da303f87fbd0d6c67c1cdd320 | |
parent | 344b95af37d33309f97ce8a57d40ebda16389350 (diff) | |
parent | f11109e31ab5745e05c534a4ffa81a04ba775b55 (diff) | |
download | crossbeam-channel-9348c4f80fd0ad668f75520d513b3643d4f61083.tar.gz |
Upgrade rust/crates/crossbeam-channel to 0.5.1 am: f6b6a424b1 am: b930e5a382 am: f11109e31a
Original change: https://android-review.googlesource.com/c/platform/external/rust/crates/crossbeam-channel/+/1712592
Change-Id: Ib59024e9bf2ce94d06ca5f9ddb7c9a8d25948b45
-rw-r--r-- | .cargo_vcs_info.json | 2 | ||||
-rw-r--r-- | Android.bp | 3 | ||||
-rw-r--r-- | CHANGELOG.md | 4 | ||||
-rw-r--r-- | Cargo.lock | 77 | ||||
-rw-r--r-- | Cargo.toml | 7 | ||||
-rw-r--r-- | Cargo.toml.orig | 7 | ||||
-rw-r--r-- | LICENSE-THIRD-PARTY | 32 | ||||
-rw-r--r-- | METADATA | 10 | ||||
-rw-r--r-- | README.md | 10 | ||||
-rw-r--r-- | examples/stopwatch.rs | 4 | ||||
-rw-r--r-- | src/channel.rs | 10 | ||||
-rw-r--r-- | src/counter.rs | 14 | ||||
-rw-r--r-- | src/flavors/array.rs | 50 | ||||
-rw-r--r-- | src/flavors/at.rs | 23 | ||||
-rw-r--r-- | src/flavors/list.rs | 116 | ||||
-rw-r--r-- | src/flavors/mod.rs | 12 | ||||
-rw-r--r-- | src/flavors/never.rs | 21 | ||||
-rw-r--r-- | src/flavors/tick.rs | 21 | ||||
-rw-r--r-- | src/flavors/zero.rs | 41 | ||||
-rw-r--r-- | src/lib.rs | 11 | ||||
-rw-r--r-- | src/select.rs | 5 | ||||
-rw-r--r-- | src/utils.rs | 12 | ||||
-rw-r--r-- | src/waker.rs | 46 | ||||
-rw-r--r-- | tests/array.rs | 4 | ||||
-rw-r--r-- | tests/golang.rs | 57 | ||||
-rw-r--r-- | tests/list.rs | 4 | ||||
-rw-r--r-- | tests/ready.rs | 8 | ||||
-rw-r--r-- | tests/select.rs | 8 | ||||
-rw-r--r-- | tests/tick.rs | 2 | ||||
-rw-r--r-- | tests/zero.rs | 2 |
30 files changed, 350 insertions, 273 deletions
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" } } @@ -44,7 +44,6 @@ license { rust_library { name: "libcrossbeam_channel", - // has rustc warnings host_supported: true, crate_name: "crossbeam_channel", srcs: ["src/lib.rs"], @@ -63,5 +62,5 @@ rust_library { // 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. @@ -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 @@ checksum = "cdb031dd78e28731d87d56cc8ffef4a8f36ca26c38fe2de700543e627f8a464a" [[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 @@ dependencies = [ [[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 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646" [[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 @@ dependencies = [ [[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 @@ dependencies = [ [[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 @@ dependencies = [ [[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 @@ dependencies = [ [[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" @@ -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 @@ default-features = false 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 @@ name = "crossbeam-channel" # - 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 @@ optional = true [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 @@ -7,13 +7,13 @@ third_party { } 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 } } @@ -2,7 +2,7 @@ [![Build Status](https://github.com/crossbeam-rs/crossbeam/workflows/CI/badge.svg)]( https://github.com/crossbeam-rs/crossbeam/actions) -[![License](https://img.shields.io/badge/license-MIT%20OR%20Apache--2.0-blue.svg)]( +[![License](https://img.shields.io/badge/license-MIT_OR_Apache--2.0-blue.svg)]( https://github.com/crossbeam-rs/crossbeam/tree/master/crossbeam-channel#license) [![Cargo](https://img.shields.io/crates/v/crossbeam-channel.svg)]( https://crates.io/crates/crossbeam-channel) @@ -10,7 +10,7 @@ https://crates.io/crates/crossbeam-channel) https://docs.rs/crossbeam-channel) [![Rust 1.36+](https://img.shields.io/badge/rust-1.36+-lightgray.svg)]( https://www.rust-lang.org) -[![chat](https://img.shields.io/discord/569610676205781012.svg?logo=discord)](https://discord.gg/BBYwKq) +[![chat](https://img.shields.io/discord/569610676205781012.svg?logo=discord)](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 @@ Add this to your `Cargo.toml`: ```toml [dependencies] -crossbeam-channel = "0.4" +crossbeam-channel = "0.5" ``` ## Compatibility @@ -73,10 +73,6 @@ This product includes copies and modifications of software developed by third pa [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 @@ fn main() { 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 @@ pub fn at(when: Instant) -> Receiver<Instant> { /// 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 @@ impl<T> Drop for Sender<T> { 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 @@ impl<T> Drop for Receiver<T> { 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 @@ impl<T> SelectHandle for Receiver<T> { } /// 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 @@ pub unsafe fn write<T>(s: &Sender<T>, token: &mut Token, msg: T) -> Result<(), T } /// 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 @@ struct Counter<C> { } /// 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 @@ pub fn new<C>(chan: C) -> (Sender<C>, Receiver<C>) { } /// The sending side. -pub struct Sender<C> { +pub(crate) struct Sender<C> { counter: *mut Counter<C>, } @@ -45,7 +45,7 @@ impl<C> Sender<C> { } /// 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 @@ impl<C> Sender<C> { /// 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 @@ impl<C> PartialEq for Sender<C> { } /// The receiving side. -pub struct Receiver<C> { +pub(crate) struct Receiver<C> { counter: *mut Counter<C>, } @@ -100,7 +100,7 @@ impl<C> Receiver<C> { } /// 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 @@ impl<C> Receiver<C> { /// 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 @@ impl Default for ArrayToken { } /// 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 @@ pub struct Channel<T> { 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Drop for Channel<T> { } /// 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::select::{Operation, SelectHandle, Token}; 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 @@ pub struct Channel { 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 @@ impl Channel { } /// 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 @@ impl Channel { /// 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 @@ impl Channel { /// 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 @@ impl Channel { /// 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 @@ impl Channel { } /// 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 @@ impl Default for ListToken { /// /// 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 @@ pub struct Channel<T> { 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } } + /// 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 @@ impl<T> Drop for Channel<T> { } /// 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::select::{Operation, SelectHandle, Token}; 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 @@ impl<T> Channel<T> { /// 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::err::{RecvTimeoutError, TryRecvError}; 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 @@ pub struct Channel { 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 @@ impl Channel { /// 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 @@ impl Channel { /// 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 @@ impl Channel { /// 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 @@ impl Channel { } /// 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::utils::Spinlock; 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 @@ struct Inner { } /// Zero-capacity channel. -pub struct Channel<T> { +pub(crate) struct Channel<T> { /// Inner representation of the channel. inner: Spinlock<Inner>, @@ -90,7 +90,7 @@ pub struct Channel<T> { 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { } /// 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 @@ impl<T> Channel<T> { /// 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 @@ impl<T> Channel<T> { } /// 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 { @@ -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 @@ pub fn select_timeout<'a>( /// 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 @@ impl<'a> Select<'a> { /// 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 std::time::{Duration, Instant}; 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 @@ pub fn shuffle<T>(v: &mut [T]) { } /// 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 @@ pub fn sleep_until(deadline: Option<Instant>) { } /// 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 @@ impl<T> Spinlock<T> { } /// 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 @@ impl<T> Spinlock<T> { } /// 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::select::{Operation, Selected}; 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 @@ pub struct Waker { 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 @@ impl Waker { /// 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 @@ impl Waker { /// 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 @@ impl Waker { /// 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 @@ impl Waker { /// 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 @@ impl Waker { /// 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 @@ impl Waker { /// 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 @@ impl Waker { /// 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 @@ impl Drop for Waker { /// 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 @@ pub struct SyncWaker { 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 @@ impl SyncWaker { /// 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 @@ impl SyncWaker { /// 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 @@ impl SyncWaker { /// 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 @@ impl SyncWaker { /// 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 @@ impl SyncWaker { /// 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 @@ impl SyncWaker { /// 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 @@ fn drops() { 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 @@ mod fifo { 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 @@ mod chan { // 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 @@ fn drops() { 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 @@ fn stress_timeout_two_threads() { 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 @@ fn stress_timeout_two_threads() { 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 @@ fn stress_timeout_two_threads() { 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 @@ fn stress_timeout_two_threads() { 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 @@ fn stress_timeout_two_threads() { 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 @@ fn recv() { 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 @@ fn select() { 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 @@ fn drops() { 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); |