fix unsoundness due to not releasing task during mapped_futures cancel()

Author: StoicDeveloper

futures-util/src/stream/futures_unordered_internal/mod.rs

@@ -253,7 +253,7 @@ impl<K, Fut, S: ReleasesTask<K>> FuturesUnorderedInternal<K, Fut, S> {
     /// Releases the task. It destroys the future inside and either drops
     /// the `Arc<Task>` or transfers ownership to the ready to run queue.
     /// The task this method is called on must have been unlinked before.
-    fn release_task(&mut self, task: Arc<Task<K, Fut>>) {
+    pub(super) fn release_task(&mut self, task: Arc<Task<K, Fut>>) {
         if let Some(key) = task.key() {
             self.inner.release_task(key);
         }

futures-util/src/stream/mapped_futures/mod.rs

@@ -167,8 +167,8 @@ impl<K: Hash + Eq, Fut> MappedFutures<K, Fut> {
         if let Some(task) = self.set().take(key) {
             unsafe {
                 if (*task.future.get()).is_some() {
-                    *task.future.get() = None;
                     self.inner.unlink(Arc::as_ptr(&task.inner));
+                    self.inner.release_task(task.inner.clone());
                     return true;
                 }
             }
@@ -186,6 +186,7 @@ impl<K: Hash + Eq, Fut> MappedFutures<K, Fut> {
             unsafe {
                 let fut = (*task.future.get()).take().unwrap();
                 self.inner.unlink(Arc::as_ptr(&task.inner));
+                self.inner.release_task(task.inner.clone());
                 return Some(fut);
             }
         }
@@ -311,7 +312,7 @@ impl<K: Hash + Eq, Fut: Future> Stream for MappedFutures<K, Fut> {
                 // let key = task.take_key();
                 // MappedFutures::set(self.get_mut()).remove(&key);
                 MappedFutures::set(self.get_mut()).remove(task.key().unwrap());
-                return Poll::Ready(Some((task.take_key(), output)));
+                Poll::Ready(Some((task.take_key(), output)))
             }
             None => Poll::Ready(None),
         }
@@ -450,13 +451,13 @@ pub mod tests {
     #[test]
     fn mutate() {
         let mut futures: MappedFutures<u32, Delay> = MappedFutures::new();
-        insert_millis(&mut futures, 1, 50);
-        insert_millis(&mut futures, 2, 100);
-        insert_millis(&mut futures, 3, 150);
-        insert_millis(&mut futures, 4, 200);
+        insert_millis(&mut futures, 1, 500);
+        insert_millis(&mut futures, 2, 1000);
+        insert_millis(&mut futures, 3, 1500);
+        insert_millis(&mut futures, 4, 2000);
 
         assert_eq!(block_on(futures.next()).unwrap().0, 1);
-        futures.get_mut(&3).unwrap().reset(Duration::from_millis(30));
+        futures.get_mut(&3).unwrap().reset(Duration::from_millis(300));
         assert_eq!(block_on(futures.next()).unwrap().0, 3);
         assert_eq!(block_on(futures.next()).unwrap().0, 2);
         assert_eq!(block_on(futures.next()).unwrap().0, 4);