chore: Clean up code a little

- Use an iterator when processing many tasks instead of a for loop.
- Add a test for spawning large numbers of tasks at a time.
- Fix Clippy warnings

Signed-off-by: John Nunley <[email protected]>

Author: notgull

src/lib.rs

@@ -199,21 +199,27 @@ impl<'a> Executor<'a> {
         futures: impl IntoIterator<Item = F>,
         handles: &mut impl Extend<Task<F::Output>>,
     ) {
-        let mut active = self.state().active.lock().unwrap();
+        let mut active = Some(self.state().active.lock().unwrap());
 
-        for (i, future) in futures.into_iter().enumerate() {
+        // Convert the futures into tasks.
+        let tasks = futures.into_iter().enumerate().map(move |(i, future)| {
             // SAFETY: `T` and the future are `Send`.
-            handles.extend(Some(unsafe { self.spawn_inner(future, &mut active) }));
+            let task = unsafe { self.spawn_inner(future, active.as_mut().unwrap()) };
 
             // Yield the lock every once in a while to ease contention.
             if i.wrapping_sub(1) % 500 == 0 {
-                drop(active);
-                active = self.state().active.lock().unwrap();
+                drop(active.take());
+                active = Some(self.state().active.lock().unwrap());
             }
-        }
+
+            task
+        });
+
+        // Push the tasks to the user's collection.
+        handles.extend(tasks);
     }
 
-    /// Spawn a future using the inner lock.
+    /// Spawn a future while holding the inner lock.
     ///
     /// # Safety
     ///
@@ -527,16 +533,14 @@ impl<'a> LocalExecutor<'a> {
         let tasks = futures.into_iter().map(|future| {
             // SAFETY: This executor is not thread safe, so the future and its result
             //         cannot be sent to another thread.
-            unsafe {
-                self.inner().spawn_inner(future, &mut active)
-            }
+            unsafe { self.inner().spawn_inner(future, &mut active) }
 
             // As only one thread can spawn or poll tasks at a time, there is no need
             // to release lock contention here.
         });
 
         // Push them to the user's collection.
-        handles.extend(tasks); 
+        handles.extend(tasks);
     }
 
     /// Attempts to run a task if at least one is scheduled.
@@ -602,16 +606,6 @@ impl<'a> LocalExecutor<'a> {
         self.inner().run(future).await
     }
 
-    /// Returns a function that schedules a runnable task when it gets woken up.
-    fn schedule(&self) -> impl Fn(Runnable) + Send + Sync + 'static {
-        let state = self.inner().state().clone();
-
-        move |runnable| {
-            state.queue.push(runnable).unwrap();
-            state.notify();
-        }
-    }
-
     /// Returns a reference to the inner executor.
     fn inner(&self) -> &Executor<'a> {
         &self.inner
@@ -1060,17 +1054,6 @@ fn debug_executor(executor: &Executor<'_>, name: &str, f: &mut fmt::Formatter<'_
         .finish()
 }
 
-/// Container with one item.
-///
-/// This implements `Extend` for one-off cases.
-struct Container<T>(Option<T>);
-
-impl<T> Extend<T> for Container<T> {
-    fn extend<X: IntoIterator<Item = T>>(&mut self, iter: X) {
-        self.0 = iter.into_iter().next();
-    }
-}
-
 /// Runs a closure when dropped.
 struct CallOnDrop<F: FnMut()>(F);
 

tests/spawn_many.rs

@@ -0,0 +1,40 @@
+use async_executor::{Executor, LocalExecutor};
+use futures_lite::future;
+
+#[test]
+fn spawn_many() {
+    future::block_on(async {
+        let ex = Executor::new();
+
+        // Spawn a lot of tasks.
+        let mut tasks = vec![];
+        ex.spawn_many((0..50_000).map(|i| future::ready(i)), &mut tasks);
+
+        // Run all of the tasks in parallel.
+        ex.run(async move {
+            for (i, task) in tasks.into_iter().enumerate() {
+                assert_eq!(task.await, i);
+            }
+        })
+        .await;
+    });
+}
+
+#[test]
+fn spawn_many_local() {
+    future::block_on(async {
+        let ex = LocalExecutor::new();
+
+        // Spawn a lot of tasks.
+        let mut tasks = vec![];
+        ex.spawn_many((0..50_000).map(|i| future::ready(i)), &mut tasks);
+
+        // Run all of the tasks in parallel.
+        ex.run(async move {
+            for (i, task) in tasks.into_iter().enumerate() {
+                assert_eq!(task.await, i);
+            }
+        })
+        .await;
+    });
+}