Repository: rayon-rs/rayon
Pull request #907 (2022-01-04 01:03:37)

Address SB/provenance issues found by cargo miri test
I've submitted a few PRs to other libraries to address errors from running `cargo miri test` on them. Not all of the code that Miri rejects as not conforming to Stacked Borrows is necessarily invalid in terms of current rustc+LLVM semantics, but some of it is a little dubious on its own, and also I think it would be really nice for core libraries to be clean against miri so that we can use miri to understand how well people follow the prototype rules without Miri being halted by non-conforming code in dependencies.

To that end, I'm here because I was running `cargo miri test --features=rayon` in the `hashbrown` crate, and I got an error that I could trace back to this crate. Then I started running `cargo miri test` with `MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-tag-raw-pointers"` on this crate's tests, and I found a few things. This PR fixes everything I found. My motivation is a little bit of column A (some of this code is like... maybe let's not do it that way, but also not definitely broken) and a little bit of column B (I got here via another crate's tests).

In two cases, this code was going through the `Deref` to a slice impl on `Vec` in order to get a pointer into the uninitialized region between len and capacity. As far as I understand, under the Stacked Borrows model, references narrow provenance, thus it is not valid to use such a pointer outside the initialized region.

The collecting code has a related problem, where it is trying to merge slices that were originally derived from the same array by rebuilding a larger slice by using a pointer based on one of the subslices. To construct a valid merged slice we need access to the original provenance of the whole array. It is possible to resolve this by keeping a `&mut[]` as well as a `*mut`, using the raw pointer to construct the merged slice because it has sufficent provenance. But I think that ends up juggling a few too many pieces of data, so I implemented a fix by replacing the slice with a pointer and length. I also tried to add a bit more description to some names to help out in reading the more arcane implementation.

saethlin: By the way, related to @adamreichold's comment above, this code already exists in `rayon` to evade the `!Send` impl on raw pointers:

https://github.com/rayon-rs/rayon/blob/f45eee8fa49c1646a00f084ca78d362f381f1b65/src/slice/mergesort.rs#L500-L502
https://github.com/rayon-rs/rayon/blob/f45eee8fa49c1646a00f084ca78d362f381f1b65/src/slice/mergesort.rs#L601-L603

Is this the preferred mechanism? I think it would be best to be consistent across the codebase, and if possible I would selfishly prefer to use a `SendPtr` type as above, since at the moment miri just doesn't know what to do with the pointers that you get back out of an integer. Miri just throws up its hands and tosses an error (below). One could also argue that a `SendPtr` type communicates intent better than passing a `usize` that is secretly a pointer.
```
error: Undefined Behavior: no item granting write access to tag <untagged> at alloc5517403 found in borrow stack.
    --> /home/ben/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/intrinsics.rs:2083:14
     |
2083 |     unsafe { copy_nonoverlapping(src, dst, count) }
     |              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ no item granting write access to tag <untagged> at alloc5517403 found in borrow stack.
     |
     = help: this indicates a potential bug in the program: it performed an invalid operation, but the rules it violated are still experimental
     = help: see https://github.com/rust-lang/unsafe-code-guidelines/blob/master/wip/stacked-borrows.md for further information
             
     = note: inside `std::intrinsics::copy_nonoverlapping::<i32>` at /home/ben/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/intrinsics.rs:2083:14
     = note: inside `rayon::slice::mergesort::merge::<i32, &for<'r, 's> fn(&'r i32, &'s i32) -> bool {<i32 as std::cmp::PartialOrd>::lt}>` at /home/ben/rayon/src/slice/mergesort.rs:160:9
     = note: inside `rayon::slice::mergesort::mergesort::<i32, for<'r, 's> fn(&'r i32, &'s i32) -> bool {<i32 as std::cmp::PartialOrd>::lt}>` at /home/ben/rayon/src/slice/mergesort.rs:366:13
```
cuviper: > I think it would be best to be consistent across the codebase, and if possible I would selfishly prefer to use a `SendPtr` type as above, since at the moment miri just doesn't know what to do with the pointers that you get back out of an integer. Miri just throws up its hands and tosses an error (below). One could also argue that a `SendPtr` type communicates intent better than passing a `usize` that is secretly a pointer.

I agree, that sounds like a good idea.

Patch:
diff --git a/src/iter/collect/consumer.rs b/src/iter/collect/consumer.rs
index 3a8eea0..acd67df 100644
--- a/src/iter/collect/consumer.rs
+++ b/src/iter/collect/consumer.rs
@@ -1,19 +1,38 @@
 use super::super::plumbing::*;
+use crate::SendPtr;
 use std::marker::PhantomData;
-use std::mem::MaybeUninit;
 use std::ptr;
 use std::slice;
 
 pub(super) struct CollectConsumer<'c, T: Send> {
-    /// A slice covering the target memory, not yet initialized!
-    target: &'c mut [MaybeUninit<T>],
+    /// See `CollectResult` for explanation of why this is not a slice
+    start: SendPtr<T>,
+    len: usize,
+    marker: PhantomData<&'c mut T>,
+}
+
+impl<T: Send> CollectConsumer<'_, T> {
+    /// Create a collector for `len` items in the unused capacity of the vector.
+    pub(super) fn appender(vec: &mut Vec<T>, len: usize) -> CollectConsumer<'_, T> {
+        let start = vec.len();
+        assert!(vec.capacity() - start >= len);
+
+        // SAFETY: We already made sure to have the additional space allocated.
+        // The pointer is derived from `Vec` directly, not through a `Deref`,
+        // so it has provenance over the whole allocation.
+        unsafe { CollectConsumer::new(vec.as_mut_ptr().add(start), len) }
+    }
 }
 
 impl<'c, T: Send + 'c> CollectConsumer<'c, T> {
     /// The target memory is considered uninitialized, and will be
     /// overwritten without reading or dropping existing values.
-    pub(super) fn new(target: &'c mut [MaybeUninit<T>]) -> Self {
-        CollectConsumer { target }
+    unsafe fn new(start: *mut T, len: usize) -> Self {
+        CollectConsumer {
+            start: SendPtr(start),
+            len,
+            marker: PhantomData,
+        }
     }
 }
 
@@ -23,10 +42,13 @@ impl<'c, T: Send + 'c> CollectConsumer<'c, T> {
 /// the elements will be dropped, unless its ownership is released before then.
 #[must_use]
 pub(super) struct CollectResult<'c, T> {
-    /// A slice covering the target memory, initialized up to our separate `len`.
-    target: &'c mut [MaybeUninit<T>],
-    /// The current initialized length in `target`
-    len: usize,
+    /// This pointer and length has the same representation as a slice,
+    /// but retains the provenance of the entire array so that we can merge
+    /// these regions together in `CollectReducer`.
+    start: SendPtr<T>,
+    total_len: usize,
+    /// The current initialized length after `start`
+    initialized_len: usize,
     /// Lifetime invariance guarantees that the data flows from consumer to result,
     /// especially for the `scope_fn` callback in `Collect::with_consumer`.
     invariant_lifetime: PhantomData<&'c mut &'c mut [T]>,
