Merge pull request #2631 from DataDog/nogorodnikov/rum-9854/introduce-event-processing-thread

RUM-9854: Introduce event processing thread

Author: 0xnm

dd-sdk-android-core/api/apiSurface

@@ -108,9 +108,11 @@ interface com.datadog.android.api.feature.FeatureEventReceiver
   fun onReceive(Any)
 interface com.datadog.android.api.feature.FeatureScope
   val dataStore: com.datadog.android.api.storage.datastore.DataStoreHandler
-  fun withWriteContext((com.datadog.android.api.context.DatadogContext, com.datadog.android.api.storage.EventBatchWriter) -> Unit)
+  fun withWriteContext((com.datadog.android.api.context.DatadogContext) -> Unit)
+  fun getWriteContextSync(): Pair<com.datadog.android.api.context.DatadogContext, EventWriteScope>?
   fun sendEvent(Any)
   fun <T: Feature> unwrap(): T
+typealias EventWriteScope = ((com.datadog.android.api.storage.EventBatchWriter) -> Unit) -> Unit
 fun <R: Any?> com.datadog.android.api.InternalLogger.measureMethodCallPerf(Class<*>, String, Float = 100f, () -> R): R
 interface com.datadog.android.api.feature.FeatureSdkCore : com.datadog.android.api.SdkCore
   val internalLogger: com.datadog.android.api.InternalLogger
