[Performance] Use ArrayDeque for remaining LinkedList queues

formula/src/main/java/com/instacart/formula/FormulaRuntime.kt

@@ -11,7 +11,6 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
-import java.util.LinkedList
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.CoroutineContext
@@ -25,6 +24,15 @@ class FormulaRuntime<Input : Any, Output : Any>(
     private val formula: IFormula<Input, Output>,
     private val config: RuntimeConfig,
 ) : ManagerDelegate, BatchManager.Executor {
+    private companion object {
+        // Sized for the common case — a few effects per drain cycle. Stays under the stdlib
+        // default of 10 (which only kicks in on the no-arg constructor's first add).
+        private const val INITIAL_GLOBAL_EFFECT_QUEUE_CAPACITY = 8
+        // After a drain, replace the deque if we observed a burst this large. Picked well above
+        // typical cascading-transition depth so steady-state workloads never trigger a replace.
+        private const val GLOBAL_EFFECT_QUEUE_SHRINK_THRESHOLD = 64
+    }
+
     override val scope = CoroutineScope(
         context = coroutineContext + SupervisorJob(parent = coroutineContext[Job])
     )
@@ -81,8 +89,12 @@ class FormulaRuntime<Input : Any, Output : Any>(
 
     /**
      * Global transition effect queue which executes side-effects after all formulas are idle.
+     * Replaced with a fresh deque after a drain that observed a burst exceeding
+     * [GLOBAL_EFFECT_QUEUE_SHRINK_THRESHOLD] — [ArrayDeque] never shrinks its backing array on its
+     * own, so the deque would otherwise stay pinned at the peak size for the life of the runtime.
      */
-    private val globalEffectQueue = LinkedList<Effect>()
+    private var globalEffectQueue = ArrayDeque<Effect>(INITIAL_GLOBAL_EFFECT_QUEUE_CAPACITY)
+    private var globalEffectQueuePeakSize = 0
 
     /**
      * Determines if we are iterating through [globalEffectQueue]. It prevents us from
@@ -204,6 +216,8 @@ class FormulaRuntime<Input : Any, Output : Any>(
             for (index in effects.indices) {
                 globalEffectQueue.addLast(effects[index])
             }
+            val size = globalEffectQueue.size
+            if (size > globalEffectQueuePeakSize) globalEffectQueuePeakSize = size
         }
 
         pendingEvaluation = pendingEvaluation || evaluate
@@ -342,14 +356,18 @@ class FormulaRuntime<Input : Any, Output : Any>(
     private fun executeTransitionEffects() {
         isExecutingEffects = true
         while (globalEffectQueue.isNotEmpty()) {
-            val effect = globalEffectQueue.pollFirst()
+            val effect = globalEffectQueue.removeFirst()
             val dispatcher = when (effect.type) {
                 Effect.Unconfined -> Dispatcher.None
                 Effect.Main -> Dispatcher.Main
                 Effect.Background -> Dispatcher.Background
             }
             dispatcher.dispatch(effect)
         }
+        if (globalEffectQueuePeakSize > GLOBAL_EFFECT_QUEUE_SHRINK_THRESHOLD) {
+            globalEffectQueue = ArrayDeque(INITIAL_GLOBAL_EFFECT_QUEUE_CAPACITY)
+        }
+        globalEffectQueuePeakSize = 0
         isExecutingEffects = false
     }
 

formula/src/main/java/com/instacart/formula/batch/BatchImpl.kt

@@ -1,6 +1,5 @@
 package com.instacart.formula.batch
 
-import java.util.LinkedList
 import java.util.concurrent.atomic.AtomicBoolean
 
 internal class BatchImpl internal constructor(
@@ -9,11 +8,17 @@ internal class BatchImpl internal constructor(
     private val key: Any,
 ) : BatchScheduler.Batch {
 
+    private companion object {
+        // BatchImpl is single-shot — the instance is discarded after execute() — so no shrink
+        // logic is needed. Initial capacity sized for the common small-batch case.
+        private const val INITIAL_UPDATES_CAPACITY = 4
+    }
+
     private val isScheduled = AtomicBoolean(false)
-    private val updates = LinkedList<() -> Unit>()
+    private val updates = ArrayDeque<() -> Unit>(INITIAL_UPDATES_CAPACITY)
 
     fun add(update: () -> Unit) {
-        updates.add(update)
+        updates.addLast(update)
     }
 
     override fun execute() {

formula/src/main/java/com/instacart/formula/runtime/FormulaManagerImpl.kt

@@ -16,7 +16,6 @@ import com.instacart.formula.lifecycle.LifecycleScheduler
 import com.instacart.formula.lifecycle.ValidationException
 import com.instacart.formula.plugin.ChildAlreadyUsedException
 import com.instacart.formula.plugin.FormulaError
-import java.util.LinkedList
 
 /**
  * Responsible for keeping track of formula's state, running actions, and child formulas. The
@@ -36,6 +35,14 @@ internal class FormulaManagerImpl<Input, State, Output>(
     EffectDelegate,
     LifecycleComponent {
 
+    private companion object {
+        // Most evaluations queue at most a handful of deferred transitions in flight.
+        private const val INITIAL_TRANSITION_QUEUE_CAPACITY = 4
+        // After a drain, replace the deque if we observed a burst this large. Long-lived
+        // formulas would otherwise pin the high-water capacity for their entire lifetime.
+        private const val TRANSITION_QUEUE_SHRINK_THRESHOLD = 32
+    }
+
     private val lifecycleCache = LifecycleCacheImpl(this)
     private var state: State = formula.initialState(initialInput)
     private var frame: Frame<Input, State, Output>? = null
@@ -67,9 +74,13 @@ internal class FormulaManagerImpl<Input, State, Output>(
     /**
      * Pending transition queue which will be populated and executed within [run] function
      * while [isRunning] is true. If [isRunning] is false, we will pass the transitions
-     * to [ManagerDelegate].
+     * to [ManagerDelegate]. Replaced with a fresh deque after a drain that observed a burst
+     * exceeding [TRANSITION_QUEUE_SHRINK_THRESHOLD] — [ArrayDeque] never shrinks its backing
+     * array on its own, so the deque would otherwise stay pinned at the peak size for the
+     * life of this formula manager.
      */
-    private val transitionQueue = LinkedList<DeferredTransition<*, *, *>>()
+    private var transitionQueue = ArrayDeque<DeferredTransition<*, *, *>>(INITIAL_TRANSITION_QUEUE_CAPACITY)
+    private var transitionQueuePeakSize = 0
 
     fun canUpdatesContinue(evaluationId: Long): Boolean {
         return !isEvaluationNeeded(evaluationId) && transitionQueue.isEmpty()
@@ -307,7 +318,7 @@ internal class FormulaManagerImpl<Input, State, Output>(
 
         // Execute deferred transitions
         while (transitionQueue.isNotEmpty()) {
-            transitionQueue.pollFirst().execute()
+            transitionQueue.removeFirst().execute()
         }
 
         inspector?.onFormulaFinished(formulaType)
@@ -317,19 +328,25 @@ internal class FormulaManagerImpl<Input, State, Output>(
         if (terminated) {
             transition.execute()
         } else if (isRunning) {
-            transitionQueue.addLast(transition)
+            enqueueTransition(transition)
         } else {
             val lastFrame = frame
             if (lastFrame == null || isEvaluationNeeded(lastFrame.associatedEvaluationId)) {
                 // Since evaluation is already needed, we can wait for it to happen and
                 // then we'll execute the transition.
-                transitionQueue.addLast(transition)
+                enqueueTransition(transition)
             } else {
                 transition.execute()
             }
         }
     }
 
+    private fun enqueueTransition(transition: DeferredTransition<*, *, *>) {
+        transitionQueue.addLast(transition)
+        val size = transitionQueue.size
+        if (size > transitionQueuePeakSize) transitionQueuePeakSize = size
+    }
+
     override fun onPostTransition(effects: List<Effect>, evaluate: Boolean) {
         if (evaluate) {
             globalEvaluationId += 1
@@ -381,13 +398,18 @@ internal class FormulaManagerImpl<Input, State, Output>(
      */
     private fun handleTransitionQueue(evaluationId: Long): Boolean {
         while (transitionQueue.isNotEmpty()) {
-            val event = transitionQueue.pollFirst()
+            val event = transitionQueue.removeFirst()
             event.execute()
             if (isEvaluationNeeded(evaluationId)) {
                 return true
             }
         }
-
+        // Queue drained without forcing re-evaluation — safe point to reclaim the backing array
+        // if a burst pushed us past the shrink threshold.
+        if (transitionQueuePeakSize > TRANSITION_QUEUE_SHRINK_THRESHOLD) {
+            transitionQueue = ArrayDeque(INITIAL_TRANSITION_QUEUE_CAPACITY)
+        }
+        transitionQueuePeakSize = 0
         return false
     }