@@ -37,25 +59,26 @@ unsafe impl<'c, T> Send for CollectResult<'c, T> where T: Send {}
 impl<'c, T> CollectResult<'c, T> {
     /// The current length of the collect result
     pub(super) fn len(&self) -> usize {
-        self.len
+        self.initialized_len
     }
 
     /// Release ownership of the slice of elements, and return the length
     pub(super) fn release_ownership(mut self) -> usize {
-        let ret = self.len;
-        self.len = 0;
+        let ret = self.initialized_len;
+        self.initialized_len = 0;
         ret
     }
 }
 
 impl<'c, T> Drop for CollectResult<'c, T> {
     fn drop(&mut self) {
-        // Drop the first `self.len` elements, which have been recorded
+        // Drop the first `self.initialized_len` elements, which have been recorded
         // to be initialized by the folder.
         unsafe {
-            // TODO: use `MaybeUninit::slice_as_mut_ptr`
-            let start = self.target.as_mut_ptr() as *mut T;
-            ptr::drop_in_place(slice::from_raw_parts_mut(start, self.len));
+            ptr::drop_in_place(slice::from_raw_parts_mut(
+                self.start.0,
+                self.initialized_len,
+            ));
         }
     }
 }
@@ -66,24 +89,27 @@ impl<'c, T: Send + 'c> Consumer<T> for CollectConsumer<'c, T> {
     type Result = CollectResult<'c, T>;
 
     fn split_at(self, index: usize) -> (Self, Self, CollectReducer) {
-        let CollectConsumer { target } = self;
-
-        // Produce new consumers. Normal slicing ensures that the
-        // memory range given to each consumer is disjoint.
-        let (left, right) = target.split_at_mut(index);
-        (
-            CollectConsumer::new(left),
-            CollectConsumer::new(right),
-            CollectReducer,
-        )
+        let CollectConsumer { start, len, .. } = self;
+
+        // Produce new consumers.
+        // SAFETY: This assert checks that `index` is a valid offset for `start`
+        unsafe {
+            assert!(index <= len);
+            (
+                CollectConsumer::new(start.0, index),
+                CollectConsumer::new(start.0.add(index), len - index),
+                CollectReducer,
+            )
+        }
     }
 
     fn into_folder(self) -> Self::Folder {
         // Create a result/folder that consumes values and writes them
-        // into target. The initial result has length 0.
+        // into the region after start. The initial result has length 0.
         CollectResult {
-            target: self.target,
-            len: 0,
+            start: self.start,
+            total_len: self.len,
+            initialized_len: 0,
             invariant_lifetime: PhantomData,
         }
     }