@@ -308,6 +310,7 @@ fun Collection<ByteArray>.join(ByteArray, ByteArray = ByteArray(0), ByteArray =
 fun java.util.concurrent.Executor.executeSafe(String, com.datadog.android.api.InternalLogger, Runnable)
 fun java.util.concurrent.ScheduledExecutorService.scheduleSafe(String, Long, java.util.concurrent.TimeUnit, com.datadog.android.api.InternalLogger, Runnable): java.util.concurrent.ScheduledFuture<*>?
 fun java.util.concurrent.ExecutorService.submitSafe(String, com.datadog.android.api.InternalLogger, Runnable): java.util.concurrent.Future<*>?
+fun <T> java.util.concurrent.ExecutorService.submitSafe(String, com.datadog.android.api.InternalLogger, java.util.concurrent.Callable<T>): java.util.concurrent.Future<T>?
 val NULL_MAP_VALUE: Object
 object com.datadog.android.core.internal.utils.JsonSerializer
   fun toJsonElement(Any?): com.google.gson.JsonElement

dd-sdk-android-core/api/dd-sdk-android-core.api

@@ -342,6 +342,7 @@ public abstract interface class com/datadog/android/api/feature/FeatureEventRece
 
 public abstract interface class com/datadog/android/api/feature/FeatureScope {
 	public abstract fun getDataStore ()Lcom/datadog/android/api/storage/datastore/DataStoreHandler;
+	public abstract fun getWriteContextSync ()Lkotlin/Pair;
 	public abstract fun sendEvent (Ljava/lang/Object;)V
 	public abstract fun unwrap ()Lcom/datadog/android/api/feature/Feature;
 	public abstract fun withWriteContext (Lkotlin/jvm/functions/Function2;)V
@@ -812,6 +813,7 @@ public final class com/datadog/android/core/internal/utils/ConcurrencyExtKt {
 	public static final fun executeSafe (Ljava/util/concurrent/Executor;Ljava/lang/String;Lcom/datadog/android/api/InternalLogger;Ljava/lang/Runnable;)V
 	public static final fun scheduleSafe (Ljava/util/concurrent/ScheduledExecutorService;Ljava/lang/String;JLjava/util/concurrent/TimeUnit;Lcom/datadog/android/api/InternalLogger;Ljava/lang/Runnable;)Ljava/util/concurrent/ScheduledFuture;
 	public static final fun submitSafe (Ljava/util/concurrent/ExecutorService;Ljava/lang/String;Lcom/datadog/android/api/InternalLogger;Ljava/lang/Runnable;)Ljava/util/concurrent/Future;
+	public static final fun submitSafe (Ljava/util/concurrent/ExecutorService;Ljava/lang/String;Lcom/datadog/android/api/InternalLogger;Ljava/util/concurrent/Callable;)Ljava/util/concurrent/Future;
 }
 
 public final class com/datadog/android/core/internal/utils/JsonSerializer {

dd-sdk-android-core/src/main/kotlin/com/datadog/android/api/feature/FeatureScope.kt

@@ -10,6 +10,7 @@ import androidx.annotation.AnyThread
 import com.datadog.android.api.context.DatadogContext
 import com.datadog.android.api.storage.EventBatchWriter
 import com.datadog.android.api.storage.datastore.DataStoreHandler
+import com.datadog.android.lint.InternalApi
 
 /**
  * Represents a Datadog feature.
@@ -24,15 +25,24 @@ interface FeatureScope {
     /**
      * Utility to write an event, asynchronously.
      * @param callback an operation called with an up-to-date [DatadogContext]
-     * and an [EventBatchWriter]. Callback will be executed on a worker thread from I/O pool.
-     * [DatadogContext] will have a state created at the moment this method is called, before the
-     * thread switch for the callback invocation.
+     * and an [EventWriteScope]. Callback will be executed on a single context processing worker thread.
+     * [DatadogContext] will have a state created at the moment this method is called.
      */
     @AnyThread
     fun withWriteContext(
-        callback: (DatadogContext, EventBatchWriter) -> Unit
+        callback: (datadogContext: DatadogContext, write: EventWriteScope) -> Unit
     )
 
+    // TODO RUM-9852 Implement better passthrough mechanism for the JVM crash scenario
+    /**
+     * Same as [withWriteContext] but will be executed in the blocking manner.
+     *
+     * **NOTE**: This API is for the internal use only and is not guaranteed to be stable.
+     */
+    @AnyThread
+    @InternalApi
+    fun getWriteContextSync(): Pair<DatadogContext, EventWriteScope>?
+
     /**
      * Send event to a given feature. It will be sent in a synchronous way.
      *
@@ -45,3 +55,9 @@ interface FeatureScope {
      */
     fun <T : Feature> unwrap(): T
 }
+
+/**
+ * Scope for the event write operation which is invoked on the worker thread from I/O pool, which is different
+ * from the context processing worker thread used for [FeatureScope.withWriteContext] callback invocation.
+ */
+typealias EventWriteScope = ((EventBatchWriter) -> Unit) -> Unit

dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/CoreFeature.kt

@@ -59,6 +59,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.DatadogThreadFactory
 import com.datadog.android.core.internal.thread.LoggingScheduledThreadPoolExecutor
 import com.datadog.android.core.internal.thread.ScheduledExecutorServiceFactory
 import com.datadog.android.core.internal.time.AppStartTimeProvider
@@ -97,8 +98,10 @@ import java.lang.ref.WeakReference
 import java.util.Locale
 import java.util.concurrent.ConcurrentHashMap
 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
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicBoolean
 
@@ -144,6 +147,7 @@ internal class CoreFeature(
 
     internal lateinit var uploadExecutorService: ScheduledThreadPoolExecutor
     internal lateinit var persistenceExecutorService: FlushableExecutorService
+    internal lateinit var contextExecutorService: ThreadPoolExecutor
     internal lateinit var backpressureStrategy: BackPressureStrategy
 
     internal var localDataEncryption: Encryption? = null
@@ -286,6 +290,7 @@ internal class CoreFeature(
     fun drainAndShutdownExecutors() {
         val tasks = arrayListOf<Runnable>()
 
+        contextExecutorService.queue.drainTo(tasks)
         persistenceExecutorService.drainTo(tasks)
 
         uploadExecutorService
@@ -295,9 +300,11 @@ internal class CoreFeature(
         // we need to make sure we drain the runnable list in both executors first
         // then we shut them down by using the await termination method to make sure we block
         // the thread until the active task is finished.
+        contextExecutorService.shutdown()
         persistenceExecutorService.shutdown()
         uploadExecutorService.shutdown()
 
+        contextExecutorService.awaitTermination(DRAIN_WAIT_SECONDS, TimeUnit.SECONDS)
         persistenceExecutorService.awaitTermination(DRAIN_WAIT_SECONDS, TimeUnit.SECONDS)
         uploadExecutorService.awaitTermination(DRAIN_WAIT_SECONDS, TimeUnit.SECONDS)
 
@@ -613,6 +620,19 @@ internal class CoreFeature(
             executorContext = "storage",
             backPressureStrategy = backpressureStrategy
         )
+        // TODO RUM-9851 Switch to the executor which is aware of backpressure, but only logs it
+        @Suppress("UnsafeThirdPartyFunctionCall") // all parameters are safe
+        contextExecutorService = ThreadPoolExecutor(
+            // core pool size
+            1,
+            // max pool size,
+            1,
+            // keep-alive time
+            0L,
+            TimeUnit.MILLISECONDS,
+            LinkedBlockingQueue(),
+            DatadogThreadFactory("context")
+        )
     }
 
     private fun resolveProcessInfo(appContext: Context) {
@@ -637,10 +657,12 @@ internal class CoreFeature(
 
     private fun shutDownExecutors() {
         uploadExecutorService.shutdownNow()
+        contextExecutorService.shutdownNow()
         persistenceExecutorService.shutdownNow()
 
         try {
             uploadExecutorService.awaitTermination(1, TimeUnit.SECONDS)
+            contextExecutorService.awaitTermination(1, TimeUnit.SECONDS)
             persistenceExecutorService.awaitTermination(1, TimeUnit.SECONDS)
         } catch (e: InterruptedException) {
             try {

dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/DatadogCore.kt

@@ -179,12 +179,14 @@ internal class DatadogCore(
     /** @inheritDoc */
     @AnyThread
     override fun clearAllData() {
-        features.values.forEach {
-            it.clearAllData()
-        }
-        getPersistenceExecutorService().executeSafe("Clear all data", internalLogger) {
-            coreFeature.deleteLastViewEvent()
-            coreFeature.deleteLastFatalAnrSent()
+        coreFeature.contextExecutorService.executeSafe("DatadogCore.clearAllData", internalLogger) {
+            features.values.forEach {
+                it.clearAllData()
+            }
+            getPersistenceExecutorService().executeSafe("Clear all data", internalLogger) {
+                coreFeature.deleteLastViewEvent()
+                coreFeature.deleteLastFatalAnrSent()
+            }
         }
     }
 

dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/SdkFeature.kt

@@ -12,13 +12,13 @@ import androidx.annotation.AnyThread
 import androidx.annotation.WorkerThread
 import com.datadog.android.api.InternalLogger
 import com.datadog.android.api.context.DatadogContext
+import com.datadog.android.api.feature.EventWriteScope
 import com.datadog.android.api.feature.Feature
 import com.datadog.android.api.feature.FeatureContextUpdateReceiver
 import com.datadog.android.api.feature.FeatureEventReceiver
 import com.datadog.android.api.feature.FeatureScope
 import com.datadog.android.api.feature.StorageBackedFeature
 import com.datadog.android.api.net.RequestFactory
-import com.datadog.android.api.storage.EventBatchWriter
 import com.datadog.android.api.storage.FeatureStorageConfiguration
 import com.datadog.android.api.storage.datastore.DataStoreHandler
 import com.datadog.android.core.configuration.UploadSchedulerStrategy
@@ -52,14 +52,19 @@ import com.datadog.android.core.internal.persistence.file.NoOpFileOrchestrator
 import com.datadog.android.core.internal.persistence.file.advanced.FeatureFileOrchestrator
 import com.datadog.android.core.internal.persistence.file.batch.BatchFileReaderWriter
 import com.datadog.android.core.internal.persistence.tlvformat.TLVBlockFileReader
+import com.datadog.android.core.internal.utils.executeSafe
+import com.datadog.android.core.internal.utils.submitSafe
 import com.datadog.android.core.persistence.PersistenceStrategy
 import com.datadog.android.internal.profiler.BenchmarkSdkUploads
 import com.datadog.android.internal.profiler.GlobalBenchmark
 import com.datadog.android.privacy.TrackingConsentProviderCallback
 import com.datadog.android.security.Encryption
 import java.util.Collections
 import java.util.Locale
+import java.util.concurrent.Callable
+import java.util.concurrent.CancellationException
 import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ExecutionException
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicReference
 
@@ -169,16 +174,43 @@ internal class SdkFeature(
     // region FeatureScope
 
     override fun withWriteContext(
-        callback: (DatadogContext, EventBatchWriter) -> Unit
+        callback: (DatadogContext, EventWriteScope) -> Unit
     ) {
-        // TODO RUM-1462 thread safety. Thread switch happens in Storage right now. Open questions:
-        // * what if caller wants to have a sync operation, without thread switch
-        // * should context read and write be on the dedicated thread? risk - time gap between
-        // caller and context
-        val contextProvider = coreFeature.contextProvider
-        if (contextProvider is NoOpContextProvider) return
-        val context = contextProvider.context
-        storage.writeCurrentBatch(context) { callback(context, it) }
+        coreFeature.contextExecutorService
+            .executeSafe("withWriteContext-${wrappedFeature.name}", internalLogger) {
+                val contextProvider = coreFeature.contextProvider
+                if (contextProvider is NoOpContextProvider) return@executeSafe
+                val context = contextProvider.context
+                val eventBatchWriteScope = storage.getEventWriteScope(context)
+                callback(context, eventBatchWriteScope)
+            }
+    }
+
+    override fun getWriteContextSync(): Pair<DatadogContext, EventWriteScope>? {
+        val future = coreFeature.contextExecutorService
+            .submitSafe(
+                "getWriteContextSync-${wrappedFeature.name}",
+                internalLogger,
+                Callable {
+                    val contextProvider = coreFeature.contextProvider
+                    if (contextProvider is NoOpContextProvider) return@Callable null
+                    val context = contextProvider.context
+                    val eventBatchWriteScope = storage.getEventWriteScope(context)
+                    context to eventBatchWriteScope
+                }
+            )
+        return try {
+            future?.get()
+        } catch (e: CancellationException) {
+            logGetWriteContextSyncError(e)
+            null
+        } catch (e: ExecutionException) {
+            logGetWriteContextSyncError(e)
+            null
+        } catch (e: InterruptedException) {
+            logGetWriteContextSyncError(e)
+            null
+        }
     }
 
     override fun sendEvent(event: Any) {
@@ -429,6 +461,15 @@ internal class SdkFeature(
         )
     }
 
+    private fun logGetWriteContextSyncError(e: Exception) {
+        internalLogger.log(
+            level = InternalLogger.Level.ERROR,
+            target = InternalLogger.Target.USER,
+            { FAILED_TO_GET_WRITE_CONTEXT_SYNC },
+            e
+        )
+    }
+
     // endregion
 
     // Used for nightly tests only
@@ -452,6 +493,7 @@ internal class SdkFeature(
             "Feature \"%s\" already has this listener registered."
         const val NO_EVENT_RECEIVER =
             "Feature \"%s\" has no event receiver registered, ignoring event."
+        internal const val FAILED_TO_GET_WRITE_CONTEXT_SYNC = "Failed to get write context in a sync manner."
         internal const val TRACK_NAME = "track"
         internal const val METER_NAME = "dd-sdk-android"
         internal const val BATCH_COUNT_METRIC_NAME = "android.benchmark.batch_count"

dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/persistence/AbstractStorage.kt

@@ -10,6 +10,7 @@ import androidx.annotation.AnyThread
 import androidx.annotation.WorkerThread
 import com.datadog.android.api.InternalLogger
 import com.datadog.android.api.context.DatadogContext
+import com.datadog.android.api.feature.EventWriteScope
 import com.datadog.android.api.storage.EventBatchWriter
 import com.datadog.android.api.storage.EventType
 import com.datadog.android.api.storage.FeatureStorageConfiguration
@@ -49,6 +50,8 @@ internal class AbstractStorage(
         )
     }
 
+    private val writeLock = Any()
+
     private val notGrantedPersistenceStrategy: PersistenceStrategy = NoOpPersistenceStrategy()
 
     init {
@@ -59,28 +62,26 @@ internal class AbstractStorage(
     // region Storage
 
     @AnyThread
-    override fun writeCurrentBatch(
-        datadogContext: DatadogContext,
-        callback: (EventBatchWriter) -> Unit
-    ) {
-        executorService.executeSafe("Data write", internalLogger) {
-            val strategy = resolvePersistenceStrategy(datadogContext)
-            val writer = object : EventBatchWriter {
-                @WorkerThread
-                override fun currentMetadata(): ByteArray? {
-                    return strategy.currentMetadata()
-                }
+    override fun getEventWriteScope(
+        datadogContext: DatadogContext
+    ): EventWriteScope {
+        val strategy = resolvePersistenceStrategy(datadogContext)
+        val writer = object : EventBatchWriter {
+            @WorkerThread
+            override fun currentMetadata(): ByteArray? {
+                return strategy.currentMetadata()
+            }
 
-                @WorkerThread
-                override fun write(event: RawBatchEvent, batchMetadata: ByteArray?, eventType: EventType): Boolean {
-                    return strategy.write(event, batchMetadata, eventType)
-                }
+            @WorkerThread
+            override fun write(event: RawBatchEvent, batchMetadata: ByteArray?, eventType: EventType): Boolean {
+                return strategy.write(event, batchMetadata, eventType)
             }
-            callback.invoke(writer)
         }
+        // although we don't know what storage is backed by the persistence strategy, so maybe writing in a concurrent
+        // way is fine there and lock is not needed, but taking precautions
+        return AsyncEventWriteScope(executorService, writer, writeLock, featureName, internalLogger)
     }
 
-    @WorkerThread
     private fun resolvePersistenceStrategy(datadogContext: DatadogContext) =
         when (datadogContext.trackingConsent) {
             TrackingConsent.GRANTED -> grantedPersistenceStrategy

dd-sdk-android-core/src/main/kotlin/com/datadog/android/core/internal/persistence/AsyncEventWriteScope.kt

@@ -0,0 +1,31 @@
+/*
+ * 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.persistence
+
+import com.datadog.android.api.InternalLogger
+import com.datadog.android.api.feature.EventWriteScope
+import com.datadog.android.api.storage.EventBatchWriter
+import com.datadog.android.core.internal.utils.executeSafe
+import java.util.concurrent.Executor
+
+internal class AsyncEventWriteScope(
+    private val executor: Executor,
+    private val writer: EventBatchWriter,
+    private val featureWriteLock: Any,
+    private val featureName: String,
+    private val internalLogger: InternalLogger
+) : EventWriteScope {
+    override fun invoke(block: (EventBatchWriter) -> Unit) {
+        executor.executeSafe("eventWriteScopeInvoke-$featureName", internalLogger) {
+            // since writing may not be atomic: we can write batch data + batch metadata, there is a gap between
+            // getting file for writing and write op, we sync file operation with a feature-wide lock
+            synchronized(featureWriteLock) {
+                block.invoke(writer)
+            }
+        }
+    }
+}