-
Notifications
You must be signed in to change notification settings - Fork 57
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1938 from DataDog/xgouchet/RUM-3670/backpressure_…
…strategy RUM-3670 backpressure strategy
- Loading branch information
Showing
14 changed files
with
1,222 additions
and
22 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
19 changes: 19 additions & 0 deletions
19
...oid-core/src/main/kotlin/com/datadog/android/core/configuration/BackPressureMitigation.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| /* | ||
| * 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.core.configuration | ||
|
|
||
| /** | ||
| * Defines the mitigation to use when a queue hits the maximum back pressure capacity. | ||
| */ | ||
| enum class BackPressureMitigation { | ||
|
|
||
| /** Drop the oldest items already in the queue to make room for new ones. */ | ||
| DROP_OLDEST, | ||
|
|
||
| /** Ignore newest items that are not yet in the queue. */ | ||
| IGNORE_NEWEST | ||
| } |
20 changes: 20 additions & 0 deletions
20
...droid-core/src/main/kotlin/com/datadog/android/core/configuration/BackPressureStrategy.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| /* | ||
| * 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.core.configuration | ||
|
|
||
| /** | ||
| * @param capacity the maximum size of the queue | ||
| * @param onThresholdReached callback called when the queue reaches full capacity | ||
| * @param onItemDropped called when an item is dropped because of this backpressure strategy | ||
| * @param backpressureMitigation the mitigation to use when reaching the capacity | ||
| */ | ||
| data class BackPressureStrategy( | ||
| val capacity: Int, | ||
| val onThresholdReached: () -> Unit, | ||
| val onItemDropped: (Any) -> Unit, | ||
| val backpressureMitigation: BackPressureMitigation | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
71 changes: 71 additions & 0 deletions
71
...e/src/main/kotlin/com/datadog/android/core/internal/thread/BackPressureExecutorService.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| /* | ||
| * 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.core.internal.thread | ||
|
|
||
| import com.datadog.android.api.InternalLogger | ||
| import com.datadog.android.core.configuration.BackPressureStrategy | ||
| import com.datadog.android.core.thread.FlushableExecutorService | ||
| import java.util.concurrent.ThreadPoolExecutor | ||
| import java.util.concurrent.TimeUnit | ||
|
|
||
| /** | ||
| * A single threaded executor service using a BackPressureStrategy. | ||
| */ | ||
| internal class BackPressureExecutorService( | ||
| val logger: InternalLogger, | ||
| backpressureStrategy: BackPressureStrategy | ||
| ) : ThreadPoolExecutor( | ||
| CORE_POOL_SIZE, | ||
| CORE_POOL_SIZE, | ||
| THREAD_POOL_MAX_KEEP_ALIVE_MS, | ||
| TimeUnit.MILLISECONDS, | ||
| BackPressuredBlockingQueue(logger, backpressureStrategy) | ||
| ), | ||
| FlushableExecutorService { | ||
|
|
||
| // region FlushableExecutorService | ||
|
|
||
| @Suppress("TooGenericExceptionCaught") | ||
| override fun drainTo(destination: MutableCollection<Runnable>) { | ||
| try { | ||
| queue.drainTo(destination) | ||
| } catch (e: IllegalArgumentException) { | ||
| onDrainException(e) | ||
| } catch (e: NullPointerException) { | ||
| onDrainException(e) | ||
| } catch (e: UnsupportedOperationException) { | ||
| onDrainException(e) | ||
| } catch (e: ClassCastException) { | ||
| onDrainException(e) | ||
| } | ||
| } | ||
|
|
||
| // endregion | ||
|
|
||
| // region ThreadPoolExecutor | ||
|
|
||
| override fun afterExecute(r: Runnable?, t: Throwable?) { | ||
| super.afterExecute(r, t) | ||
| loggingAfterExecute(r, t, logger) | ||
| } | ||
|
|
||
| // endregion | ||
|
|
||
| private fun onDrainException(e: RuntimeException) { | ||
| logger.log( | ||
| InternalLogger.Level.ERROR, | ||
| listOf(InternalLogger.Target.MAINTAINER, InternalLogger.Target.TELEMETRY), | ||
| { "Unable to drain BackPressureExecutorService queue" }, | ||
| e | ||
| ) | ||
| } | ||
|
|
||
| companion object { | ||
| private const val CORE_POOL_SIZE = 1 | ||
| private val THREAD_POOL_MAX_KEEP_ALIVE_MS = TimeUnit.SECONDS.toMillis(5) | ||
| } | ||
| } |
91 changes: 91 additions & 0 deletions
91
...re/src/main/kotlin/com/datadog/android/core/internal/thread/BackPressuredBlockingQueue.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| /* | ||
| * 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.core.internal.thread | ||
|
|
||
| import com.datadog.android.api.InternalLogger | ||
| import com.datadog.android.core.configuration.BackPressureMitigation | ||
| import com.datadog.android.core.configuration.BackPressureStrategy | ||
| import java.util.concurrent.LinkedBlockingQueue | ||
| import java.util.concurrent.TimeUnit | ||
|
|
||
| internal class BackPressuredBlockingQueue<E : Any>( | ||
| private val logger: InternalLogger, | ||
| private val backPressureStrategy: BackPressureStrategy | ||
| ) : LinkedBlockingQueue<E>( | ||
| backPressureStrategy.capacity | ||
| ) { | ||
| override fun offer(e: E): Boolean { | ||
| return addWithBackPressure(e) { | ||
| @Suppress("UnsafeThirdPartyFunctionCall") // can't have NPE here | ||
| super.offer(it) | ||
| } | ||
| } | ||
|
|
||
| override fun offer(e: E, timeout: Long, unit: TimeUnit?): Boolean { | ||
| @Suppress("UnsafeThirdPartyFunctionCall") // can't have NPE here | ||
| val accepted = super.offer(e, timeout, unit) | ||
| if (!accepted) { | ||
| return offer(e) | ||
| } else { | ||
| if (remainingCapacity() == 0) { | ||
| onThresholdReached() | ||
| } | ||
| return true | ||
| } | ||
| } | ||
|
|
||
| private fun addWithBackPressure( | ||
| e: E, | ||
| operation: (E) -> Boolean | ||
| ): Boolean { | ||
| val remainingCapacity = remainingCapacity() | ||
| return if (remainingCapacity == 0) { | ||
| when (backPressureStrategy.backpressureMitigation) { | ||
| BackPressureMitigation.DROP_OLDEST -> { | ||
| val first = take() | ||
| onItemDropped(first) | ||
| operation(e) | ||
| } | ||
|
|
||
| BackPressureMitigation.IGNORE_NEWEST -> { | ||
| onItemDropped(e) | ||
| true | ||
| } | ||
| } | ||
| } else { | ||
| if (remainingCapacity == 1) { | ||
| onThresholdReached() | ||
| } | ||
| operation(e) | ||
| } | ||
| } | ||
|
|
||
| private fun onThresholdReached() { | ||
| backPressureStrategy.onThresholdReached() | ||
| logger.log( | ||
| level = InternalLogger.Level.WARN, | ||
| targets = listOf(InternalLogger.Target.MAINTAINER, InternalLogger.Target.TELEMETRY), | ||
| messageBuilder = { "BackPressuredBlockingQueue reached capacity:${backPressureStrategy.capacity}" }, | ||
| throwable = null, | ||
| onlyOnce = false, | ||
| additionalProperties = mapOf("backpressure.capacity" to backPressureStrategy.capacity) | ||
| ) | ||
| } | ||
|
|
||
| private fun onItemDropped(item: E) { | ||
| backPressureStrategy.onItemDropped(item) | ||
|
|
||
| logger.log( | ||
| level = InternalLogger.Level.ERROR, | ||
| targets = listOf(InternalLogger.Target.MAINTAINER, InternalLogger.Target.TELEMETRY), | ||
| messageBuilder = { "Dropped item in BackPressuredBlockingQueue queue: $item" }, | ||
| throwable = null, | ||
| onlyOnce = false, | ||
| additionalProperties = mapOf("backpressure.capacity" to backPressureStrategy.capacity) | ||
| ) | ||
| } | ||
| } |
Oops, something went wrong.