@@ -97,15 +123,17 @@ impl<'c, T: Send + 'c> Folder<T> for CollectResult<'c, T> {
     type Result = Self;
 
     fn consume(mut self, item: T) -> Self {
-        let dest = self
-            .target
-            .get_mut(self.len)
-            .expect("too many values pushed to consumer");
+        assert!(
+            self.initialized_len < self.total_len,
+            "too many values pushed to consumer"
+        );
 
-        // Write item and increase the initialized length
+        // SAFETY: The assert above is a bounds check for this write, and we
+        // avoid assignment here so we do not drop an uninitialized T.
         unsafe {
-            dest.as_mut_ptr().write(item);
-            self.len += 1;
+            // Write item and increase the initialized length
+            self.start.0.add(self.initialized_len).write(item);
+            self.initialized_len += 1;
         }
 
         self
@@ -146,14 +174,13 @@ impl<'c, T> Reducer<CollectResult<'c, T>> for CollectReducer {
         // Merge if the CollectResults are adjacent and in left to right order
         // else: drop the right piece now and total length will end up short in the end,
         // when the correctness of the collected result is asserted.
-        let left_end = left.target[left.len..].as_ptr();
-        if left_end == right.target.as_ptr() {
-            let len = left.len + right.release_ownership();
-            unsafe {
-                left.target = slice::from_raw_parts_mut(left.target.as_mut_ptr(), len);
+        unsafe {
+            let left_end = left.start.0.add(left.initialized_len);
+            if left_end == right.start.0 {
+                left.total_len += right.total_len;
+                left.initialized_len += right.release_ownership();
             }
-            left.len = len;
+            left
         }
-        left
     }
 }
diff --git a/src/iter/collect/mod.rs b/src/iter/collect/mod.rs
index f4ff49a..4044a68 100644
--- a/src/iter/collect/mod.rs
+++ b/src/iter/collect/mod.rs
@@ -1,6 +1,4 @@
 use super::{IndexedParallelIterator, ParallelIterator};
-use std::mem::MaybeUninit;
-use std::slice;
 
 mod consumer;
 use self::consumer::CollectConsumer;
@@ -19,7 +17,7 @@ where
 {
     v.truncate(0); // clear any old data
     let len = pi.len();
-    Collect::new(v, len).with_consumer(|consumer| pi.drive(consumer));
+    collect_with_consumer(v, len, |consumer| pi.drive(consumer));
 }
 
 /// Collects the results of the iterator into the specified vector.
@@ -38,7 +36,7 @@ where
     I: ParallelIterator<Item = T>,
     T: Send,
 {
-    Collect::new(v, len).with_consumer(|consumer| pi.drive_unindexed(consumer));
+    collect_with_consumer(v, len, |consumer| pi.drive_unindexed(consumer));
 }
 
 /// Unzips the results of the exact iterator into the specified vectors.
@@ -55,9 +53,9 @@ where
     right.truncate(0);
 
     let len = pi.len();
-    Collect::new(right, len).with_consumer(|right_consumer| {
+    collect_with_consumer(right, len, |right_consumer| {
         let mut right_result = None;
-        Collect::new(left, len).with_consumer(|left_consumer| {
+        collect_with_consumer(left, len, |left_consumer| {
             let (left_r, right_r) = unzip_indexed(pi, left_consumer, right_consumer);
             right_result = Some(right_r);
             left_r
@@ -66,78 +64,53 @@ where
     });
 }
 
-/// Manage the collection vector.
-struct Collect<'c, T: Send> {
-    vec: &'c mut Vec<T>,
-    len: usize,
-}
-
-impl<'c, T: Send + 'c> Collect<'c, T> {
-    fn new(vec: &'c mut Vec<T>, len: usize) -> Self {
-        Collect { vec, len }
-    }
-
-    /// Create a consumer on the slice of memory we are collecting into.
-    ///
-    /// The consumer needs to be used inside the scope function, and the
-    /// complete collect result passed back.
-    ///
-    /// This method will verify the collect result, and panic if the slice
-    /// was not fully written into. Otherwise, in the successful case,
-    /// the vector is complete with the collected result.
-    fn with_consumer<F>(mut self, scope_fn: F)
-    where
-        F: FnOnce(CollectConsumer<'_, T>) -> CollectResult<'_, T>,
-    {
-        let slice = Self::reserve_get_tail_slice(&mut self.vec, self.len);
-        let result = scope_fn(CollectConsumer::new(slice));
-
-        // The CollectResult represents a contiguous part of the
-        // slice, that has been written to.
-        // On unwind here, the CollectResult will be dropped.
-        // If some producers on the way did not produce enough elements,
-        // partial CollectResults may have been dropped without
-        // being reduced to the final result, and we will see
-        // that as the length coming up short.
-        //
-        // Here, we assert that `slice` is fully initialized. This is
-        // checked by the following assert, which verifies if a
-        // complete CollectResult was produced; if the length is
-        // correct, it is necessarily covering the target slice.
-        // Since we know that the consumer cannot have escaped from
-        // `drive` (by parametricity, essentially), we know that any
-        // stores that will happen, have happened. Unless some code is buggy,
-        // that means we should have seen `len` total writes.
-        let actual_writes = result.len();
-        assert!(
-            actual_writes == self.len,
-            "expected {} total writes, but got {}",
-            self.len,
-            actual_writes
-        );
+/// Create a consumer on the slice of memory we are collecting into.
+///
+/// The consumer needs to be used inside the scope function, and the
+/// complete collect result passed back.
+///
+/// This method will verify the collect result, and panic if the slice
+/// was not fully written into. Otherwise, in the successful case,
+/// the vector is complete with the collected result.
+fn collect_with_consumer<T, F>(vec: &mut Vec<T>, len: usize, scope_fn: F)
+where
+    T: Send,
+    F: FnOnce(CollectConsumer<'_, T>) -> CollectResult<'_, T>,
+{
+    // Reserve space for `len` more elements in the vector,
+    vec.reserve(len);
 
-        // Release the result's mutable borrow and "proxy ownership"
-        // of the elements, before the vector takes it over.
-        result.release_ownership();
+    // Create the consumer and run the callback for collection.
+    let result = scope_fn(CollectConsumer::appender(vec, len));
 
-        let new_len = self.vec.len() + self.len;
+    // The `CollectResult` represents a contiguous part of the slice, that has
+    // been written to. On unwind here, the `CollectResult` will be dropped. If
+    // some producers on the way did not produce enough elements, partial
+    // `CollectResult`s may have been dropped without being reduced to the final
+    // result, and we will see that as the length coming up short.
+    //
+    // Here, we assert that added length is fully initialized. This is checked
+    // by the following assert, which verifies if a complete `CollectResult`
+    // was produced; if the length is correct, it is necessarily covering the
+    // target slice. Since we know that the consumer cannot have escaped from
+    // `drive` (by parametricity, essentially), we know that any stores that
+    // will happen, have happened. Unless some code is buggy, that means we
+    // should have seen `len` total writes.
+    let actual_writes = result.len();
+    assert!(
+        actual_writes == len,
+        "expected {} total writes, but got {}",
+        len,
+        actual_writes
+    );
 
-        unsafe {
-            self.vec.set_len(new_len);
-        }
-    }
+    // Release the result's mutable borrow and "proxy ownership"
+    // of the elements, before the vector takes it over.
+    result.release_ownership();
 
-    /// Reserve space for `len` more elements in the vector,
-    /// and return a slice to the uninitialized tail of the vector
-    fn reserve_get_tail_slice(vec: &mut Vec<T>, len: usize) -> &mut [MaybeUninit<T>] {
-        // Reserve the new space.
-        vec.reserve(len);
+    let new_len = vec.len() + len;
 
-        // TODO: use `Vec::spare_capacity_mut` instead
-        // SAFETY: `MaybeUninit<T>` is guaranteed to have the same layout
-        // as `T`, and we already made sure to have the additional space.
-        let start = vec.len();
-        let tail_ptr = vec[start..].as_mut_ptr() as *mut MaybeUninit<T>;
-        unsafe { slice::from_raw_parts_mut(tail_ptr, len) }
+    unsafe {
+        vec.set_len(new_len);
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index b1e6fab..6f9271a 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -119,3 +119,29 @@ pub use rayon_core::{in_place_scope, scope, Scope};
 pub use rayon_core::{in_place_scope_fifo, scope_fifo, ScopeFifo};
 pub use rayon_core::{join, join_context};
 pub use rayon_core::{spawn, spawn_fifo};
+
+/// We need to transmit raw pointers across threads. It is possible to do this
+/// without any unsafe code by converting pointers to usize or to AtomicPtr<T>
+/// then back to a raw pointer for use. We prefer this approach because code
+/// that uses this type is more explicit.
+///
+/// Unsafe code is still required to dereference the pointer, so this type is
+/// not unsound on its own, although it does partly lift the unconditional
+/// !Send and !Sync on raw pointers. As always, dereference with care.
+struct SendPtr<T>(*mut T);
+
+// SAFETY: !Send for raw pointers is not for safety, just as a lint
+unsafe impl<T: Send> Send for SendPtr<T> {}
+
+// SAFETY: !Sync for raw pointers is not for safety, just as a lint
+unsafe impl<T: Send> Sync for SendPtr<T> {}
+
+// Implement Clone without the T: Clone bound from the derive
+impl<T> Clone for SendPtr<T> {
+    fn clone(&self) -> Self {
+        Self(self.0)
+    }
+}
+
+// Implement Copy without the T: Copy bound from the derive
+impl<T> Copy for SendPtr<T> {}
diff --git a/src/slice/mergesort.rs b/src/slice/mergesort.rs
index e9a5d43..a24ba65 100644
--- a/src/slice/mergesort.rs
+++ b/src/slice/mergesort.rs
@@ -6,6 +6,7 @@
 
 use crate::iter::*;
 use crate::slice::ParallelSliceMut;
+use crate::SendPtr;
 use std::mem;
 use std::mem::size_of;
 use std::ptr;
@@ -24,7 +25,7 @@ unsafe fn decrement_and_get<T>(ptr: &mut *mut T) -> *mut T {
 
 /// When dropped, copies from `src` into `dest` a sequence of length `len`.
 struct CopyOnDrop<T> {
-    src: *mut T,
+    src: *const T,
     dest: *mut T,
     len: usize,
 }
@@ -63,9 +64,7 @@ where
             //    performance than with the 2nd method.
             //
             // All methods were benchmarked, and the 3rd showed best results. So we chose that one.
-            let mut tmp = NoDrop {
-                value: Some(ptr::read(&v[0])),
-            };
+            let tmp = mem::ManuallyDrop::new(ptr::read(&v[0]));
 
             // Intermediate state of the insertion process is always tracked by `hole`, which
             // serves two purposes:
@@ -78,13 +77,13 @@ where
             // fill the hole in `v` with `tmp`, thus ensuring that `v` still holds every object it
             // initially held exactly once.
             let mut hole = InsertionHole {
-                src: tmp.value.as_mut().unwrap(),
+                src: &*tmp,
                 dest: &mut v[1],
             };
             ptr::copy_nonoverlapping(&v[1], &mut v[0], 1);
 
             for i in 2..v.len() {
-                if !is_less(&v[i], tmp.value.as_ref().unwrap()) {
+                if !is_less(&v[i], &*tmp) {
                     break;
                 }
                 ptr::copy_nonoverlapping(&v[i], &mut v[i - 1], 1);
@@ -94,20 +93,9 @@ where
         }
     }
 
-    // Holds a value, but never drops it.
-    struct NoDrop<T> {
-        value: Option<T>,
-    }
-
-    impl<T> Drop for NoDrop<T> {
-        fn drop(&mut self) {
-            mem::forget(self.value.take());
-        }
-    }
-
     // When dropped, copies from `src` into `dest`.
     struct InsertionHole<T> {
-        src: *mut T,
+        src: *const T,
         dest: *mut T,
     }
 
@@ -219,6 +207,7 @@ where
             // `T` is not a zero-sized type, so it's okay to divide by its size.
             let len = (self.end as usize - self.start as usize) / size_of::<T>();
             unsafe {
+                // TODO 1.47: let len = self.end.offset_from(self.start) as usize;
                 ptr::copy_nonoverlapping(self.start, self.dest, len);
             }
         }
@@ -284,7 +273,7 @@ fn collapse(runs: &[Run]) -> Option<usize> {
 /// Otherwise, it sorts the slice into non-descending order.
 ///
 /// This merge sort borrows some (but not all) ideas from TimSort, which is described in detail
-/// [here](https://svn.python.org/projects/python/trunk/Objects/listsort.txt).
+/// [here](https://github.com/python/cpython/blob/main/Objects/listsort.txt).
 ///
 /// The algorithm identifies strictly descending and non-descending subsequences, which are called
 /// natural runs. There is a stack of pending runs yet to be merged. Each newly found run is pushed
@@ -294,7 +283,7 @@ fn collapse(runs: &[Run]) -> Option<usize> {
 /// 1. for every `i` in `1..runs.len()`: `runs[i - 1].len > runs[i].len`
 /// 2. for every `i` in `2..runs.len()`: `runs[i - 2].len > runs[i - 1].len + runs[i].len`
 ///
-/// The invariants ensure that the total running time is `O(n log n)` worst-case.
+/// The invariants ensure that the total running time is *O*(*n* \* log(*n*)) worst-case.
 ///
 /// # Safety
 ///
@@ -497,12 +486,13 @@ where
         // get copied into `dest_left` and `dest_right``.
         mem::forget(s);
 
-        // Convert the pointers to `usize` because `*mut T` is not `Send`.
-        let dest_l = dest as usize;
-        let dest_r = dest.add(left_l.len() + right_l.len()) as usize;
+        // Wrap pointers in SendPtr so that they can be sent to another thread
+        // See the documentation of SendPtr for a full explanation
+        let dest_l = SendPtr(dest);
+        let dest_r = SendPtr(dest.add(left_l.len() + right_l.len()));
         rayon_core::join(
-            || par_merge(left_l, right_l, dest_l as *mut T, is_less),
-            || par_merge(left_r, right_r, dest_r as *mut T, is_less),
+            || par_merge(left_l, right_l, dest_l.0, is_less),
+            || par_merge(left_r, right_r, dest_r.0, is_less),
         );
     }
     // Finally, `s` gets dropped if we used sequential merge, thus copying the remaining elements
@@ -598,12 +588,13 @@ unsafe fn recurse<T, F>(
         len: end - start,
     };
 
-    // Convert the pointers to `usize` because `*mut T` is not `Send`.
-    let v = v as usize;
-    let buf = buf as usize;
+    // Wrap pointers in SendPtr so that they can be sent to another thread
+    // See the documentation of SendPtr for a full explanation
+    let v = SendPtr(v);
+    let buf = SendPtr(buf);
     rayon_core::join(
-        || recurse(v as *mut T, buf as *mut T, left, !into_buf, is_less),
-        || recurse(v as *mut T, buf as *mut T, right, !into_buf, is_less),
+        || recurse(v.0, buf.0, left, !into_buf, is_less),
+        || recurse(v.0, buf.0, right, !into_buf, is_less),
     );
 
     // Everything went all right - recursive calls didn't panic.
@@ -667,8 +658,9 @@ where
     // Split the slice into chunks and merge sort them in parallel.
     // However, descending chunks will not be sorted - they will be simply left intact.
     let mut iter = {
-        // Convert the pointer to `usize` because `*mut T` is not `Send`.
-        let buf = buf as usize;
+        // Wrap pointer in SendPtr so that it can be sent to another thread
+        // See the documentation of SendPtr for a full explanation
+        let buf = SendPtr(buf);
 
         v.par_chunks_mut(CHUNK_LENGTH)
             .with_max_len(1)
@@ -677,7 +669,7 @@ where
                 let l = CHUNK_LENGTH * i;
                 let r = l + chunk.len();
                 unsafe {
-                    let buf = (buf as *mut T).add(l);
+                    let buf = buf.0.add(l);
                     (l, r, mergesort(chunk, buf, &is_less))
                 }
             })
diff --git a/src/slice/quicksort.rs b/src/slice/quicksort.rs
index 17c6f33..87fb724 100644
--- a/src/slice/quicksort.rs
+++ b/src/slice/quicksort.rs
@@ -5,46 +5,20 @@
 //! `rayon_core::join`.
 
 use std::cmp;
-use std::mem;
+use std::mem::{self, MaybeUninit};
 use std::ptr;
 
-/// When dropped, takes the value out of `Option` and writes it into `dest`.
-///
-/// This allows us to safely read the pivot into a stack-allocated variable for efficiency, and
-/// write it back into the slice after partitioning. This way we ensure that the write happens
-/// even if `is_less` panics in the meantime.
-struct WriteOnDrop<T> {
-    value: Option<T>,
-    dest: *mut T,
-}
-
-impl<T> Drop for WriteOnDrop<T> {
-    fn drop(&mut self) {
-        unsafe {
-            ptr::write(self.dest, self.value.take().unwrap());
-        }
-    }
-}
-
-/// Holds a value, but never drops it.
-struct NoDrop<T> {
-    value: Option<T>,
-}
-
-impl<T> Drop for NoDrop<T> {
-    fn drop(&mut self) {
-        mem::forget(self.value.take());
-    }
-}
-
 /// When dropped, copies from `src` into `dest`.
 struct CopyOnDrop<T> {
-    src: *mut T,
+    src: *const T,
     dest: *mut T,
 }
 
 impl<T> Drop for CopyOnDrop<T> {
     fn drop(&mut self) {
+        // SAFETY:  This is a helper class.
+        //          Please refer to its usage for correctness.
+        //          Namely, one must be sure that `src` and `dst` does not overlap as required by `ptr::copy_nonoverlapping`.
         unsafe {
             ptr::copy_nonoverlapping(self.src, self.dest, 1);
         }
@@ -57,29 +31,43 @@ where
     F: Fn(&T, &T) -> bool,
 {
     let len = v.len();
+    // SAFETY: The unsafe operations below involves indexing without a bounds check (by offsetting a
+    // pointer) and copying memory (`ptr::copy_nonoverlapping`).
+    //
+    // a. Indexing:
+    //  1. We checked the size of the array to >=2.
+    //  2. All the indexing that we will do is always between {0 <= index < len} at most.
+    //
+    // b. Memory copying
+    //  1. We are obtaining pointers to references which are guaranteed to be valid.
+    //  2. They cannot overlap because we obtain pointers to difference indices of the slice.
+    //     Namely, `i` and `i-1`.
+    //  3. If the slice is properly aligned, the elements are properly aligned.
+    //     It is the caller's responsibility to make sure the slice is properly aligned.
+    //
+    // See comments below for further detail.
     unsafe {
         // If the first two elements are out-of-order...
         if len >= 2 && is_less(v.get_unchecked(1), v.get_unchecked(0)) {
             // Read the first element into a stack-allocated variable. If a following comparison
             // operation panics, `hole` will get dropped and automatically write the element back
             // into the slice.
-            let mut tmp = NoDrop {
-                value: Some(ptr::read(v.get_unchecked(0))),
-            };
+            let tmp = mem::ManuallyDrop::new(ptr::read(v.get_unchecked(0)));
+            let v = v.as_mut_ptr();
             let mut hole = CopyOnDrop {
-                src: tmp.value.as_mut().unwrap(),
-                dest: v.get_unchecked_mut(1),
+                src: &*tmp,
+                dest: v.add(1),
             };
-            ptr::copy_nonoverlapping(v.get_unchecked(1), v.get_unchecked_mut(0), 1);
+            ptr::copy_nonoverlapping(v.add(1), v.add(0), 1);
 
             for i in 2..len {
-                if !is_less(v.get_unchecked(i), tmp.value.as_ref().unwrap()) {
+                if !is_less(&*v.add(i), &*tmp) {
                     break;
                 }
 
                 // Move `i`-th element one place to the left, thus shifting the hole to the right.
-                ptr::copy_nonoverlapping(v.get_unchecked(i), v.get_unchecked_mut(i - 1), 1);
-                hole.dest = v.get_unchecked_mut(i);
+                ptr::copy_nonoverlapping(v.add(i), v.add(i - 1), 1);
+                hole.dest = v.add(i);
             }
             // `hole` gets dropped and thus copies `tmp` into the remaining hole in `v`.
         }
@@ -92,29 +80,43 @@ where
     F: Fn(&T, &T) -> bool,
 {
     let len = v.len();
+    // SAFETY: The unsafe operations below involves indexing without a bound check (by offsetting a
+    // pointer) and copying memory (`ptr::copy_nonoverlapping`).
+    //
+    // a. Indexing:
+    //  1. We checked the size of the array to >= 2.
+    //  2. All the indexing that we will do is always between `0 <= index < len-1` at most.
+    //
+    // b. Memory copying
+    //  1. We are obtaining pointers to references which are guaranteed to be valid.
+    //  2. They cannot overlap because we obtain pointers to difference indices of the slice.
+    //     Namely, `i` and `i+1`.
+    //  3. If the slice is properly aligned, the elements are properly aligned.
+    //     It is the caller's responsibility to make sure the slice is properly aligned.
+    //
+    // See comments below for further detail.
     unsafe {
         // If the last two elements are out-of-order...
         if len >= 2 && is_less(v.get_unchecked(len - 1), v.get_unchecked(len - 2)) {
             // Read the last element into a stack-allocated variable. If a following comparison
             // operation panics, `hole` will get dropped and automatically write the element back
             // into the slice.
-            let mut tmp = NoDrop {
-                value: Some(ptr::read(v.get_unchecked(len - 1))),
-            };
+            let tmp = mem::ManuallyDrop::new(ptr::read(v.get_unchecked(len - 1)));
+            let v = v.as_mut_ptr();
             let mut hole = CopyOnDrop {
-                src: tmp.value.as_mut().unwrap(),
-                dest: v.get_unchecked_mut(len - 2),
+                src: &*tmp,
+                dest: v.add(len - 2),
             };
-            ptr::copy_nonoverlapping(v.get_unchecked(len - 2), v.get_unchecked_mut(len - 1), 1);
+            ptr::copy_nonoverlapping(v.add(len - 2), v.add(len - 1), 1);
 
             for i in (0..len - 2).rev() {
-                if !is_less(&tmp.value.as_ref().unwrap(), v.get_unchecked(i)) {
+                if !is_less(&*tmp, &*v.add(i)) {
                     break;
                 }
 
                 // Move `i`-th element one place to the right, thus shifting the hole to the left.
-                ptr::copy_nonoverlapping(v.get_unchecked(i), v.get_unchecked_mut(i + 1), 1);
-                hole.dest = v.get_unchecked_mut(i);
+                ptr::copy_nonoverlapping(v.add(i), v.add(i + 1), 1);
+                hole.dest = v.add(i);
             }
             // `hole` gets dropped and thus copies `tmp` into the remaining hole in `v`.
         }
@@ -123,7 +125,7 @@ where
 
 /// Partially sorts a slice by shifting several out-of-order elements around.
 ///
-/// Returns `true` if the slice is sorted at the end. This function is `O(n)` worst-case.
+/// Returns `true` if the slice is sorted at the end. This function is *O*(*n*) worst-case.
 #[cold]
 fn partial_insertion_sort<T, F>(v: &mut [T], is_less: &F) -> bool
 where
@@ -138,6 +140,8 @@ where
     let mut i = 1;
 
     for _ in 0..MAX_STEPS {
+        // SAFETY: We already explicitly did the bound checking with `i < len`.
+        // All our subsequent indexing is only in the range `0 <= index < len`
         unsafe {
             // Find the next pair of adjacent out-of-order elements.
             while i < len && !is_less(v.get_unchecked(i), v.get_unchecked(i - 1)) {
@@ -168,17 +172,17 @@ where
     false
 }
 
-/// Sorts a slice using insertion sort, which is `O(n^2)` worst-case.
+/// Sorts a slice using insertion sort, which is *O*(*n*^2) worst-case.
 fn insertion_sort<T, F>(v: &mut [T], is_less: &F)
 where
     F: Fn(&T, &T) -> bool,
 {
     for i in 1..v.len() {
-        shift_tail(&mut v[..=i], is_less);
+        shift_tail(&mut v[..i + 1], is_less);
     }
 }
 
-/// Sorts `v` using heapsort, which guarantees `O(n log n)` worst-case.
+/// Sorts `v` using heapsort, which guarantees *O*(*n* \* log(*n*)) worst-case.
 #[cold]
 fn heapsort<T, F>(v: &mut [T], is_less: &F)
 where
@@ -250,23 +254,30 @@ where
     // 3. `end` - End pointer into the `offsets` array.
     // 4. `offsets - Indices of out-of-order elements within the block.
 
-    // The current block on the left side (from `l` to `l.offset(block_l)`).
+    // The current block on the left side (from `l` to `l.add(block_l)`).
     let mut l = v.as_mut_ptr();
     let mut block_l = BLOCK;
     let mut start_l = ptr::null_mut();
     let mut end_l = ptr::null_mut();
-    let mut offsets_l = [0u8; BLOCK];
+    let mut offsets_l = [MaybeUninit::<u8>::uninit(); BLOCK];
 
-    // The current block on the right side (from `r.offset(-block_r)` to `r`).
+    // The current block on the right side (from `r.sub(block_r)` to `r`).
+    // SAFETY: The documentation for .add() specifically mention that `vec.as_ptr().add(vec.len())` is always safe`
     let mut r = unsafe { l.add(v.len()) };
     let mut block_r = BLOCK;
     let mut start_r = ptr::null_mut();
     let mut end_r = ptr::null_mut();
-    let mut offsets_r = [0u8; BLOCK];
+    let mut offsets_r = [MaybeUninit::<u8>::uninit(); BLOCK];
+
+    // FIXME: When we get VLAs, try creating one array of length `min(v.len(), 2 * BLOCK)` rather
+    // than two fixed-size arrays of length `BLOCK`. VLAs might be more cache-efficient.
 
     // Returns the number of elements between pointers `l` (inclusive) and `r` (exclusive).
     fn width<T>(l: *mut T, r: *mut T) -> usize {
         assert!(mem::size_of::<T>() > 0);
+        // FIXME: this should *likely* use `offset_from`, but more
+        // investigation is needed (including running tests in miri).
+        // TODO unstable: (r.addr() - l.addr()) / mem::size_of::<T>()
         (r as usize - l as usize) / mem::size_of::<T>()
     }
 
@@ -289,6 +300,9 @@ where
             } else if start_r < end_r {
                 block_l = rem;
             } else {
+                // There were the same number of elements to switch on both blocks during the last
+                // iteration, so there are no remaining elements on either block. Cover the remaining
+                // items with roughly equally-sized blocks.
                 block_l = rem / 2;
                 block_r = rem - block_l;
             }
@@ -298,11 +312,22 @@ where
 
         if start_l == end_l {
             // Trace `block_l` elements from the left side.
-            start_l = offsets_l.as_mut_ptr();
-            end_l = offsets_l.as_mut_ptr();
+            // TODO unstable: start_l = MaybeUninit::slice_as_mut_ptr(&mut offsets_l);
+            start_l = offsets_l.as_mut_ptr() as *mut u8;
+            end_l = start_l;
             let mut elem = l;
 
             for i in 0..block_l {
+                // SAFETY: The unsafety operations below involve the usage of the `offset`.
+                //         According to the conditions required by the function, we satisfy them because:
+                //         1. `offsets_l` is stack-allocated, and thus considered separate allocated object.
+                //         2. The function `is_less` returns a `bool`.
+                //            Casting a `bool` will never overflow `isize`.
+                //         3. We have guaranteed that `block_l` will be `<= BLOCK`.
+                //            Plus, `end_l` was initially set to the begin pointer of `offsets_` which was declared on the stack.
+                //            Thus, we know that even in the worst case (all invocations of `is_less` returns false) we will only be at most 1 byte pass the end.
+                //        Another unsafety operation here is dereferencing `elem`.
+                //        However, `elem` was initially the begin pointer to the slice which is always valid.
                 unsafe {
                     // Branchless comparison.
                     *end_l = i as u8;
@@ -314,11 +339,23 @@ where
 
         if start_r == end_r {
             // Trace `block_r` elements from the right side.
-            start_r = offsets_r.as_mut_ptr();
-            end_r = offsets_r.as_mut_ptr();
+            // TODO unstable: start_r = MaybeUninit::slice_as_mut_ptr(&mut offsets_r);
+            start_r = offsets_r.as_mut_ptr() as *mut u8;
+            end_r = start_r;
             let mut elem = r;
 
             for i in 0..block_r {
+                // SAFETY: The unsafety operations below involve the usage of the `offset`.
+                //         According to the conditions required by the function, we satisfy them because:
+                //         1. `offsets_r` is stack-allocated, and thus considered separate allocated object.
+                //         2. The function `is_less` returns a `bool`.
+                //            Casting a `bool` will never overflow `isize`.
+                //         3. We have guaranteed that `block_r` will be `<= BLOCK`.
+                //            Plus, `end_r` was initially set to the begin pointer of `offsets_` which was declared on the stack.
+                //            Thus, we know that even in the worst case (all invocations of `is_less` returns true) we will only be at most 1 byte pass the end.
+                //        Another unsafety operation here is dereferencing `elem`.
+                //        However, `elem` was initially `1 * sizeof(T)` past the end and we decrement it by `1 * sizeof(T)` before accessing it.
+                //        Plus, `block_r` was asserted to be less than `BLOCK` and `elem` will therefore at most be pointing to the beginning of the slice.
                 unsafe {
                     // Branchless comparison.
                     elem = elem.offset(-1);
@@ -346,6 +383,22 @@ where
             // Instead of swapping one pair at the time, it is more efficient to perform a cyclic
             // permutation. This is not strictly equivalent to swapping, but produces a similar
             // result using fewer memory operations.
+
+            // SAFETY: The use of `ptr::read` is valid because there is at least one element in
+            // both `offsets_l` and `offsets_r`, so `left!` is a valid pointer to read from.
+            //
+            // The uses of `left!` involve calls to `offset` on `l`, which points to the
+            // beginning of `v`. All the offsets pointed-to by `start_l` are at most `block_l`, so
+            // these `offset` calls are safe as all reads are within the block. The same argument
+            // applies for the uses of `right!`.
+            //
+            // The calls to `start_l.offset` are valid because there are at most `count-1` of them,
+            // plus the final one at the end of the unsafe block, where `count` is the minimum number
+            // of collected offsets in `offsets_l` and `offsets_r`, so there is no risk of there not
+            // being enough elements. The same reasoning applies to the calls to `start_r.offset`.
+            //
+            // The calls to `copy_nonoverlapping` are safe because `left!` and `right!` are guaranteed
+            // not to overlap, and are valid because of the reasoning above.
             unsafe {
                 let tmp = ptr::read(left!());
                 ptr::copy_nonoverlapping(right!(), left!(), 1);
@@ -366,12 +419,22 @@ where
 
         if start_l == end_l {
             // All out-of-order elements in the left block were moved. Move to the next block.
-            l = unsafe { l.add(block_l) };
+
+            // block-width-guarantee
+            // SAFETY: if `!is_done` then the slice width is guaranteed to be at least `2*BLOCK` wide. There
+            // are at most `BLOCK` elements in `offsets_l` because of its size, so the `offset` operation is
+            // safe. Otherwise, the debug assertions in the `is_done` case guarantee that
+            // `width(l, r) == block_l + block_r`, namely, that the block sizes have been adjusted to account
+            // for the smaller number of remaining elements.
+            l = unsafe { l.offset(block_l as isize) };
         }
 
         if start_r == end_r {
             // All out-of-order elements in the right block were moved. Move to the previous block.
-            r = unsafe { r.sub(block_r) };
+
+            // SAFETY: Same argument as [block-width-guarantee]. Either this is a full block `2*BLOCK`-wide,
+            // or `block_r` has been adjusted for the last handful of elements.
+            r = unsafe { r.offset(-(block_r as isize)) };
         }
 
         if is_done {
@@ -385,9 +448,20 @@ where
 
     if start_l < end_l {
         // The left block remains.
-        // Move it's remaining out-of-order elements to the far right.
+        // Move its remaining out-of-order elements to the far right.
         debug_assert_eq!(width(l, r), block_l);
         while start_l < end_l {
+            // remaining-elements-safety
+            // SAFETY: while the loop condition holds there are still elements in `offsets_l`, so it
+            // is safe to point `end_l` to the previous element.
+            //
+            // The `ptr::swap` is safe if both its arguments are valid for reads and writes:
+            //  - Per the debug assert above, the distance between `l` and `r` is `block_l`
+            //    elements, so there can be at most `block_l` remaining offsets between `start_l`
+            //    and `end_l`. This means `r` will be moved at most `block_l` steps back, which
+            //    makes the `r.offset` calls valid (at that point `l == r`).
+            //  - `offsets_l` contains valid offsets into `v` collected during the partitioning of
+            //    the last block, so the `l.offset` calls are valid.
             unsafe {
                 end_l = end_l.offset(-1);
                 ptr::swap(l.offset(*end_l as isize), r.offset(-1));
@@ -397,9 +471,10 @@ where
         width(v.as_mut_ptr(), r)
     } else if start_r < end_r {
         // The right block remains.
-        // Move it's remaining out-of-order elements to the far left.
+        // Move its remaining out-of-order elements to the far left.
         debug_assert_eq!(width(l, r), block_r);
         while start_r < end_r {
+            // SAFETY: See the reasoning in [remaining-elements-safety].
             unsafe {
                 end_r = end_r.offset(-1);
                 ptr::swap(l, r.offset(-(*end_r as isize) - 1));
@@ -432,17 +507,25 @@ where
 
         // Read the pivot into a stack-allocated variable for efficiency. If a following comparison
         // operation panics, the pivot will be automatically written back into the slice.
-        let write_on_drop = WriteOnDrop {
-            value: unsafe { Some(ptr::read(pivot)) },
+
+        // SAFETY: `pivot` is a reference to the first element of `v`, so `ptr::read` is safe.
+        let tmp = mem::ManuallyDrop::new(unsafe { ptr::read(pivot) });
+        let _pivot_guard = CopyOnDrop {
+            src: &*tmp,
             dest: pivot,
         };
-        let pivot = write_on_drop.value.as_ref().unwrap();
+        let pivot = &*tmp;
 
         // Find the first pair of out-of-order elements.
         let mut l = 0;
         let mut r = v.len();
+
+        // SAFETY: The unsafety below involves indexing an array.
+        // For the first one: We already do the bounds checking here with `l < r`.
+        // For the second one: We initially have `l == 0` and `r == v.len()` and we checked that `l < r` at every indexing operation.
+        //                     From here we know that `r` must be at least `r == l` which was shown to be valid from the first one.
         unsafe {
-            // Find the first element greater then or equal to the pivot.
+            // Find the first element greater than or equal to the pivot.
             while l < r && is_less(v.get_unchecked(l), pivot) {
                 l += 1;
             }
@@ -458,7 +541,7 @@ where
             l >= r,
         )
 
-        // `write_on_drop` goes out of scope and writes the pivot (which is a stack-allocated
+        // `_pivot_guard` goes out of scope and writes the pivot (which is a stack-allocated
         // variable) back into the slice where it originally was. This step is critical in ensuring
         // safety!
     };
@@ -484,18 +567,24 @@ where
 
     // Read the pivot into a stack-allocated variable for efficiency. If a following comparison
     // operation panics, the pivot will be automatically written back into the slice.
-    let write_on_drop = WriteOnDrop {
-        value: unsafe { Some(ptr::read(pivot)) },
+    // SAFETY: The pointer here is valid because it is obtained from a reference to a slice.
+    let tmp = mem::ManuallyDrop::new(unsafe { ptr::read(pivot) });
+    let _pivot_guard = CopyOnDrop {
+        src: &*tmp,
         dest: pivot,
     };
-    let pivot = write_on_drop.value.as_ref().unwrap();
+    let pivot = &*tmp;
 
     // Now partition the slice.
     let mut l = 0;
     let mut r = v.len();
     loop {
+        // SAFETY: The unsafety below involves indexing an array.
+        // For the first one: We already do the bounds checking here with `l < r`.
+        // For the second one: We initially have `l == 0` and `r == v.len()` and we checked that `l < r` at every indexing operation.
+        //                     From here we know that `r` must be at least `r == l` which was shown to be valid from the first one.
         unsafe {
-            // Find the first element greater that the pivot.
+            // Find the first element greater than the pivot.
             while l < r && !is_less(pivot, v.get_unchecked(l)) {
                 l += 1;
             }
@@ -512,7 +601,8 @@ where
 
             // Swap the found pair of out-of-order elements.
             r -= 1;
-            ptr::swap(v.get_unchecked_mut(l), v.get_unchecked_mut(r));
+            let ptr = v.as_mut_ptr();
+            ptr::swap(ptr.add(l), ptr.add(r));
             l += 1;
         }
     }
@@ -520,7 +610,7 @@ where
     // We found `l` elements equal to the pivot. Add 1 to account for the pivot itself.
     l + 1
 
-    // `write_on_drop` goes out of scope and writes the pivot (which is a stack-allocated variable)
+    // `_pivot_guard` goes out of scope and writes the pivot (which is a stack-allocated variable)
     // back into the slice where it originally was. This step is critical in ensuring safety!
 }
 
@@ -539,10 +629,11 @@ fn break_patterns<T>(v: &mut [T]) {
             random
         };
         let mut gen_usize = || {
+            // TODO 1.53: if usize::BITS <= 32 {
             if mem::size_of::<usize>() <= 4 {
                 gen_u32() as usize
             } else {
-                ((u64::from(gen_u32()) << 32) | u64::from(gen_u32())) as usize
+                (((gen_u32() as u64) << 32) | (gen_u32() as u64)) as usize
             }
         };
 
@@ -594,6 +685,12 @@ where
 
     if len >= 8 {
         // Swaps indices so that `v[a] <= v[b]`.
+        // SAFETY: `len >= 8` so there are at least two elements in the neighborhoods of
+        // `a`, `b` and `c`. This means the three calls to `sort_adjacent` result in
+        // corresponding calls to `sort3` with valid 3-item neighborhoods around each
+        // pointer, which in turn means the calls to `sort2` are done with valid
+        // references. Thus the `v.get_unchecked` calls are safe, as is the `ptr::swap`
+        // call.
         let mut sort2 = |a: &mut usize, b: &mut usize| unsafe {
             if is_less(v.get_unchecked(*b), v.get_unchecked(*a)) {
                 ptr::swap(a, b);
@@ -641,7 +738,7 @@ where
 ///
 /// `limit` is the number of allowed imbalanced partitions before switching to `heapsort`. If zero,
 /// this function will immediately switch to heapsort.
-fn recurse<'a, T, F>(mut v: &'a mut [T], is_less: &F, mut pred: Option<&'a mut T>, mut limit: usize)
+fn recurse<'a, T, F>(mut v: &'a mut [T], is_less: &F, mut pred: Option<&'a mut T>, mut limit: u32)
 where
     T: Send,
     F: Fn(&T, &T) -> bool + Sync,
@@ -667,7 +764,7 @@ where
         }
 
         // If too many bad pivot choices were made, simply fall back to heapsort in order to
-        // guarantee `O(n log n)` worst-case.
+        // guarantee `O(n * log(n))` worst-case.
         if limit == 0 {
             heapsort(v, is_less);
             return;
@@ -701,7 +798,7 @@ where
                 let mid = partition_equal(v, pivot, is_less);
 
                 // Continue sorting elements greater than the pivot.
-                v = &mut { v }[mid..];
+                v = &mut v[mid..];
                 continue;
             }
         }
@@ -712,7 +809,7 @@ where
         was_partitioned = was_p;
 
         // Split the slice into `left`, `pivot`, and `right`.
-        let (left, right) = { v }.split_at_mut(mid);
+        let (left, right) = v.split_at_mut(mid);
         let (pivot, right) = right.split_at_mut(1);
         let pivot = &mut pivot[0];
 
@@ -741,7 +838,7 @@ where
 
 /// Sorts `v` using pattern-defeating quicksort in parallel.
 ///
-/// The algorithm is unstable, in-place, and `O(n log n)` worst-case.
+/// The algorithm is unstable, in-place, and *O*(*n* \* log(*n*)) worst-case.
 pub(super) fn par_quicksort<T, F>(v: &mut [T], is_less: F)
 where
     T: Send,
@@ -753,7 +850,8 @@ where
     }
 
     // Limit the number of imbalanced partitions to `floor(log2(len)) + 1`.
-    let limit = mem::size_of::<usize>() * 8 - v.len().leading_zeros() as usize;
+    // TODO 1.53: let limit = usize::BITS - v.len().leading_zeros();
+    let limit = mem::size_of::<usize>() as u32 * 8 - v.len().leading_zeros();
 
     recurse(v, &is_less, None, limit);
 }
diff --git a/src/vec.rs b/src/vec.rs
index 1e68aa0..e079639 100644
--- a/src/vec.rs
+++ b/src/vec.rs
@@ -132,22 +132,16 @@ impl<'data, T: Send> IndexedParallelIterator for Drain<'data, T> {
         self.range.len()
     }
 
-    fn with_producer<CB>(self, callback: CB) -> CB::Output
+    fn with_producer<CB>(mut self, callback: CB) -> CB::Output
     where
         CB: ProducerCallback<Self::Item>,
     {
         unsafe {
             // Make the vector forget about the drained items, and temporarily the tail too.
-            let start = self.range.start;
-            self.vec.set_len(start);
+            self.vec.set_len(self.range.start);
 
             // Create the producer as the exclusive "owner" of the slice.
-            let producer = {
-                // Get a correct borrow lifetime, then extend it to the original length.
-                let mut slice = &mut self.vec[start..];
-                slice = slice::from_raw_parts_mut(slice.as_mut_ptr(), self.range.len());
-                DrainProducer::new(slice)
-            };
+            let producer = DrainProducer::from_vec(&mut self.vec, self.range.len());
 
             // The producer will move or drop each item from the drained range.
             callback.callback(producer)
@@ -184,13 +178,27 @@ pub(crate) struct DrainProducer<'data, T: Send> {
     slice: &'data mut [T],
 }
 
-impl<'data, T: 'data + Send> DrainProducer<'data, T> {
+impl<T: Send> DrainProducer<'_, T> {
     /// Creates a draining producer, which *moves* items from the slice.
     ///
     /// Unsafe bacause `!Copy` data must not be read after the borrow is released.
-    pub(crate) unsafe fn new(slice: &'data mut [T]) -> Self {
+    pub(crate) unsafe fn new(slice: &mut [T]) -> DrainProducer<'_, T> {
         DrainProducer { slice }
     }
+
+    /// Creates a draining producer, which *moves* items from the tail of the vector.
+    ///
+    /// Unsafe because we're moving from beyond `vec.len()`, so the caller must ensure
+    /// that data is initialized and not read after the borrow is released.
+    unsafe fn from_vec(vec: &mut Vec<T>, len: usize) -> DrainProducer<'_, T> {
+        let start = vec.len();
+        assert!(vec.capacity() - start >= len);
+
+        // The pointer is derived from `Vec` directly, not through a `Deref`,
+        // so it has provenance over the whole allocation.
+        let ptr = vec.as_mut_ptr().add(start);
+        DrainProducer::new(slice::from_raw_parts_mut(ptr, len))
+    }
 }
 
 impl<'data, T: 'data + Send> Producer for DrainProducer<'data, T> {


Test patch:
diff --git a/src/iter/collect/test.rs b/src/iter/collect/test.rs
index ddf7757..b5f676f 100644
--- a/src/iter/collect/test.rs
+++ b/src/iter/collect/test.rs
@@ -5,7 +5,7 @@
 // try to drive the "collect consumer" incorrectly. These should
 // result in panics.
 
-use super::Collect;
+use super::collect_with_consumer;
 use crate::iter::plumbing::*;
 use rayon_core::join;
 
@@ -20,7 +20,7 @@ use std::thread::Result as ThreadResult;
 #[should_panic(expected = "too many values")]
 fn produce_too_many_items() {
     let mut v = vec![];
-    Collect::new(&mut v, 2).with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 2, |consumer| {
         let mut folder = consumer.into_folder();
         folder = folder.consume(22);
         folder = folder.consume(23);
@@ -35,8 +35,7 @@ fn produce_too_many_items() {
 #[should_panic(expected = "expected 5 total writes, but got 2")]
 fn produce_fewer_items() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 5);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 5, |consumer| {
         let mut folder = consumer.into_folder();
         folder = folder.consume(22);
         folder = folder.consume(23);
@@ -49,8 +48,7 @@ fn produce_fewer_items() {
 #[should_panic(expected = "expected 4 total writes, but got 2")]
 fn left_produces_items_with_no_complete() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 4);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 4, |consumer| {
         let (left_consumer, right_consumer, _) = consumer.split_at(2);
         let mut left_folder = left_consumer.into_folder();
         let mut right_folder = right_consumer.into_folder();
@@ -66,8 +64,7 @@ fn left_produces_items_with_no_complete() {
 #[should_panic(expected = "expected 4 total writes, but got 2")]
 fn right_produces_items_with_no_complete() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 4);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 4, |consumer| {
         let (left_consumer, right_consumer, _) = consumer.split_at(2);
         let mut left_folder = left_consumer.into_folder();
         let mut right_folder = right_consumer.into_folder();
@@ -83,8 +80,7 @@ fn produces_items_with_no_complete() {
     let counter = DropCounter::default();
     let mut v = vec![];
     let panic_result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
-        let collect = Collect::new(&mut v, 2);
-        collect.with_consumer(|consumer| {
+        collect_with_consumer(&mut v, 2, |consumer| {
             let mut folder = consumer.into_folder();
             folder = folder.consume(counter.element());
             folder = folder.consume(counter.element());
@@ -102,8 +98,7 @@ fn produces_items_with_no_complete() {
 #[should_panic(expected = "too many values")]
 fn left_produces_too_many_items() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 4);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 4, |consumer| {
         let (left_consumer, right_consumer, _) = consumer.split_at(2);
         let mut left_folder = left_consumer.into_folder();
         let mut right_folder = right_consumer.into_folder();
@@ -120,8 +115,7 @@ fn left_produces_too_many_items() {
 #[should_panic(expected = "too many values")]
 fn right_produces_too_many_items() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 4);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 4, |consumer| {
         let (left_consumer, right_consumer, _) = consumer.split_at(2);
         let mut left_folder = left_consumer.into_folder();
         let mut right_folder = right_consumer.into_folder();
@@ -138,8 +132,7 @@ fn right_produces_too_many_items() {
 #[should_panic(expected = "expected 4 total writes, but got 1")]
 fn left_produces_fewer_items() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 4);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 4, |consumer| {
         let reducer = consumer.to_reducer();
         let (left_consumer, right_consumer, _) = consumer.split_at(2);
         let mut left_folder = left_consumer.into_folder();
@@ -158,8 +151,7 @@ fn left_produces_fewer_items() {
 #[should_panic(expected = "expected 4 total writes, but got 2")]
 fn only_left_result() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 4);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 4, |consumer| {
         let (left_consumer, right_consumer, _) = consumer.split_at(2);
         let mut left_folder = left_consumer.into_folder();
         let mut right_folder = right_consumer.into_folder();
@@ -177,8 +169,7 @@ fn only_left_result() {
 #[should_panic(expected = "expected 4 total writes, but got 2")]
 fn only_right_result() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 4);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 4, |consumer| {
         let (left_consumer, right_consumer, _) = consumer.split_at(2);
         let mut left_folder = left_consumer.into_folder();
         let mut right_folder = right_consumer.into_folder();
@@ -195,8 +186,7 @@ fn only_right_result() {
 #[should_panic(expected = "expected 4 total writes, but got 2")]
 fn reducer_does_not_preserve_order() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 4);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 4, |consumer| {
         let reducer = consumer.to_reducer();
         let (left_consumer, right_consumer, _) = consumer.split_at(2);
         let mut left_folder = left_consumer.into_folder();
@@ -215,8 +205,7 @@ fn reducer_does_not_preserve_order() {
 #[should_panic(expected = "expected 4 total writes, but got 3")]
 fn right_produces_fewer_items() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 4);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 4, |consumer| {
         let reducer = consumer.to_reducer();
         let (left_consumer, right_consumer, _) = consumer.split_at(2);
         let mut left_folder = left_consumer.into_folder();
@@ -230,13 +219,12 @@ fn right_produces_fewer_items() {
 }
 
 // The left consumer panics and the right stops short, like `panic_fuse()`.
-// We should get the left panic without finishing `Collect::with_consumer`.
+// We should get the left panic without finishing `collect_with_consumer`.
 #[test]
 #[should_panic(expected = "left consumer panic")]
 fn left_panics() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 4);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 4, |consumer| {
         let reducer = consumer.to_reducer();
         let (left_consumer, right_consumer, _) = consumer.split_at(2);
         let (left_result, right_result) = join(
@@ -257,13 +245,12 @@ fn left_panics() {
 }
 
 // The right consumer panics and the left stops short, like `panic_fuse()`.
-// We should get the right panic without finishing `Collect::with_consumer`.
+// We should get the right panic without finishing `collect_with_consumer`.
 #[test]
 #[should_panic(expected = "right consumer panic")]
 fn right_panics() {
     let mut v = vec![];
-    let collect = Collect::new(&mut v, 4);
-    collect.with_consumer(|consumer| {
+    collect_with_consumer(&mut v, 4, |consumer| {
         let reducer = consumer.to_reducer();
         let (left_consumer, right_consumer, _) = consumer.split_at(2);
         let (left_result, right_result) = join(
@@ -290,8 +277,7 @@ fn left_produces_fewer_items_drops() {
     let counter = DropCounter::default();
     let mut v = vec![];
     let panic_result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
-        let collect = Collect::new(&mut v, 4);
-        collect.with_consumer(|consumer| {
+        collect_with_consumer(&mut v, 4, |consumer| {
             let reducer = consumer.to_reducer();
             let (left_consumer, right_consumer, _) = consumer.split_at(2);
             let mut left_folder = left_consumer.into_folder();