channel: unbounded: synchronize receiver disconnect with initialization

We must take into account the case where the channel has messages in the
block pointed to by `tail` but the head is still pointing to a null
pointer. This can happen with two concurrent senders if one gets
preempted during initialization.

Signed-off-by: Petros Angelatos <[email protected]>

Author: petrosagg

crossbeam-channel/src/flavors/list.rs

@@ -585,6 +585,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 {