Upgrade rust/crates/crossbeam-channel to 0.5.1 am: f6b6a424b1 am: b930e5a382

Original change: https://android-review.googlesource.com/c/platform/external/rust/crates/crossbeam-channel/+/1712592

Change-Id: I9f8e9ccd992673f2a137df4ccdaecf67c407b02e

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"
   }
 }
diff --git a/Android.bp b/Android.bp
index 97ef73e..710a437 100644
--- a/Android.bp
+++ b/Android.bp
@@ -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.
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 @@ 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"
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 @@ 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
 
diff --git a/METADATA b/METADATA
index 00d9dde..c1dad0e 100644
--- a/METADATA
+++ b/METADATA
@@ -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
   }
 }
diff --git a/README.md b/README.md
index eab623a..f5077c5 100644
--- a/README.md
+++ b/README.md
@@ -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 {
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 @@ 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);