@@ -11,102 +11,104 @@ import com.datadog.android.api.InternalLogger
1111import com.datadog.android.core.internal.utils.scheduleSafe
1212import com.datadog.android.rum.internal.domain.scope.RumViewType
1313import java.util.concurrent.ScheduledExecutorService
14+ import java.util.concurrent.ScheduledFuture
1415import java.util.concurrent.TimeUnit
15- import java.util.concurrent.atomic.AtomicInteger
16- import java.util.concurrent.atomic.AtomicReference
1716
1817/* *
1918 * Per-session timeseries collector.
2019 *
21- * Lifecycle invariants (enforced by [onSessionStart]/[onSessionStop] idempotency) :
20+ * Lifecycle:
2221 * - Each instance is single-use: exactly one [onSessionStart] followed by exactly one [onSessionStop].
23- * - Multiple [onSessionStart] / [onSessionStop] calls are safe — duplicate calls are no-ops .
24- * - After [onSessionStop], the instance must not be restarted; create a new instance .
22+ * Duplicate calls are no-ops, and `RumSessionScope` builds a fresh instance per session .
23+ * - [onViewTypeUpdate] is only meaningful between those two calls .
2524 *
26- * Background sampling:
27- * - When [collectInBackground] is `false`, [onViewTypeUpdate] pauses sampling on leaving foreground
28- * and resumes it on returning to foreground.
25+ * Foreground gating:
26+ * - Sampling only runs while the app is in the foreground, so [onSessionStart] starts a sampling
27+ * chain only when the session begins on a foreground view.
28+ * - Leaving the foreground schedules a deferred suspension which flushes whatever is buffered, so
29+ * the pending batch reaches the writer instead of being dropped when the app is backgrounded.
30+ * - Returning to the foreground cancels a suspension that has not fired yet, or starts a fresh
31+ * sampling generation if it already did.
2932 *
3033 * Threading:
3134 * - [onSessionStart] / [onSessionStop] / [onViewTypeUpdate] are called from the RUM event-handler thread.
32- * - Sampling tasks run on a [scheduledExecutorService] shared with other RUM components
33- * (owned by the SDK core); the instance neither owns nor shuts it down.
34- * - [Pipeline] is responsible for its own thread safety via internal synchronization.
35- * - Duplicate-chain prevention: each sampling chain carries a generation number.
36- * When [startSampling] starts a new generation, any in-flight or queued ticks from the
37- * previous generation self-terminate on their first check.
35+ * - Sampling and suspension tasks run on a [scheduledExecutorService] shared with other RUM
36+ * components (owned by the SDK core); the instance neither owns nor shuts it down.
37+ * - Access to a [Pipeline] is serialized on the pipeline instance, so a flush from the suspension
38+ * task cannot interleave with a sampling tick.
39+ * - Duplicate-chain prevention: each sampling chain carries a generation number. When a new
40+ * generation starts, in-flight or queued ticks from the previous one self-terminate on their
41+ * first check.
3842 */
3943internal class DefaultTimeseriesCollector (
4044 private val internalLogger : InternalLogger ,
4145 internal val pipelines : List <Pipeline <* >>,
42- private val collectInBackground : Boolean ,
43- internal val scheduledExecutorService : ScheduledExecutorService
46+ internal val scheduledExecutorService : ScheduledExecutorService ,
47+ @Volatile private var currentViewType : RumViewType ?
4448) : TimeseriesCollector {
4549
46- private enum class State { IDLE , RUNNING , SUSPENDED , STOPPED }
47- private val state = AtomicReference (State .IDLE )
48-
49- // Incremented on every start/resume. Ticks carrying a stale generation self-terminate.
50- private val currentGeneration = AtomicInteger (0 )
50+ private val state = State ()
5151
5252 @Volatile
53- private var currentViewType: RumViewType ? = null
53+ private var recentSuspension: ScheduledFuture <* >? = null
54+
55+ // region PUBLIC API
5456
5557 @WorkerThread
5658 override fun onSessionStart () {
57- if (state.compareAndSet(State .IDLE , State .RUNNING )) {
58- startSampling()
59- }
59+ state.set(isActive = currentViewType.isForeground)?.let { generation -> scheduleSampling(generation) }
6060 }
6161
6262 @WorkerThread
6363 override fun onSessionStop () {
64- if (state.getAndSet(State .STOPPED ) != State .STOPPED ) {
65- pipelines.forEach { pipeline ->
66- try {
67- synchronized(pipeline, pipeline::flush)
68- } catch (@Suppress(" TooGenericExceptionCaught" : Throwable ) {
69- internalLogger.log(
70- level = InternalLogger .Level .ERROR ,
71- targets = listOf (InternalLogger .Target .MAINTAINER , InternalLogger .Target .TELEMETRY ),
72- messageBuilder = { ERROR_FLUSH_FAILED },
73- throwable = t
74- )
75- }
76- }
64+ cancelSuspension()
65+ if (state.set(false ) != null ) {
66+ flushPipelines()
7767 }
7868 }
7969
8070 @WorkerThread
8171 override fun onViewTypeUpdate (newViewType : RumViewType ) {
82- if (newViewType == currentViewType) return
83- val isEnterForeground = ! currentViewType.isForeground && newViewType.isForeground
72+ val oldViewType = currentViewType
73+ if (newViewType == oldViewType) return
74+
75+ val isEnterForeground = ! oldViewType.isForeground && newViewType.isForeground
76+ val isLeaveForeground = oldViewType.isForeground && ! newViewType.isForeground
77+
8478 currentViewType = newViewType
85- if (! collectInBackground) {
86- if (isEnterForeground && state.compareAndSet(State .SUSPENDED , State .RUNNING )) {
87- startSampling()
88- } else if (! newViewType.isForeground) {
89- state.compareAndSet(State .RUNNING , State .SUSPENDED )
90- }
79+
80+ if (isLeaveForeground) {
81+ scheduleStop(state.currentGeneration)
82+ } else if (isEnterForeground) {
83+ scheduleSampling(generation = state.startGeneration())
9184 }
9285 }
9386
94- private fun startSampling () {
95- val generation = currentGeneration.incrementAndGet()
96- pipelines.forEach { schedulePipeline(it, generation) }
87+ // endregion
88+
89+ // region SAMPLING
90+
91+ private fun scheduleSampling (generation : Int ) {
92+ cancelSuspension()
93+ pipelines.forEach { pipeline -> scheduledExecutorService.schedulePipeline(pipeline, generation) }
9794 }
9895
99- private fun schedulePipeline (pipeline : Pipeline <* >, generation : Int ) {
100- scheduledExecutorService.scheduleSafe(
101- OPERATION_NAME ,
96+ private fun ScheduledExecutorService.schedulePipeline (
97+ pipeline : Pipeline <* >,
98+ generation : Int
99+ ) {
100+ scheduleSafe(
101+ TIMESERIES_OPERATION_NAME ,
102102 pipeline.intervalMs,
103103 TimeUnit .MILLISECONDS ,
104104 internalLogger
105105 ) {
106- if (! isActive (generation)) return @scheduleSafe
106+ if (! state.isGenerationActive (generation)) return @scheduleSafe
107107 try {
108- synchronized(pipeline) {
109- if (isActive(generation)) pipeline.execute()
108+ // The chain stays alive during the suspension delay, so a tick landing while the
109+ // app is already out of the foreground must skip the sample but keep the chain.
110+ if (currentViewType.isForeground) {
111+ synchronized(pipeline) { pipeline.execute() }
110112 }
111113 } catch (@Suppress(" TooGenericExceptionCaught" : Throwable ) {
112114 internalLogger.log(
@@ -116,24 +118,109 @@ internal class DefaultTimeseriesCollector(
116118 throwable = t
117119 )
118120 } finally {
119- if (isActive(generation)) schedulePipeline(pipeline, generation)
121+ if (state.isGenerationActive(generation)) schedulePipeline(pipeline, generation)
122+ }
123+ }
124+ }
125+
126+ @WorkerThread
127+ private fun flushPipelines () {
128+ pipelines.forEach { pipeline ->
129+ try {
130+ synchronized(pipeline, pipeline::flush)
131+ } catch (@Suppress(" TooGenericExceptionCaught" : Throwable ) {
132+ internalLogger.log(
133+ level = InternalLogger .Level .ERROR ,
134+ targets = listOf (InternalLogger .Target .MAINTAINER , InternalLogger .Target .TELEMETRY ),
135+ messageBuilder = { ERROR_FLUSH_FAILED },
136+ throwable = t
137+ )
138+ }
139+ }
140+ }
141+
142+ // endregion
143+
144+ // region SUSPENSION
145+
146+ private fun scheduleStop (stopRequestGeneration : Int ) {
147+ recentSuspension = scheduledExecutorService.scheduleSafe(
148+ SUSPEND_OPERATION_NAME ,
149+ SUSPEND_DELAY_MS ,
150+ TimeUnit .MILLISECONDS ,
151+ internalLogger
152+ ) {
153+ if (! currentViewType.isForeground && state.stopGeneration(stopRequestGeneration)) {
154+ flushPipelines()
120155 }
121156 }
122157 }
123158
124- private fun isActive (generation : Int ): Boolean =
125- state.get() == State .RUNNING && currentGeneration.get() == generation
159+ private fun cancelSuspension () {
160+ recentSuspension?.cancel(false )
161+ recentSuspension = null
162+ }
163+
164+ // endregion
126165
127166 internal companion object {
128- const val OPERATION_NAME = " Timeseries sampling"
167+ const val TIMESERIES_OPERATION_NAME = " Timeseries sampling"
168+ const val SUSPEND_OPERATION_NAME = " Timeseries suspend"
129169 const val ERROR_SAMPLING_FAILED = " Timeseries sampling iteration failed; rescheduling next sample."
130- const val ERROR_FLUSH_FAILED = " Timeseries flush on session stop failed."
170+ const val ERROR_FLUSH_FAILED = " Timeseries flush failed."
171+
172+ // Matches ActivityViewTrackingStrategy.STOP_VIEW_DELAY_MS, which guards the same race:
173+ // an Activity-to-Activity transition leaves no active view for a moment when the tracking
174+ // strategy stops the view on pause rather than on stop.
175+ const val SUSPEND_DELAY_MS = 200L
176+
131177 val RumViewType ?.isForeground: Boolean
132178 get() = when (this ) {
133179 RumViewType .FOREGROUND , RumViewType .APPLICATION_LAUNCH -> true
134180 RumViewType .BACKGROUND -> false
135181 RumViewType .NONE -> false
136182 null -> false
137183 }
184+
185+ private class State {
186+ // Written under the monitor only, but read outside of it, hence @Volatile: a stale
187+ // generation would make the guards below reject a live sampling chain or flush.
188+ @Volatile
189+ var currentGeneration: Int = 0
190+ private set
191+
192+ private var active: Boolean = false
193+
194+ fun isGenerationActive (generation : Int ): Boolean = synchronized(this ) {
195+ currentGeneration == generation && active
196+ }
197+
198+ /* *
199+ * Always starts a fresh generation, so that a suspension pending on the previous one
200+ * can no longer stop the sampling chain. Returns the new generation.
201+ */
202+ fun startGeneration (): Int = synchronized(this ) {
203+ active = true
204+ ++ currentGeneration
205+ }
206+
207+ /* *
208+ * Deactivates [generation] if it is still the current one.
209+ * Returns true when this call is the one that deactivated it.
210+ */
211+ fun stopGeneration (generation : Int ): Boolean = synchronized(this ) {
212+ currentGeneration == generation && set(false ) != null
213+ }
214+
215+ /* * Returns the new generation if this call changed the state, null otherwise. */
216+ fun set (isActive : Boolean ): Int? = synchronized(this ) {
217+ if (isActive == active) {
218+ null
219+ } else {
220+ active = isActive
221+ ++ currentGeneration
222+ }
223+ }
224+ }
138225 }
139226}
0 commit comments