diff --git a/formula/src/main/java/com/instacart/formula/FormulaRuntime.kt b/formula/src/main/java/com/instacart/formula/FormulaRuntime.kt index cbb80e7a..73df30e5 100644 --- a/formula/src/main/java/com/instacart/formula/FormulaRuntime.kt +++ b/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( private val formula: IFormula, 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( /** * 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() + private var globalEffectQueue = ArrayDeque(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( for (index in effects.indices) { globalEffectQueue.addLast(effects[index]) } + val size = globalEffectQueue.size + if (size > globalEffectQueuePeakSize) globalEffectQueuePeakSize = size } pendingEvaluation = pendingEvaluation || evaluate @@ -342,7 +356,7 @@ class FormulaRuntime( 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 @@ -350,6 +364,10 @@ class FormulaRuntime( } dispatcher.dispatch(effect) } + if (globalEffectQueuePeakSize > GLOBAL_EFFECT_QUEUE_SHRINK_THRESHOLD) { + globalEffectQueue = ArrayDeque(INITIAL_GLOBAL_EFFECT_QUEUE_CAPACITY) + } + globalEffectQueuePeakSize = 0 isExecutingEffects = false } diff --git a/formula/src/main/java/com/instacart/formula/batch/BatchImpl.kt b/formula/src/main/java/com/instacart/formula/batch/BatchImpl.kt index ab3645e3..665d5491 100644 --- a/formula/src/main/java/com/instacart/formula/batch/BatchImpl.kt +++ b/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() { diff --git a/formula/src/main/java/com/instacart/formula/runtime/FormulaManagerImpl.kt b/formula/src/main/java/com/instacart/formula/runtime/FormulaManagerImpl.kt index cb1482d3..5437d3c5 100644 --- a/formula/src/main/java/com/instacart/formula/runtime/FormulaManagerImpl.kt +++ b/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( 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? = null @@ -67,9 +74,13 @@ internal class FormulaManagerImpl( /** * 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>() + private var transitionQueue = ArrayDeque>(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( // Execute deferred transitions while (transitionQueue.isNotEmpty()) { - transitionQueue.pollFirst().execute() + transitionQueue.removeFirst().execute() } inspector?.onFormulaFinished(formulaType) @@ -317,19 +328,25 @@ internal class FormulaManagerImpl( 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, evaluate: Boolean) { if (evaluate) { globalEvaluationId += 1 @@ -381,13 +398,18 @@ internal class FormulaManagerImpl( */ 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 }