perf: avoid boxing provider futures for the default runtime

src/solver/encoding.rs

@@ -19,6 +19,14 @@ use indexmap::IndexMap;
 
 type PendingTask<'cache, D> = LocalBoxFuture<'cache, Result<TaskResult<'cache, D>, Box<dyn Any>>>;
 
+/// Selects how provider futures are queued. This is deliberately an internal solver detail rather
+/// than a capability exposed by [`crate::runtime::AsyncRuntime`].
+#[derive(Copy, Clone)]
+pub(super) enum FutureQueueMode {
+    Immediate,
+    PendingCapable,
+}
+
 type RequirementCondition<'a, S> = Option<(ConditionId, Vec<Vec<DisjunctionComplement<'a, S>>>)>;
 
 /// Fetches each version set's sorted candidates while avoiding `try_join_all`'s
@@ -86,9 +94,10 @@ pub(crate) struct Encoder<'a, 'cache, D: DependencyProvider> {
     /// A set of packages that should have an at-least-once tracker.
     new_at_least_one_packages: IndexMap<D::NameId, VariableId, ahash::RandomState>,
 
-    /// Results from futures that completed immediately during
-    /// `try_immediate_or_queue`.
+    /// Results from futures that completed immediately during `queue_future`.
     pending_results: VecDeque<Result<TaskResult<'cache, D>, Box<dyn Any>>>,
+
+    future_queue_mode: FutureQueueMode,
 }
 
 /// The result of a future that was queued for processing.
@@ -185,6 +194,7 @@ impl<'a, 'cache, D: DependencyProvider> Encoder<'a, 'cache, D> {
         cache: &'cache SolverCache<D>,
         root_dependencies: &'cache Dependencies,
         level: u32,
+        future_queue_mode: FutureQueueMode,
     ) -> Self {
         Self {
             state,
@@ -197,28 +207,34 @@ impl<'a, 'cache, D: DependencyProvider> Encoder<'a, 'cache, D> {
             level,
             new_at_least_one_packages: IndexMap::default(),
             pending_results: VecDeque::new(),
+            future_queue_mode,
         }
     }
 
-    /// Poll `future` once on the stack. If it resolves immediately (as all
-    /// futures do under [`NowOrNeverRuntime`]), record the result for
-    /// iterative processing; otherwise hand the boxed future off to
-    /// [`Self::pending_futures`] for later polling.
+    /// Queue a provider future while avoiding allocation for the default non-yielding runtime.
     ///
-    /// We still box the future, so the allocation cost is the same. The
-    /// win over pushing straight to `FuturesUnordered` is avoiding its slab
-    /// and waker bookkeeping.
+    /// In immediate mode, [`FutureExt::now_or_never`] owns the stack future and pins it in place for
+    /// its single poll. A ready future is consumed, while a pending future is dropped and causes the
+    /// caller's `expect` to panic; it is never moved after being polled. In pending-capable mode the
+    /// future must instead be boxed before its first poll, because moving a polled `!Unpin` future
+    /// from the stack into the queue would be unsound.
     fn queue_future<F>(&mut self, future: F)
     where
         F: std::future::Future<Output = Result<TaskResult<'cache, D>, Box<dyn Any>>> + 'cache,
     {
-        let mut boxed = future.boxed_local();
-        match boxed.as_mut().now_or_never() {
-            Some(result) => self.pending_results.push_back(result),
-            None => {
-                // Future is still pending. Hand the boxed future to
-                // `pending_futures` so it can be polled asynchronously.
-                self.pending_futures.push(boxed);
+        match self.future_queue_mode {
+            FutureQueueMode::Immediate => {
+                let result = future
+                    .now_or_never()
+                    .expect("can only use non-yielding futures with the NowOrNeverRuntime");
+                self.pending_results.push_back(result);
+            }
+            FutureQueueMode::PendingCapable => {
+                let mut boxed = future.boxed_local();
+                match boxed.as_mut().now_or_never() {
+                    Some(result) => self.pending_results.push_back(result),
+                    None => self.pending_futures.push(boxed),
+                }
             }
         }
     }

src/solver/mod.rs

@@ -6,7 +6,7 @@ use clause::{Clause, Literal, WatchedLiterals};
 use conditions::{DeferredRequirement, Disjunction, DisjunctionId, condition_disjunct_holds};
 use decision::Decision;
 use decision_tracker::DecisionTracker;
-use encoding::Encoder;
+use encoding::{Encoder, FutureQueueMode};
 use indexmap::IndexMap;
 use itertools::Itertools;
 use variable_map::VariableMap;
@@ -187,6 +187,10 @@ pub struct Solver<D: DependencyProvider, RT: AsyncRuntime = NowOrNeverRuntime> {
     /// Holds the current state of the solver.
     pub(crate) state: SolverState<D>,
 
+    /// Determines whether encoder futures may be polled later. Kept private so every custom
+    /// runtime remains pending-capable without extending the public runtime contract.
+    future_queue_mode: FutureQueueMode,
+
     /// The activity add factor. This is a value that is added to the activity
     /// score of each package that is part of a conflict.
     activity_add: f32,
@@ -372,6 +376,7 @@ impl<D: DependencyProvider> Solver<D, NowOrNeverRuntime> {
             cache: SolverCache::new(provider),
             async_runtime: NowOrNeverRuntime,
             state: SolverState::default(),
+            future_queue_mode: FutureQueueMode::Immediate,
             activity_add: 1.0,
             activity_decay: 0.95,
         }
@@ -448,6 +453,7 @@ impl<D: DependencyProvider, RT: AsyncRuntime> Solver<D, RT> {
             async_runtime: runtime,
             cache: self.cache,
             state: self.state,
+            future_queue_mode: FutureQueueMode::PendingCapable,
             activity_decay: self.activity_decay,
             activity_add: self.activity_add,
         }
@@ -634,8 +640,14 @@ impl<D: DependencyProvider, RT: AsyncRuntime> Solver<D, RT> {
                 #[cfg(feature = "diagnostics")]
                 let encoding_start = std::time::Instant::now();
                 let conflicting_clauses = self.async_runtime.block_on(
-                    Encoder::new(&mut self.state, &self.cache, root_deps, level)
-                        .encode([root_solvable]),
+                    Encoder::new(
+                        &mut self.state,
+                        &self.cache,
+                        root_deps,
+                        level,
+                        self.future_queue_mode,
+                    )
+                    .encode([root_solvable]),
                 )?;
                 #[cfg(feature = "diagnostics")]
                 {
@@ -789,8 +801,14 @@ impl<D: DependencyProvider, RT: AsyncRuntime> Solver<D, RT> {
             #[cfg(feature = "diagnostics")]
             let encoding_start = std::time::Instant::now();
             let conflicting_clauses = self.async_runtime.block_on(
-                Encoder::new(&mut self.state, &self.cache, root_deps, level)
-                    .encode_with_deferred(solvable_ids.iter().copied(), deferred_to_encode),
+                Encoder::new(
+                    &mut self.state,
+                    &self.cache,
+                    root_deps,
+                    level,
+                    self.future_queue_mode,
+                )
+                .encode_with_deferred(solvable_ids.iter().copied(), deferred_to_encode),
             )?;
             #[cfg(feature = "diagnostics")]
             {