Merge pull request #2745 from DataDog/nogorodnikov/rum-9851/monitor-backpressure-for-context-executor

0xnm · web-flow · commit 1379c79e6731 · 2025-06-20T10:15:41.000+02:00
RUM-9851: Monitor backpressure of context executor
diff --git a/dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/CoreFeature.kt b/dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/CoreFeature.kt
@@ -61,6 +61,7 @@ import com.datadog.android.core.internal.system.NoOpAppVersionProvider
 import com.datadog.android.core.internal.system.NoOpSystemInfoProvider
 import com.datadog.android.core.internal.system.SystemInfoProvider
 import com.datadog.android.core.internal.thread.BackPressureExecutorService
+import com.datadog.android.core.internal.thread.BackPressuredBlockingQueue
 import com.datadog.android.core.internal.thread.DatadogThreadFactory
 import com.datadog.android.core.internal.thread.LoggingScheduledThreadPoolExecutor
 import com.datadog.android.core.internal.thread.ScheduledExecutorServiceFactory
@@ -100,7 +101,6 @@ import java.io.FileNotFoundException
 import java.lang.ref.WeakReference
 import java.util.Locale
 import java.util.concurrent.ExecutorService
-import java.util.concurrent.LinkedBlockingQueue
 import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.ScheduledThreadPoolExecutor
 import java.util.concurrent.ThreadPoolExecutor
@@ -636,7 +636,16 @@ internal class CoreFeature(
             executorContext = "storage",
             backPressureStrategy = backpressureStrategy
         )
-        // TODO RUM-9851 Switch to the executor which is aware of backpressure, but only logs it
+        val contextQueue = BackPressuredBlockingQueue<Runnable>(
+            internalLogger,
+            executorContext = "context",
+            capacity = Int.MAX_VALUE,
+            notifyThreshold = 1024,
+            // just notify when reached
+            onItemDropped = {},
+            onThresholdReached = {},
+            backpressureMitigation = null
+        )
         @Suppress("UnsafeThirdPartyFunctionCall") // all parameters are safe
         contextExecutorService = ThreadPoolExecutor(
             // core pool size
@@ -646,7 +655,7 @@ internal class CoreFeature(
             // keep-alive time
             0L,
             TimeUnit.MILLISECONDS,
-            LinkedBlockingQueue(),
+            contextQueue,
             DatadogThreadFactory("context")
         )
     }
diff --git a/dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/thread/BackPressuredBlockingQueue.kt b/dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/thread/BackPressuredBlockingQueue.kt
@@ -12,13 +12,55 @@ import com.datadog.android.core.configuration.BackPressureStrategy
 import com.datadog.android.internal.thread.NamedExecutionUnit
 import java.util.concurrent.TimeUnit
 
