Fix pinned drop violations; make types Unpin where practical

* FutureGroup and StreamGroup are now `Unpin` types. At heart,
  they kinda always were
* ChunkedVec is now built around pinning, fulfilling its destiny as
  a data structure made specifically for pinnable data
* Brief changes to tests, to catch the bug that spawned this commit

Properly speaking, neither FutureGroup nor StreamGroup need to be
pinned. The futures/streams are placed in allocations managed by
ChunkedVec, which is responsible for keeping the futures/streams
pinned. This has been true. In fact, this means the entire
pin_project/`#[pin]` management done by these types was superfluous.
FutureGroup and StreamGroup don't store any pinnable data inline.

Going forward, these types can be safely exposed as Unpin. Even if
their implementation changes again, FutureGroup and StreamGroup are
never going to store their streams/futures on stack. Only inline
storage makes pinning self-referential data meaningful; heap
storage will always be stable. Thus, making them Unpin is totally
safe from the perspective of future compatibility.

The same goes for derived types. As long as the futures/streams
are polled on the heap, there's no way they will need to be pinned
in the old-fashioned (inline) way.

Author: A248

src/concurrent_stream/for_each.rs

@@ -12,7 +12,6 @@ use core::sync::atomic::{AtomicUsize, Ordering};
 use core::task::{ready, Context, Poll};
 
 // OK: validated! - all bounds should check out
-#[pin_project]
 pub(crate) struct ForEachConsumer<FutT, T, F, FutB>
 where
     FutT: Future<Output = T>,
@@ -21,13 +20,20 @@ where
 {
     // NOTE: we can remove the `Arc` here if we're willing to make this struct self-referential
     count: Arc<AtomicUsize>,
-    #[pin]
     group: FutureGroup<ForEachFut<F, FutT, T, FutB>>,
     limit: usize,
     f: F,
     _phantom: PhantomData<(T, FutB)>,
 }
 
+impl<FutT, T, F, FutB> Unpin for ForEachConsumer<FutT, T, F, FutB>
+where
+    FutT: Future<Output = T>,
+    F: Fn(T) -> FutB,
+    FutB: Future<Output = ()>,
+{
+}
+
 impl<A, T, F, B> ForEachConsumer<A, T, F, B>
 where
     A: Future<Output = T>,
