Merge #973

973: v0.8: Prepare for the next release r=taiki-e a=taiki-e - crossbeam-channel 0.5.7 -> 0.5.8 - Fix race condition in unbounded channel. (#972) Also, yanking `>= 0.5.1, <= 0.5.7` that affected by the bug fixed in this release. Co-authored-by: Petros Angelatos <[email protected]> Co-authored-by: Taiki Endo <[email protected]>
crossbeam-rs · Apr 9, 2023 · 81ff802 · 81ff802
2 parents 721382b + 04fa0aa
commit 81ff802
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 1 deletion.
diff --git a/crossbeam-channel/CHANGELOG.md b/crossbeam-channel/CHANGELOG.md
@@ -1,17 +1,29 @@
+# Version 0.5.8
+
+- Fix race condition in unbounded channel. (#972)
+
 # Version 0.5.7
 
+**Note:** This release has been yanked due to bug fixed in 0.5.8.
+
 - Improve handling of very large timeout. (#953)
 
 # Version 0.5.6
 
+**Note:** This release has been yanked due to bug fixed in 0.5.8.
+
 - Bump the minimum supported Rust version to 1.38. (#877)
 
 # Version 0.5.5
 
+**Note:** This release has been yanked due to bug fixed in 0.5.8.
+
 - Replace Spinlock with Mutex. (#835)
 
 # Version 0.5.4
 
+**Note:** This release has been yanked due to bug fixed in 0.5.8.
+
 - Workaround a bug in upstream related to TLS access on AArch64 Linux. (#802)
 
 # Version 0.5.3
@@ -28,6 +40,8 @@
 
 # Version 0.5.1
 
+**Note:** This release has been yanked due to bug fixed in 0.5.8.
+
 - Fix memory leak in unbounded channel. (#669)
 
 # Version 0.5.0

diff --git a/crossbeam-channel/Cargo.toml b/crossbeam-channel/Cargo.toml
@@ -4,7 +4,7 @@ name = "crossbeam-channel"
 # - Update CHANGELOG.md
 # - Update README.md
 # - Create "crossbeam-channel-X.Y.Z" git tag
-version = "0.5.7"
+version = "0.5.8"
 edition = "2018"
 rust-version = "1.38"
 license = "MIT OR Apache-2.0"

diff --git a/crossbeam-channel/src/flavors/list.rs b/crossbeam-channel/src/flavors/list.rs
@@ -584,6 +584,17 @@ impl<T> Channel<T> {
         let mut head = self.head.index.load(Ordering::Acquire);
         let mut block = self.head.block.load(Ordering::Acquire);
 
+        // If we're going to be dropping messages we need to synchronize with initialization
+        if head >> SHIFT != tail >> SHIFT {
+            // The block can be null here only if a sender is in the process of initializing the
+            // channel while another sender managed to send a message by inserting it into the
+            // semi-initialized channel and advanced the tail.
+            // In that case, just wait until it gets initialized.
+            while block.is_null() {
+                backoff.snooze();
+                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 {