-internal class BackPressuredBlockingQueue<E : Any>(
-    private val logger: InternalLogger,
-    private val executorContext: String,
-    private val backPressureStrategy: BackPressureStrategy
-) : ObservableLinkedBlockingQueue<E>(
-    backPressureStrategy.capacity
-) {
+/**
+ * [LinkedBlockingQueue] that supports backpressure handling via the chosen backpressure mitigation strategy.
+ *
+ * This queue may be either bounded or unbounded by specifying capacity. See docs of [LinkedBlockingQueue] for more
+ * details.
+ *
+ * If queue is unbounded, there is still a possibility to be notified if certain size threshold is reached.
+ */
+internal class BackPressuredBlockingQueue<E : Any> : ObservableLinkedBlockingQueue<E> {
+
+    private val logger: InternalLogger
+    private val executorContext: String
+    internal val capacity: Int
+    private val notifyThreshold: Int
+    private val onThresholdReached: () -> Unit
+    private val onItemDropped: (Any) -> Unit
+    private val backpressureMitigation: BackPressureMitigation?
+
+    constructor(
+        logger: InternalLogger,
+        executorContext: String,
+        backPressureStrategy: BackPressureStrategy
+    ) : this(
+        logger,
+        executorContext,
+        backPressureStrategy.capacity,
+        backPressureStrategy.capacity,
+        backPressureStrategy.onThresholdReached,
+        backPressureStrategy.onItemDropped,
+        backPressureStrategy.backpressureMitigation
+    )
+
+    constructor(
+        logger: InternalLogger,
+        executorContext: String,
+        notifyThreshold: Int,
+        capacity: Int,
+        onThresholdReached: () -> Unit,
+        onItemDropped: (Any) -> Unit,
+        backpressureMitigation: BackPressureMitigation?
+    ) : super(capacity) {
+        this.logger = logger
+        this.executorContext = executorContext
+        this.capacity = capacity
+        this.notifyThreshold = notifyThreshold
+        this.onThresholdReached = onThresholdReached
+        this.onItemDropped = onItemDropped
+        this.backpressureMitigation = backpressureMitigation
+    }
 
     override fun offer(e: E): Boolean {
         return addWithBackPressure(e) {
@@ -33,52 +75,59 @@ internal class BackPressuredBlockingQueue<E : Any>(
         if (!accepted) {
             return offer(e)
         } else {
-            if (remainingCapacity() == 0) {
-                onThresholdReached()
+            if (size == notifyThreshold) {
+                notifyThresholdReached()
             }
             return true
         }
     }
 
+    override fun put(e: E) {
+        if (size + 1 == notifyThreshold) {
+            notifyThresholdReached()
+        }
+        super.put(e)
+    }
+
     private fun addWithBackPressure(
         e: E,
         operation: (E) -> Boolean
     ): Boolean {
         val remainingCapacity = remainingCapacity()
         return if (remainingCapacity == 0) {
-            when (backPressureStrategy.backpressureMitigation) {
+            when (backpressureMitigation) {
                 BackPressureMitigation.DROP_OLDEST -> {
                     val first = take()
-                    onItemDropped(first)
+                    notifyItemDropped(first)
                     operation(e)
                 }
 
-                BackPressureMitigation.IGNORE_NEWEST -> {
-                    onItemDropped(e)
+                BackPressureMitigation.IGNORE_NEWEST, null -> {
+                    notifyItemDropped(e)
                     true
                 }
             }
         } else {
-            if (remainingCapacity == 1) {
-                onThresholdReached()
+            if (size + 1 == notifyThreshold) {
+                notifyThresholdReached()
             }
             operation(e)
         }
     }
 
-    private fun onThresholdReached() {
+    private fun notifyThresholdReached() {
         val dump = dumpQueue()
         val backPressureMap = buildMap {
-            put("capacity", backPressureStrategy.capacity)
+            put("capacity", capacity)
             if (!dump.isNullOrEmpty()) {
                 put("dump", dump)
             }
         }
-        backPressureStrategy.onThresholdReached()
+        onThresholdReached()
         logger.log(
             level = InternalLogger.Level.WARN,
             targets = listOf(InternalLogger.Target.MAINTAINER, InternalLogger.Target.TELEMETRY),
-            messageBuilder = { "BackPressuredBlockingQueue reached capacity:${backPressureStrategy.capacity}" },
+            messageBuilder = { "BackPressuredBlockingQueue reached capacity:$notifyThreshold" },
             throwable = null,
             onlyOnce = false,
             additionalProperties = mapOf(
@@ -88,8 +137,8 @@ internal class BackPressuredBlockingQueue<E : Any>(
         )
     }
 
-    private fun onItemDropped(item: E) {
-        backPressureStrategy.onItemDropped(item)
+    private fun notifyItemDropped(item: E) {
+        onItemDropped(item)
         val name = (item as? NamedExecutionUnit)?.name ?: item.toString()
         // Note, do not send this to telemetry as it might cause a stack overflow
         logger.log(
@@ -99,7 +148,7 @@ internal class BackPressuredBlockingQueue<E : Any>(
             throwable = null,
             onlyOnce = false,
             additionalProperties = mapOf(
-                "backpressure.capacity" to backPressureStrategy.capacity,
+                "backpressure.capacity" to capacity,
                 "executor.context" to executorContext
             )
         )
diff --git a/dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/thread/ObservableLinkedBlockingQueue.kt b/dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/thread/ObservableLinkedBlockingQueue.kt
@@ -11,12 +11,22 @@ import java.util.concurrent.LinkedBlockingQueue
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicLong
 
-internal open class ObservableLinkedBlockingQueue<E : Any>(
-    capacity: Int,
-    private val currentTimeProvider: () -> Long = { System.currentTimeMillis() }
-) : LinkedBlockingQueue<E>(capacity) {
+internal open class ObservableLinkedBlockingQueue<E : Any> : LinkedBlockingQueue<E> {
 
-    private var lastDumpTimestamp: AtomicLong = AtomicLong(0)
+    private val currentTimeProvider: () -> Long
+
+    constructor(
+        currentTimeProvider: () -> Long = { System.currentTimeMillis() }
+    ) : this(Int.MAX_VALUE, currentTimeProvider)
+
+    constructor(
+        capacity: Int,
+        currentTimeProvider: () -> Long = { System.currentTimeMillis() }
+    ) : super(capacity) {
+        this.currentTimeProvider = currentTimeProvider
+    }
+
+    private val lastDumpTimestamp: AtomicLong = AtomicLong(0)
 
     fun dumpQueue(): Map<String, Int>? {
         val currentTime = currentTimeProvider.invoke()
diff --git a/dd-sdk-android-core/src/test/kotlin/com/datadog/android/core/internal/CoreFeatureTest.kt b/dd-sdk-android-core/src/test/kotlin/com/datadog/android/core/internal/CoreFeatureTest.kt
@@ -29,6 +29,7 @@ import com.datadog.android.core.internal.privacy.NoOpConsentProvider
 import com.datadog.android.core.internal.privacy.TrackingConsentProvider
 import com.datadog.android.core.internal.system.BroadcastReceiverSystemInfoProvider
 import com.datadog.android.core.internal.system.NoOpSystemInfoProvider
+import com.datadog.android.core.internal.thread.BackPressuredBlockingQueue
 import com.datadog.android.core.internal.time.AppStartTimeProvider
 import com.datadog.android.core.internal.time.KronosTimeProvider
 import com.datadog.android.core.internal.time.NoOpTimeProvider
@@ -667,6 +668,24 @@ internal class CoreFeatureTest {
         assertThat(testedFeature.contextExecutorService).isNotNull()
     }
 
+    @Test
+    fun `M initialize context executor with unbounded + observable queue W initialize()`() {
+        // When
+        testedFeature.initialize(
+            appContext.mockInstance,
+            fakeSdkInstanceId,
+            fakeConfig,
+            fakeConsent
+        )
+
+        // Then
+        assertThat(testedFeature.contextExecutorService).isNotNull()
+        with(testedFeature.contextExecutorService.queue) {
+            check(this is BackPressuredBlockingQueue)
+            assertThat(capacity).isEqualTo(Int.MAX_VALUE)
+        }
+    }
+
     @Test
     fun `M initialize only once W initialize() twice`(
         @Forgery otherConfig: Configuration
diff --git a/dd-sdk-android-core/src/test/kotlin/com/datadog/android/core/internal/thread/DropOldestBackPressuredBlockingQueueTest.kt b/dd-sdk-android-core/src/test/kotlin/com/datadog/android/core/internal/thread/DropOldestBackPressuredBlockingQueueTest.kt
@@ -396,7 +396,8 @@ class DropOldestBackPressuredBlockingQueueTest {
         // Then
         assertThat(testedQueue).hasSize(fakeBackPressureThreshold)
         assertThat(testedQueue).contains(fakeNewItem)
-        verifyNoInteractions(mockOnItemsDropped, mockOnThresholdReached, mockLogger)
+        verify(mockOnThresholdReached).invoke()
+        verifyNoInteractions(mockOnItemsDropped)
     }
 
     @Test
@@ -421,7 +422,8 @@ class DropOldestBackPressuredBlockingQueueTest {
         // Then
         assertThat(testedQueue).hasSize(fakeBackPressureThreshold)
         assertThat(testedQueue).contains(fakeNewItem)
-        verifyNoInteractions(mockOnItemsDropped, mockOnThresholdReached, mockLogger)
+        verify(mockOnThresholdReached).invoke()
+        verifyNoInteractions(mockOnItemsDropped)
     }
 
     // endregion
diff --git a/dd-sdk-android-core/src/test/kotlin/com/datadog/android/core/internal/thread/IgnoreNewestBackPressuredBlockingQueueTest.kt b/dd-sdk-android-core/src/test/kotlin/com/datadog/android/core/internal/thread/IgnoreNewestBackPressuredBlockingQueueTest.kt
@@ -396,7 +396,8 @@ class IgnoreNewestBackPressuredBlockingQueueTest {
         // Then
         assertThat(testedQueue).hasSize(fakeBackPressureThreshold)
         assertThat(testedQueue).contains(fakeNewItem)
-        verifyNoInteractions(mockOnItemsDropped, mockOnThresholdReached, mockLogger)
+        verify(mockOnThresholdReached).invoke()
+        verifyNoInteractions(mockOnItemsDropped)
     }
 
     @Test
@@ -421,7 +422,8 @@ class IgnoreNewestBackPressuredBlockingQueueTest {
         // Then
         assertThat(testedQueue).hasSize(fakeBackPressureThreshold)
         assertThat(testedQueue).contains(fakeNewItem)
-        verifyNoInteractions(mockOnItemsDropped, mockOnThresholdReached, mockLogger)
+        verify(mockOnThresholdReached).invoke()
+        verifyNoInteractions(mockOnItemsDropped)
     }
 
     // endregion
diff --git a/dd-sdk-android-core/src/test/kotlin/com/datadog/android/core/internal/thread/NotifyOnlyUnboundedBackPressuredBlockingQueueTest.kt b/dd-sdk-android-core/src/test/kotlin/com/datadog/android/core/internal/thread/NotifyOnlyUnboundedBackPressuredBlockingQueueTest.kt
