Merge pull request #2247 from DataDog/yl/sr-opt/cpu-bank

RUM-6093:Add TimeBank in session replay recorder for dynamic optimisation

features/dd-sdk-android-session-replay/src/main/kotlin/com/datadog/android/sessionreplay/internal/recorder/Debouncer.kt

@@ -8,11 +8,14 @@ package com.datadog.android.sessionreplay.internal.recorder
 
 import android.os.Handler
 import android.os.Looper
+import com.datadog.android.api.InternalLogger
 import java.util.concurrent.TimeUnit
 
 internal class Debouncer(
     private val handler: Handler = Handler(Looper.getMainLooper()),
-    private val maxRecordDelayInNs: Long = MAX_DELAY_THRESHOLD_NS
+    private val maxRecordDelayInNs: Long = MAX_DELAY_THRESHOLD_NS,
+    private val timeBank: TimeBank = RecordingTimeBank(),
+    private val internalLogger: InternalLogger
 ) {
 
     private var lastTimeRecordWasPerformed = 0L
@@ -37,8 +40,28 @@ internal class Debouncer(
     }
 
     private fun executeRunnable(runnable: Runnable) {
-        runnable.run()
-        lastTimeRecordWasPerformed = System.nanoTime()
+        runInTimeBalance {
+            runnable.run()
+            lastTimeRecordWasPerformed = System.nanoTime()
+        }
+    }
+
+    private fun runInTimeBalance(block: () -> Unit) {
+        if (timeBank.updateAndCheck(System.nanoTime())) {
+            val startTimeInNano = System.nanoTime()
+            block()
+            val endTimeInNano = System.nanoTime()
+            timeBank.consume(endTimeInNano - startTimeInNano)
+        } else {
+            // TODO RUM-5188 report with telemetry about the frames skipped
+            internalLogger.log(
+                InternalLogger.Level.INFO,
+                InternalLogger.Target.MAINTAINER,
+                messageBuilder = {
+                    MSG_FRAME_SKIP
+                }
+            )
+        }
     }
 
     companion object {
@@ -47,5 +70,7 @@ internal class Debouncer(
 
         // one frame time
         internal const val DEBOUNCE_TIME_IN_MS: Long = 64
+
+        private const val MSG_FRAME_SKIP = "Session Replay skipped this recording due to the time limit"
     }
 }

features/dd-sdk-android-session-replay/src/main/kotlin/com/datadog/android/sessionreplay/internal/recorder/RecordingTimeBank.kt

@@ -0,0 +1,53 @@
+/*
+ * Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+ * This product includes software developed at Datadog (https://www.datadoghq.com/).
+ * Copyright 2016-Present Datadog, Inc.
+ */
+
+package com.datadog.android.sessionreplay.internal.recorder
+
+import java.util.concurrent.TimeUnit
+import kotlin.math.min
+
+/**
+ * Time Bank is a concept representing an allocated execution quota per second. For example, if the quota is set to
+ * 100 milliseconds per second, it means that within any given second, no more than 100 milliseconds can be used for
+ * executing operations. If the full quota of 100 milliseconds has already been used within a second, further execution
+ * is not permitted until the next second begins and the quota is recharged. Conversely, if less than 100 milliseconds
+ * has been used and the second has not yet elapsed, execution may continue until the quota is reached.
+ */
+internal class RecordingTimeBank(
+    private val maxTimeBalancePerSecondInMs: Long = DEFAULT_MAX_TIME_BALANCE_PER_SEC_IN_MS
+) : TimeBank {
+
+    // The normalized factor of balance increasing by time. If increasing 100ms balance in the bank takes 1000ms,
+    // then the factor will be 100ms/1000ms = 0.1f
+    private val balanceFactor = maxTimeBalancePerSecondInMs.toDouble() / TimeUnit.SECONDS.toMillis(1)
+
+    @Volatile
+    private var recordingTimeBalanceInNano = TimeUnit.MILLISECONDS.toNanos(maxTimeBalancePerSecondInMs)
+
+    @Volatile
+    private var lastCheckTime: Long = 0
+
+    override fun consume(executionTime: Long) {
+        recordingTimeBalanceInNano -= executionTime
+    }
+
+    override fun updateAndCheck(timestamp: Long): Boolean {
+        increaseTimeBank(timestamp)
+        lastCheckTime = timestamp
+        return recordingTimeBalanceInNano >= 0
+    }
+
+    private fun increaseTimeBank(timestamp: Long) {
+        val timePassedSinceLastExecution = timestamp - lastCheckTime
+        recordingTimeBalanceInNano += (timePassedSinceLastExecution * balanceFactor).toLong()
+        recordingTimeBalanceInNano =
+            min(TimeUnit.MILLISECONDS.toNanos(maxTimeBalancePerSecondInMs), recordingTimeBalanceInNano)
+    }
+
+    companion object {
+        private const val DEFAULT_MAX_TIME_BALANCE_PER_SEC_IN_MS = 100L
+    }
+}

features/dd-sdk-android-session-replay/src/main/kotlin/com/datadog/android/sessionreplay/internal/recorder/TimeBank.kt

@@ -0,0 +1,26 @@
+/*
+ * Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+ * This product includes software developed at Datadog (https://www.datadoghq.com/).
+ * Copyright 2016-Present Datadog, Inc.
+ */
+
+package com.datadog.android.sessionreplay.internal.recorder
+
+import com.datadog.tools.annotation.NoOpImplementation
+
+@NoOpImplementation
+internal interface TimeBank {
+
+    /**
+     * Called to consume execution time from the bank.
+     */
+    fun consume(executionTime: Long)
+
+    /**
+     * Called to update time bank balance and check if the given timestamp
+     * is allowed according to the current time balance.
+     *
+     * @return true if the given timestamp is allowed by time bank to execute a task, false otherwise.
+     */
+    fun updateAndCheck(timestamp: Long): Boolean
+}

features/dd-sdk-android-session-replay/src/main/kotlin/com/datadog/android/sessionreplay/internal/recorder/listener/WindowsOnDrawListener.kt

@@ -28,9 +28,9 @@ internal class WindowsOnDrawListener(
     private val snapshotProducer: SnapshotProducer,
     private val privacy: SessionReplayPrivacy,
     private val imagePrivacy: ImagePrivacy,
-    private val debouncer: Debouncer = Debouncer(),
     private val miscUtils: MiscUtils = MiscUtils,
     private val internalLogger: InternalLogger,
+    private val debouncer: Debouncer = Debouncer(internalLogger = internalLogger),
     private val methodCallSamplingRate: Float
 ) : ViewTreeObserver.OnDrawListener {
 

features/dd-sdk-android-session-replay/src/test/kotlin/com/datadog/android/sessionreplay/internal/recorder/DebouncerTest.kt

@@ -7,6 +7,7 @@
 package com.datadog.android.sessionreplay.internal.recorder
 
 import android.os.Handler
+import com.datadog.android.api.InternalLogger
 import com.datadog.android.sessionreplay.forge.ForgeConfigurator
 import fr.xgouchet.elmyr.Forge
 import fr.xgouchet.elmyr.junit5.ForgeConfiguration
@@ -38,11 +39,18 @@ internal class DebouncerTest {
     @Mock
     lateinit var mockHandler: Handler
 
+    @Mock
+    lateinit var mockInternalLogger: InternalLogger
+
     lateinit var testedDebouncer: Debouncer
 
     @BeforeEach
     fun `set up`() {
-        testedDebouncer = Debouncer(mockHandler, TEST_MAX_DELAY_THRESHOLD_IN_NS)
+        testedDebouncer = Debouncer(
+            mockHandler,
+            TEST_MAX_DELAY_THRESHOLD_IN_NS,
+            internalLogger = mockInternalLogger
+        )
     }
 
     @Test

features/dd-sdk-android-session-replay/src/test/kotlin/com/datadog/android/sessionreplay/recorder/RecordingTimeBankTest.kt

@@ -0,0 +1,135 @@
+/*
+ * Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+ * This product includes software developed at Datadog (https://www.datadoghq.com/).
+ * Copyright 2016-Present Datadog, Inc.
+ */
+
+package com.datadog.android.sessionreplay.recorder
+
+import com.datadog.android.sessionreplay.forge.ForgeConfigurator
+import com.datadog.android.sessionreplay.internal.recorder.RecordingTimeBank
+import com.datadog.android.sessionreplay.internal.recorder.TimeBank
+import fr.xgouchet.elmyr.Forge
+import fr.xgouchet.elmyr.junit5.ForgeConfiguration
+import fr.xgouchet.elmyr.junit5.ForgeExtension
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.ExtendWith
+import org.junit.jupiter.api.extension.Extensions
+import org.mockito.junit.jupiter.MockitoExtension
+import org.mockito.junit.jupiter.MockitoSettings
+import org.mockito.quality.Strictness
+import java.util.concurrent.TimeUnit
+
+@Extensions(
+    ExtendWith(MockitoExtension::class),
+    ExtendWith(ForgeExtension::class)
+)
+@MockitoSettings(strictness = Strictness.LENIENT)
+@ForgeConfiguration(ForgeConfigurator::class)
+class RecordingTimeBankTest {
+
+    private lateinit var recordingTimeBank: TimeBank
+
+    @BeforeEach
+    fun `set up`() {
+        recordingTimeBank = RecordingTimeBank(TEST_MAX_BALANCE_IN_MS)
+    }
+
+    @Test
+    fun `M allow the first execution W check`(forge: Forge) {
+        // Given
+        val timestamp = forge.aLong(min = 0)
+
+        // When
+        val actual = recordingTimeBank.updateAndCheck(timestamp)
+
+        // Then
+        assertThat(actual).isTrue()
+    }
+
+    @Test
+    fun `M skip the next execution W previous consume out the balance`(forge: Forge) {
+        // Given
+        val firstTimestamp = forge.aLong(min = 0)
+        val firstExecutionTime = forge.aLong(
+            min = TimeUnit.MILLISECONDS.toNanos(TEST_MAX_BALANCE_IN_MS),
+            max = TimeUnit.MILLISECONDS.toNanos(TEST_MAX_BALANCE_IN_MS) * 100
+        )
+        val interval = forge.aLong(min = 0, max = firstExecutionTime)
+        val secondTimestamp = forge.aLong(
+            min = firstTimestamp + firstExecutionTime,
+            max = firstTimestamp + firstExecutionTime + interval
+        )
+
+        // When
+        recordingTimeBank.updateAndCheck(firstTimestamp)
+        recordingTimeBank.consume(firstExecutionTime)
+        val actual = recordingTimeBank.updateAndCheck(secondTimestamp)
+
+        // Then
+        assertThat(actual).isFalse()
+    }
+
+    @Test
+    fun `M allow the next execution W balance is recovery`(forge: Forge) {
+        // Given
+        val firstTimestamp = forge.aLong(min = 0)
+        val firstExecutionTime = forge.aLong(
+            min = TimeUnit.MILLISECONDS.toNanos(TEST_MAX_BALANCE_IN_MS),
+            max = TimeUnit.MILLISECONDS.toNanos(TEST_MAX_BALANCE_IN_MS) * 100
+        )
+        val interval = forge.aLong(min = 0, max = firstExecutionTime)
+        val secondTimestamp = forge.aLong(
+            min = firstTimestamp + firstExecutionTime,
+            max = firstTimestamp + firstExecutionTime + interval
+        )
+
+        val thirdTimestamp =
+            forge.aLong(
+                min = secondTimestamp + firstExecutionTime /
+                    ((TimeUnit.SECONDS.toMillis(1) / TEST_MAX_BALANCE_IN_MS))
+            )
+
+        // When
+        recordingTimeBank.updateAndCheck(firstTimestamp)
+        recordingTimeBank.consume(firstExecutionTime)
+        recordingTimeBank.updateAndCheck(secondTimestamp)
+        check(!recordingTimeBank.updateAndCheck(secondTimestamp))
+        val actual = recordingTimeBank.updateAndCheck(thirdTimestamp)
+
+        // Then
+        assertThat(actual).isTrue()
+    }
+
+    @Test
+    fun `M allow everything W set max balance more than 1000ms per sec`(forge: Forge) {
+        val maxBalancePerSecondInMs = forge.aLong(min = 1000)
+        recordingTimeBank = RecordingTimeBank(maxBalancePerSecondInMs)
+
+        // Given
+        val firstTimestamp = forge.aLong(min = 0)
+        val firstExecutionTime = forge.aLong(
+            min = TimeUnit.MILLISECONDS.toNanos(TEST_MAX_BALANCE_IN_MS),
+            max = TimeUnit.MILLISECONDS.toNanos(TEST_MAX_BALANCE_IN_MS) * 100
+        )
+        val interval = forge.aLong(min = 0, max = firstExecutionTime)
+        val secondTimestamp = forge.aLong(
+            min = firstTimestamp + firstExecutionTime,
+            max = firstTimestamp + firstExecutionTime + interval
+        )
+
+        // When
+        recordingTimeBank.updateAndCheck(firstTimestamp)
+        recordingTimeBank.consume(firstExecutionTime)
+        val actual = recordingTimeBank.updateAndCheck(secondTimestamp)
+
+        // Then
+        assertThat(actual).isEqualTo(true)
+    }
+
+    companion object {
+        private const val TEST_MAX_BALANCE_IN_MS = 100L
+    }
+}