@@ -12,104 +12,83 @@ import com.datadog.android.core.internal.utils.scheduleSafe
1212import com.datadog.android.rum.internal.domain.RumContext
1313import com.datadog.android.rum.internal.domain.scope.RumViewType
1414import java.util.concurrent.ScheduledExecutorService
15+ import java.util.concurrent.ScheduledFuture
1516import java.util.concurrent.TimeUnit
16- import java.util.concurrent.atomic.AtomicInteger
17- import java.util.concurrent.atomic.AtomicReference
18-
19- /* *
20- * Per-session timeseries collector.
21- *
22- * Lifecycle invariants (enforced by [onSessionStart]/[onSessionStop] idempotency):
23- * - Each instance is single-use: exactly one [onSessionStart] followed by exactly one [onSessionStop].
24- * - Multiple [onSessionStart] / [onSessionStop] calls are safe — duplicate calls are no-ops.
25- * - After [onSessionStop], the instance must not be restarted; create a new instance.
26- *
27- * Background sampling:
28- * - When [collectInBackground] is `false`, [onRumContextUpdate] pauses sampling on leaving foreground
29- * and resumes it on returning to foreground.
30- *
31- * Threading:
32- * - [onSessionStart] / [onSessionStop] / [onRumContextUpdate] are called from the RUM event-handler thread.
33- * - Sampling tasks run on a [scheduledExecutorService] shared with other RUM components
34- * (owned by the SDK core); the instance neither owns nor shuts it down.
35- * - [Pipeline] is responsible for its own thread safety via internal synchronization.
36- * - Duplicate-chain prevention: each sampling chain carries a generation number.
37- * When [startSampling] starts a new generation, any in-flight or queued ticks from the
38- * previous generation self-terminate on their first check.
39- */
17+
4018internal class DefaultTimeseriesCollector (
4119 private val internalLogger : InternalLogger ,
4220 internal val pipelines : List <Pipeline <* >>,
43- private val collectInBackground : Boolean ,
4421 internal val scheduledExecutorService : ScheduledExecutorService ,
45- // Read by the sampling ticks so every batch is attributed to the view that was active when it
46- // was drained, not to the one active when the collector was created.
4722 @Volatile private var rumContext : RumContext
4823) : TimeseriesCollector {
24+ private val state = State ()
4925
50- private enum class State { IDLE , RUNNING , SUSPENDED , STOPPED }
51- private val state = AtomicReference (State .IDLE )
26+ // Batches are attributed to the view they are sent from, but a flush can happen once the app
27+ // already left the foreground, where there is no view to attribute to. Keep the last foreground
28+ // context so those batches stay attached to the view they were sent from.
29+ @Volatile
30+ private var lastForegroundRumContext: RumContext = rumContext
5231
53- // Incremented on every start/resume. Ticks carrying a stale generation self-terminate.
54- private val currentGeneration = AtomicInteger ( 0 )
32+ @Volatile
33+ private var recentSuspension : ScheduledFuture < * > ? = null
5534
35+ // region PUBLIC API
5636 @WorkerThread
5737 override fun onSessionStart () {
58- if (state.compareAndSet(State .IDLE , State .RUNNING )) {
59- startSampling()
60- }
38+ state.set(isActive = rumContext.viewType.isForeground)?.let { generation -> scheduleSampling(generation) }
6139 }
6240
6341 @WorkerThread
6442 override fun onSessionStop () {
65- if (state.getAndSet(State .STOPPED ) != State .STOPPED ) {
66- pipelines.forEach { pipeline ->
67- try {
68- pipeline.flush(rumContext)
69- } catch (@Suppress(" TooGenericExceptionCaught" : Throwable ) {
70- internalLogger.log(
71- level = InternalLogger .Level .ERROR ,
72- targets = listOf (InternalLogger .Target .MAINTAINER , InternalLogger .Target .TELEMETRY ),
73- messageBuilder = { ERROR_FLUSH_FAILED },
74- throwable = t
75- )
76- }
77- }
43+ cancelSuspension()
44+ if (state.set(false ) != null ) {
45+ flushPipelines(lastForegroundRumContext)
7846 }
7947 }
8048
8149 @WorkerThread
8250 override fun onRumContextUpdate (newRumContext : RumContext ) {
8351 val oldViewType = rumContext.viewType
8452 val newViewType = newRumContext.viewType
53+
8554 val isEnterForeground = ! oldViewType.isForeground && newViewType.isForeground
55+ val isLeaveForeground = oldViewType.isForeground && ! newViewType.isForeground
8656
8757 rumContext = newRumContext
58+ if (newViewType.isForeground) lastForegroundRumContext = newRumContext
8859
89- if (! collectInBackground) {
90- if (isEnterForeground && state.compareAndSet(State .SUSPENDED , State .RUNNING )) {
91- startSampling()
92- } else if (! newViewType.isForeground) {
93- state.compareAndSet(State .RUNNING , State .SUSPENDED )
94- }
60+ if (isLeaveForeground) {
61+ scheduleStop(state.currentGeneration)
62+ } else if (isEnterForeground) {
63+ scheduleSampling(generation = state.startGeneration())
9564 }
9665 }
9766
98- private fun startSampling () {
99- val generation = currentGeneration.incrementAndGet()
100- pipelines.forEach { schedulePipeline(it, generation) }
67+ // endregion
68+
69+ // region SAMPLING
70+
71+ private fun scheduleSampling (generation : Int ) {
72+ cancelSuspension()
73+ pipelines.forEach { pipeline -> scheduledExecutorService.schedulePipeline(pipeline, generation) }
10174 }
10275
103- private fun schedulePipeline (pipeline : Pipeline <* >, generation : Int ) {
104- scheduledExecutorService.scheduleSafe(
105- OPERATION_NAME ,
76+ private fun ScheduledExecutorService.schedulePipeline (
77+ pipeline : Pipeline <* >,
78+ generation : Int
79+ ) {
80+ scheduleSafe(
81+ TIMESERIES_OPERATION_NAME ,
10682 pipeline.intervalMs,
10783 TimeUnit .MILLISECONDS ,
10884 internalLogger
10985 ) {
110- if (! isActive (generation)) return @scheduleSafe
86+ if (! state.isGenerationActive (generation)) return @scheduleSafe
11187 try {
112- pipeline.execute(rumContext)
88+ val currentRumContext = rumContext
89+ if (currentRumContext.viewType.isForeground) {
90+ pipeline.execute(currentRumContext)
91+ }
11392 } catch (@Suppress(" TooGenericExceptionCaught" : Throwable ) {
11493 internalLogger.log(
11594 level = InternalLogger .Level .ERROR ,
@@ -118,24 +97,109 @@ internal class DefaultTimeseriesCollector(
11897 throwable = t
11998 )
12099 } finally {
121- if (isActive(generation)) schedulePipeline(pipeline, generation)
100+ if (state.isGenerationActive(generation)) schedulePipeline(pipeline, generation)
101+ }
102+ }
103+ }
104+
105+ @WorkerThread
106+ private fun flushPipelines (rumContext : RumContext ) {
107+ pipelines.forEach { pipeline ->
108+ try {
109+ pipeline.flush(rumContext)
110+ } catch (@Suppress(" TooGenericExceptionCaught" : Throwable ) {
111+ internalLogger.log(
112+ level = InternalLogger .Level .ERROR ,
113+ targets = listOf (InternalLogger .Target .MAINTAINER , InternalLogger .Target .TELEMETRY ),
114+ messageBuilder = { ERROR_FLUSH_FAILED },
115+ throwable = t
116+ )
122117 }
123118 }
124119 }
125120
126- private fun isActive (generation : Int ): Boolean =
127- state.get() == State .RUNNING && currentGeneration.get() == generation
121+ // endregion
122+
123+ // region SUSPENSION
124+
125+ private fun scheduleStop (stopRequestGeneration : Int ) {
126+ recentSuspension = scheduledExecutorService.scheduleSafe(
127+ SUSPEND_OPERATION_NAME ,
128+ SUSPEND_DELAY_MS ,
129+ TimeUnit .MILLISECONDS ,
130+ internalLogger
131+ ) {
132+ if (! rumContext.viewType.isForeground && state.stopGeneration(stopRequestGeneration)) {
133+ flushPipelines(lastForegroundRumContext)
134+ }
135+ }
136+ }
137+
138+ private fun cancelSuspension () {
139+ recentSuspension?.cancel(false )
140+ recentSuspension = null
141+ }
142+
143+ // endregion
128144
129145 internal companion object {
130- const val OPERATION_NAME = " Timeseries sampling"
146+ const val TIMESERIES_OPERATION_NAME = " Timeseries sampling"
147+ const val SUSPEND_OPERATION_NAME = " Timeseries suspend"
131148 const val ERROR_SAMPLING_FAILED = " Timeseries sampling iteration failed; rescheduling next sample."
132- const val ERROR_FLUSH_FAILED = " Timeseries flush on session stop failed."
149+ const val ERROR_FLUSH_FAILED = " Timeseries flush failed."
150+
151+ // Matches ActivityViewTrackingStrategy.STOP_VIEW_DELAY_MS, which guards the same race:
152+ // an Activity-to-Activity transition leaves no active view for a moment when the tracking
153+ // strategy stops the view on pause rather than on stop.
154+ const val SUSPEND_DELAY_MS = 200L
155+
133156 val RumViewType ?.isForeground: Boolean
134157 get() = when (this ) {
135158 RumViewType .FOREGROUND , RumViewType .APPLICATION_LAUNCH -> true
136159 RumViewType .BACKGROUND -> false
137160 RumViewType .NONE -> false
138161 null -> false
139162 }
163+
164+ private class State {
165+ // Written under the monitor only, but read outside of it, hence @Volatile: a stale
166+ // generation would make the guards below reject a live sampling chain or flush.
167+ @Volatile
168+ var currentGeneration: Int = 0
169+ private set
170+
171+ private var active: Boolean = false
172+
173+ fun isGenerationActive (generation : Int ): Boolean = synchronized(this ) {
174+ currentGeneration == generation && active
175+ }
176+
177+ /* *
178+ * Always starts a fresh generation, so that a suspension pending on the previous one
179+ * can no longer stop the sampling chain. Returns the new generation.
180+ */
181+ fun startGeneration (): Int = synchronized(this ) {
182+ active = true
183+ ++ currentGeneration
184+ }
185+
186+ /* *
187+ * Deactivates [generation] if it is still the current one.
188+ * Returns true when this call is the one that deactivated it.
189+ */
190+ fun stopGeneration (generation : Int ): Boolean = synchronized(this ) {
191+ currentGeneration == generation && set(false ) != null
192+ }
193+
194+ /* * Returns the new generation if this call changed the state, null otherwise. */
195+ fun set (isActive : Boolean ): Int? = synchronized(this ) {
196+ if (isActive == active) {
197+ null
198+ } else {
199+ active = isActive
200+ ++ currentGeneration
201+ }
202+ }
203+ }
140204 }
141205}
0 commit comments