@@ -60,32 +66,30 @@ where
     type Output = ();
 
     async fn send(self: Pin<&mut Self>, future: FutT) -> super::ConsumerState {
-        let mut this = self.project();
+        let mut this = self.get_mut();
         // If we have no space, we're going to provide backpressure until we have space
-        while this.count.load(Ordering::Relaxed) >= *this.limit {
+        while this.count.load(Ordering::Relaxed) >= this.limit {
             this.group.next().await;
         }
 
         // Space was available! - insert the item for posterity
         this.count.fetch_add(1, Ordering::Relaxed);
         let fut = ForEachFut::new(this.f.clone(), future, this.count.clone());
-        this.group.as_mut().insert_pinned(fut);
+        this.group.insert(fut);
 
         ConsumerState::Continue
     }
 
-    async fn progress(self: Pin<&mut Self>) -> super::ConsumerState {
-        let mut this = self.project();
-        while (this.group.next().await).is_some() {}
+    async fn progress(mut self: Pin<&mut Self>) -> super::ConsumerState {
+        while self.group.next().await.is_some() {}
         ConsumerState::Empty
     }
 
-    async fn flush(self: Pin<&mut Self>) -> Self::Output {
-        let mut this = self.project();
+    async fn flush(mut self: Pin<&mut Self>) -> Self::Output {
         // 4. We will no longer receive any additional futures from the
         // underlying stream; wait until all the futures in the group have
         // resolved.
-        while (this.group.next().await).is_some() {}
+        while self.group.next().await.is_some() {}
     }
 }
 

src/concurrent_stream/from_concurrent_stream.rs

@@ -41,9 +41,7 @@ impl<T, E> FromConcurrentStream<Result<T, E>> for Result<Vec<T>, E> {
 }
 
 // TODO: replace this with a generalized `fold` operation
-#[pin_project]
 pub(crate) struct VecConsumer<'a, Fut: Future> {
-    #[pin]
     group: FutureGroup<Fut>,
     output: &'a mut Vec<Fut::Output>,
 }
@@ -63,31 +61,28 @@ where
 {
     type Output = ();
 
-    async fn send(self: Pin<&mut Self>, future: Fut) -> super::ConsumerState {
-        let mut this = self.project();
+    async fn send(mut self: Pin<&mut Self>, future: Fut) -> super::ConsumerState {
         // unbounded concurrency, so we just goooo
-        this.group.as_mut().insert_pinned(future);
+        self.group.insert(future);
         ConsumerState::Continue
     }
 
     async fn progress(self: Pin<&mut Self>) -> super::ConsumerState {
-        let mut this = self.project();
+        let this = self.get_mut();
         while let Some(item) = this.group.next().await {
             this.output.push(item);
         }
         ConsumerState::Empty
     }
     async fn flush(self: Pin<&mut Self>) -> Self::Output {
-        let mut this = self.project();
+        let this = self.get_mut();
         while let Some(item) = this.group.next().await {
             this.output.push(item);
         }
     }
 }
 
-#[pin_project]
 pub(crate) struct ResultVecConsumer<'a, Fut: Future, T, E> {
-    #[pin]
     group: FutureGroup<Fut>,
     output: &'a mut Result<Vec<T>, E>,
 }
@@ -107,15 +102,14 @@ where
 {
     type Output = ();
 
-    async fn send(self: Pin<&mut Self>, future: Fut) -> super::ConsumerState {
-        let mut this = self.project();
+    async fn send(mut self: Pin<&mut Self>, future: Fut) -> super::ConsumerState {
         // unbounded concurrency, so we just goooo
-        this.group.as_mut().insert_pinned(future);
+        self.group.insert(future);
         ConsumerState::Continue
     }
 
     async fn progress(self: Pin<&mut Self>) -> super::ConsumerState {
-        let mut this = self.project();
+        let this = self.get_mut();
         let Ok(items) = this.output else {
             return ConsumerState::Break;
         };
@@ -126,7 +120,7 @@ where
                     items.push(item);
                 }
                 Err(e) => {
-                    **this.output = Err(e);
+                    *this.output = Err(e);
                     return ConsumerState::Break;
                 }
             }

src/concurrent_stream/try_for_each.rs

@@ -2,7 +2,6 @@ use crate::concurrent_stream::ConsumerState;
 use crate::future::FutureGroup;
 use crate::private::Try;
 use futures_lite::StreamExt;
-use pin_project::pin_project;
 
 use super::Consumer;
 use alloc::sync::Arc;
@@ -14,7 +13,6 @@ use core::pin::Pin;
 use core::sync::atomic::{AtomicUsize, Ordering};
 use core::task::{ready, Context, Poll};
 
-#[pin_project]
 pub(crate) struct TryForEachConsumer<FutT, T, F, FutB, B>
 where
     FutT: Future<Output = T>,
@@ -24,14 +22,22 @@ where
 {
     // NOTE: we can remove the `Arc` here if we're willing to make this struct self-referential
     count: Arc<AtomicUsize>,
-    #[pin]
     group: FutureGroup<TryForEachFut<F, FutT, T, FutB, B>>,
     limit: usize,
     residual: Option<B::Residual>,
     f: F,
     _phantom: PhantomData<(T, FutB)>,
 }
 
+impl<FutT, T, F, FutB, B> Unpin for TryForEachConsumer<FutT, T, F, FutB, B>
+where
+    FutT: Future<Output = T>,
+    F: Clone + Fn(T) -> FutB,
+    FutB: Future<Output = B>,
+    B: Try<Output = ()>,
+{
+}
+
 impl<FutT, T, F, FutB, B> TryForEachConsumer<FutT, T, F, FutB, B>
 where
     FutT: Future<Output = T>,
@@ -66,9 +72,9 @@ where
     type Output = B;
 
     async fn send(self: Pin<&mut Self>, future: FutT) -> super::ConsumerState {
-        let mut this = self.project();
+        let mut this = self.get_mut();
         // If we have no space, we're going to provide backpressure until we have space
-        while this.count.load(Ordering::Relaxed) >= *this.limit {
+        while this.count.load(Ordering::Relaxed) >= this.limit {
             match this.group.next().await {
                 // Case 1: there are no more items available in the group. We
                 // can no longer iterate over them, and necessarily should be
@@ -82,7 +88,7 @@ where
                     // entirely so we can short-circuit with an error from the
                     // `flush` method.
                     ControlFlow::Break(residual) => {
-                        *this.residual = Some(residual);
+                        this.residual = Some(residual);
                         return ConsumerState::Break;
                     }
                 },
@@ -92,23 +98,23 @@ where
         // Space was available! - insert the item for posterity
         this.count.fetch_add(1, Ordering::Relaxed);
         let fut = TryForEachFut::new(this.f.clone(), future, this.count.clone());
-        this.group.as_mut().insert_pinned(fut);
+        this.group.insert(fut);
         ConsumerState::Continue
     }
 
     async fn progress(self: Pin<&mut Self>) -> super::ConsumerState {
-        let mut this = self.project();
+        let mut this = self.get_mut();
         while let Some(res) = this.group.next().await {
             if let ControlFlow::Break(residual) = res.branch() {
-                *this.residual = Some(residual);
+                this.residual = Some(residual);
                 return ConsumerState::Break;
             }
         }
         ConsumerState::Empty
     }
 
     async fn flush(self: Pin<&mut Self>) -> Self::Output {
-        let mut this = self.project();
+        let mut this = self.get_mut();
         // Return the error if we stopped iteration because of a previous error.
         if this.residual.is_some() {
             return B::from_residual(this.residual.take().unwrap());

src/future/future_group.rs

@@ -57,9 +57,7 @@ use crate::utils::{ChunkedVec, PollState, PollVec, WakerVec};
 /// # });}
 /// ```
 #[must_use = "`FutureGroup` does nothing if not iterated over"]
-#[pin_project::pin_project]
 pub struct FutureGroup<F> {
-    #[pin]
     futures: ChunkedVec<F>,
     wakers: WakerVec,
     states: PollVec,
@@ -189,7 +187,7 @@ impl<F> FutureGroup<F> {
         let is_present = self.keys.remove(&key.0);
         if is_present {
             self.states[key.0].set_none();
-            self.futures.remove(key.0);
+            self.futures.remove_in_place(key.0);
         }
         is_present
     }
@@ -272,43 +270,6 @@ impl<F: Future> FutureGroup<F> {
         Key(index)
     }
 
-    /// Insert a value into a pinned `FutureGroup`
-    ///
-    /// This method is private because it serves as an implementation detail for
-    /// `ConcurrentStream`. We should never expose this publicly, as the entire
-    /// point of this crate is that we abstract the futures poll machinery away
-    /// from end-users.
-    ///
-    /// # Safety
-    ///
-    /// This is safe because `ChunkedVec` uses a triangular allocation
-    /// strategy that never moves existing elements when growing. Each bucket
-    /// is a heap-allocated box that remains at a stable address.
-    #[allow(unused)]
-    pub(crate) fn insert_pinned(self: Pin<&mut Self>, future: F) -> Key
-    where
-        F: Future,
-    {
-        let mut this = self.project();
-        // SAFETY: ChunkedVec guarantees that inserting a value never moves
-        // existing values. Growth allocates new buckets without touching existing ones.
-        let index = unsafe { this.futures.as_mut().get_unchecked_mut() }.insert(future);
-        this.keys.insert(index);
-        let key = Key(index);
-
-        // Update tracking structures to match the new capacity
-        let new_cap = this.futures.as_ref().capacity();
-        this.wakers.resize(new_cap);
-        this.states.resize(new_cap);
-
-        // Set the corresponding state
-        this.states[index].set_pending();
-        let mut readiness = this.wakers.readiness();
-        readiness.set_ready(index);
-
-        key
-    }
-
     /// Create a stream which also yields the key of each item.
     ///
     /// # Example
@@ -337,19 +298,14 @@ impl<F: Future> FutureGroup<F> {
 }
 
 impl<F: Future> FutureGroup<F> {
-    fn poll_next_inner(
-        self: Pin<&mut Self>,
-        cx: &Context<'_>,
-    ) -> Poll<Option<(Key, <F as Future>::Output)>> {
-        let mut this = self.project();
-
+    fn poll_next_inner(&mut self, cx: &Context<'_>) -> Poll<Option<(Key, <F as Future>::Output)>> {
         // Short-circuit if we have no futures to iterate over
-        if this.futures.is_empty() {
+        if self.futures.is_empty() {
             return Poll::Ready(None);
         }
 
         // Set the top-level waker and check readiness
-        let mut readiness = this.wakers.readiness();
+        let mut readiness = self.wakers.readiness();
         readiness.set_waker(cx.waker());
         if !readiness.any_ready() {
             // Nothing is ready yet
@@ -358,24 +314,21 @@ impl<F: Future> FutureGroup<F> {
 
         // Setup our futures state
         let mut ret = Poll::Pending;
-        let states = this.states;
-
-        // SAFETY: We unpin the future group so we can later individually access
-        // single futures. Either to read from them or to drop them.
-        let futures = unsafe { this.futures.as_mut().get_unchecked_mut() };
+        let states = &mut self.states;
+        let futures = &mut self.futures;
 
-        for index in this.keys.iter().cloned() {
+        for index in self.keys.iter().cloned() {
             if states[index].is_pending() && readiness.clear_ready(index) {
                 // unlock readiness so we don't deadlock when polling
                 #[allow(clippy::drop_non_drop)]
                 drop(readiness);
 
                 // Obtain the intermediate waker.
-                let mut cx = Context::from_waker(this.wakers.get(index).unwrap());
+                let mut cx = Context::from_waker(self.wakers.get(index).unwrap());
 
-                // SAFETY: this future here is a projection from the futures
+                // SAFETY: self future here is a projection from the futures
                 // vec, which we're reading from.
-                let future = unsafe { Pin::new_unchecked(&mut futures[index]) };
+                let future = futures.get_mut(index).expect("index ready but not init?");
                 match future.poll(&mut cx) {
                     Poll::Ready(item) => {
                         // Set the return type for the function
@@ -384,7 +337,7 @@ impl<F: Future> FutureGroup<F> {
                         // Remove all associated data with the future
                         // The only data we can't remove directly is the key entry.
                         states[index] = PollState::None;
-                        futures.remove(index);
+                        futures.remove_in_place(index);
 
                         break;
                     }
@@ -393,14 +346,14 @@ impl<F: Future> FutureGroup<F> {
                 };
 
                 // Lock readiness so we can use it again
-                readiness = this.wakers.readiness();
+                readiness = self.wakers.readiness();
             }
         }
 
-        // Now that we're no longer borrowing `this.keys` we can remove
+        // Now that we're no longer borrowing `self.keys` we can remove
         // the current key from the set
         if let Poll::Ready(Some((key, _))) = ret {
-            this.keys.remove(&key.0);
+            self.keys.remove(&key.0);
         }
 
         ret
@@ -410,7 +363,7 @@ impl<F: Future> FutureGroup<F> {
 impl<F: Future> Stream for FutureGroup<F> {
     type Item = <F as Future>::Output;
 
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         match self.poll_next_inner(cx) {
             Poll::Ready(Some((_key, item))) => Poll::Ready(Some(item)),
             Poll::Ready(None) => Poll::Ready(None),