diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/BootstrapDispatcherPools.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/BootstrapDispatcherPools.kt new file mode 100644 index 0000000000..be66d1c6e6 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/BootstrapDispatcherPools.kt @@ -0,0 +1,414 @@ +package com.onesignal.common.threading + +import com.onesignal.debug.internal.logging.Logging +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.asCoroutineDispatcher +import kotlinx.coroutines.cancel +import java.util.concurrent.CancellationException +import java.util.concurrent.ConcurrentLinkedQueue +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.ThreadFactory +import java.util.concurrent.ThreadPoolExecutor +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicInteger +import kotlin.coroutines.CoroutineContext + +private const val BASE = OneSignalDispatchers.BASE_THREAD_NAME + +// Optimized pool sizes based on CPU cores and workload analysis +private const val IO_CORE_POOL_SIZE = 2 // Increased for better concurrency +private const val IO_MAX_POOL_SIZE = 3 // Increased for better concurrency +private const val DEFAULT_CORE_POOL_SIZE = 2 // Optimal for CPU operations +private const val DEFAULT_MAX_POOL_SIZE = 3 // Slightly larger for CPU operations +private const val KEEP_ALIVE_TIME_SECONDS = + 30L // Keep threads alive longer to reduce recreation +private const val QUEUE_CAPACITY = + 200 // Increased to handle more queued operations during init, while still preventing memory bloat +private const val IO_THREAD_NAME_PREFIX = + "$BASE-IO" // Thread name prefix for I/O operations +private const val DEFAULT_THREAD_NAME_PREFIX = + "$BASE-Default" // Thread name prefix for CPU operations +private const val SERIAL_IO_THREAD_NAME = + "$BASE-SerialIO" // Single, named thread for order-sensitive work +private const val INGRESS_THREAD_NAME = "$BASE-Ingress" + +private class OptimizedThreadFactory( + private val namePrefix: String, + private val priority: Int = Thread.NORM_PRIORITY, +) : ThreadFactory { + private val threadNumber = AtomicInteger(1) + + override fun newThread(r: Runnable): Thread { + val thread = Thread(r, "$namePrefix-${threadNumber.getAndIncrement()}") + thread.isDaemon = true + thread.priority = priority + return thread + } +} + +private enum class Lane { + IO, + DEFAULT, + SERIAL_IO, + INGRESS, +} + +private enum class LaneState { + COLD, + STARTING, + DRAINING, + READY, + CLOSED, +} + +private class LaneTarget( + val dispatcher: CoroutineDispatcher, + val executor: ThreadPoolExecutor?, +) { + fun close() { + executor?.shutdownNow() + } +} + +private data class LaneConfig( + val corePoolSize: Int, + val maxPoolSize: Int, + val threadName: String, + val priority: Int = Thread.NORM_PRIORITY, + val queueCapacity: Int = QUEUE_CAPACITY, +) + +/** + * A stable dispatcher that queues work until its backing executor has been built and + * prestarted on the bootstrap thread. CoroutineScope.launch can therefore return its real Job + * without forcing executor construction or waiting on a lazy lock on the caller thread. + */ +@Suppress("TooManyFunctions") +private class GateDispatcher( + private val lane: Lane, + private val requestBootstrap: (GateDispatcher) -> Unit, +) : CoroutineDispatcher(), DispatcherLane { + override val dispatcher: CoroutineDispatcher get() = this + + private val lock = Any() + private val pending = ArrayDeque>() + private var state = LaneState.COLD + private var target: LaneTarget? = null + + override fun dispatch( + context: CoroutineContext, + block: Runnable, + ) { + var readyDispatcher: CoroutineDispatcher? = null + var shouldBootstrap = false + var rejected = false + + synchronized(lock) { + when (state) { + LaneState.COLD -> { + pending.addLast(context to block) + state = LaneState.STARTING + shouldBootstrap = true + } + LaneState.STARTING, LaneState.DRAINING -> pending.addLast(context to block) + LaneState.READY -> readyDispatcher = target!!.dispatcher + LaneState.CLOSED -> rejected = true + } + } + + when { + shouldBootstrap -> requestBootstrap(this) + readyDispatcher != null -> dispatchSafely(readyDispatcher!!, context, block) + rejected -> cancelAndComplete(context, block, "OneSignal dispatcher generation is closed") + } + } + + fun requestWarmup() { + val shouldBootstrap = + synchronized(lock) { + if (state == LaneState.COLD) { + state = LaneState.STARTING + true + } else { + false + } + } + if (shouldBootstrap) requestBootstrap(this) + } + + @Suppress("TooGenericExceptionCaught") + fun initialize() { + createTargetOrFallback()?.let(::drainPending) + } + + // Catches Throwable because thread creation in prestartAllCoreThreads fails with OutOfMemoryError, + // which is exactly the case the Dispatchers fallback exists for. + @Suppress("TooGenericExceptionCaught") + private fun createTargetOrFallback(): LaneTarget? = + try { + createTarget(lane) + } catch (t: Throwable) { + OneSignalDispatchers.logFailure("$lane fell back", t) + try { + createFallbackTarget(lane) + } catch (fallbackError: Throwable) { + OneSignalDispatchers.logFailure("$lane fallback failed", fallbackError) + failPending("OneSignal $lane dispatcher fallback failed") + null + } + } + + private fun drainPending(newTarget: LaneTarget) { + while (true) { + val next = + synchronized(lock) { + if (state == LaneState.CLOSED) { + null + } else { + state = LaneState.DRAINING + pending.removeFirstOrNull().also { + if (it == null) { + target = newTarget + state = LaneState.READY + } + } + } + } + + if (next == null) { + val closed = synchronized(lock) { state == LaneState.CLOSED } + if (closed) newTarget.close() + return + } + dispatchSafely(newTarget.dispatcher, next.first, next.second) + } + } + + fun close() { + val queued: List> + val oldTarget: LaneTarget? + synchronized(lock) { + if (state == LaneState.CLOSED) return + state = LaneState.CLOSED + queued = pending.toList() + pending.clear() + oldTarget = target + target = null + } + queued.forEach { + cancelAndComplete(it.first, it.second, "OneSignal dispatcher generation was reset") + } + oldTarget?.close() + } + + fun bootstrapFailed() { + failPending("OneSignal dispatcher bootstrap thread failed to start") + } + + override fun status(): String = + synchronized(lock) { + when (state) { + LaneState.READY -> if (target?.executor == null) "Fallback" else "Active" + LaneState.CLOSED -> "Shutdown" + else -> state.name.lowercase().replaceFirstChar { it.uppercase() } + } + } + + override fun executor(): ThreadPoolExecutor? = synchronized(lock) { target?.executor } + + private fun failPending(reason: String) { + val queued = + synchronized(lock) { + val copy = pending.toList() + pending.clear() + // CLOSED is terminal: reopening would let a discarded generation build an executor + // that no later reset tears down. + if (state != LaneState.CLOSED) state = LaneState.COLD + copy + } + queued.forEach { cancelAndComplete(it.first, it.second, reason) } + } + + private fun dispatchSafely( + dispatcher: CoroutineDispatcher, + context: CoroutineContext, + block: Runnable, + ) { + try { + dispatcher.dispatch(context, block) + } catch (e: RuntimeException) { + Logging.error("OneSignalDispatchers: $lane dispatch rejected: ${e.message}", e) + cancelAndComplete(context, block, "OneSignal $lane dispatch rejected") + } + } + + private fun cancelAndComplete( + context: CoroutineContext, + block: Runnable, + reason: String, + ) { + context[Job]?.cancel(CancellationException(reason)) + // Mirrors kotlinx's rejected-execution handling: the caller may be the main thread, + // so the cancelled continuation's catch/finally must not run inline here. + Dispatchers.IO.dispatch(context, block) + } +} + +private class BootstrapCoordinator { + private val queue = ConcurrentLinkedQueue() + private val running = AtomicBoolean(false) + + fun request(gate: GateDispatcher) { + queue.add(gate) + startIfNeeded() + } + + @Suppress("TooGenericExceptionCaught") + private fun startIfNeeded() { + if (!running.compareAndSet(false, true)) return + try { + Thread( + ::runLoop, + "$BASE-bootstrap", + ).apply { + isDaemon = true + // Not lowered: main can block waiting for init, which queues behind these builds. + priority = Thread.NORM_PRIORITY + start() + } + } catch (t: Throwable) { + running.set(false) + OneSignalDispatchers.logFailure("bootstrap thread failed to start", t) + while (true) { + val gate = queue.poll() ?: break + gate.bootstrapFailed() + } + } + } + + @Suppress("TooGenericExceptionCaught") + private fun runLoop() { + try { + while (true) { + val gate = queue.poll() ?: return + try { + gate.initialize() + } catch (t: Throwable) { + OneSignalDispatchers.logFailure("bootstrap lane failed", t) + gate.bootstrapFailed() + } + } + } finally { + running.set(false) + if (queue.isNotEmpty()) startIfNeeded() + } + } +} + +internal class BootstrapDispatcherPools : DispatcherPools { + private val coordinator = BootstrapCoordinator() + private val ioGate = GateDispatcher(Lane.IO, coordinator::request) + private val defaultGate = GateDispatcher(Lane.DEFAULT, coordinator::request) + private val serialGate = GateDispatcher(Lane.SERIAL_IO, coordinator::request) + private val ingressGate = GateDispatcher(Lane.INGRESS, coordinator::request) + override val io: DispatcherLane = ioGate + override val defaultLane: DispatcherLane = defaultGate + override val serialIO: DispatcherLane = serialGate + override val ingress: DispatcherLane = ingressGate + override val ioScope = CoroutineScope(SupervisorJob() + ioGate) + override val defaultScope = CoroutineScope(SupervisorJob() + defaultGate) + override val serialIOScope = CoroutineScope(SupervisorJob() + serialGate) + override val ingressScope = CoroutineScope(SupervisorJob() + ingressGate) + + // Ingress bootstraps first because receivers have a ~10s broadcast budget. + private val gates = listOf(ingressGate, ioGate, defaultGate, serialGate) + private val scopes = listOf(ingressScope, ioScope, defaultScope, serialIOScope) + + override fun prewarm() { + gates.forEach { it.requestWarmup() } + } + + override fun shutdown() { + scopes.forEach { it.cancel() } + gates.forEach { it.close() } + } +} + +@Suppress("TooGenericExceptionCaught") +private fun createTarget(lane: Lane): LaneTarget { + OneSignalDispatchers.beforeLaneCreateForTest?.invoke(lane.name) + val config = lane.config() + val executor = + ThreadPoolExecutor( + config.corePoolSize, + config.maxPoolSize, + KEEP_ALIVE_TIME_SECONDS, + TimeUnit.SECONDS, + LinkedBlockingQueue(config.queueCapacity), + OptimizedThreadFactory(config.threadName, config.priority), + ) + try { + executor.allowCoreThreadTimeOut(false) + OneSignalDispatchers.afterExecutorCreateForTest?.invoke(lane.name, executor) + executor.prestartAllCoreThreads() + } catch (t: Throwable) { + executor.shutdownNow() + throw t + } + return LaneTarget(executor.asCoroutineDispatcher(), executor) +} + +private fun Lane.config(): LaneConfig = + when (this) { + Lane.IO -> + LaneConfig( + IO_CORE_POOL_SIZE, + IO_MAX_POOL_SIZE, + IO_THREAD_NAME_PREFIX, + Thread.NORM_PRIORITY - 1, + ) + Lane.DEFAULT -> + LaneConfig( + DEFAULT_CORE_POOL_SIZE, + DEFAULT_MAX_POOL_SIZE, + DEFAULT_THREAD_NAME_PREFIX, + ) + Lane.SERIAL_IO -> + LaneConfig( + 1, + 1, + SERIAL_IO_THREAD_NAME, + Thread.NORM_PRIORITY - 1, + // Unbounded like Executors.newSingleThreadExecutor; rejecting would drop ordered lifecycle work. + Int.MAX_VALUE, + ) + // Kept separate from IO so receiver handoffs never queue behind SDK HTTP. + Lane.INGRESS -> + LaneConfig( + 1, + 1, + INGRESS_THREAD_NAME, + Thread.NORM_PRIORITY - 1, + Int.MAX_VALUE, + ) + } + +private fun createFallbackTarget(lane: Lane): LaneTarget { + OneSignalDispatchers.beforeFallbackCreateForTest?.invoke(lane.name) + val dispatcher = + when (lane) { + Lane.IO -> Dispatchers.IO + Lane.DEFAULT -> Dispatchers.Default + Lane.SERIAL_IO, Lane.INGRESS -> { + @Suppress("OPT_IN_USAGE") + Dispatchers.IO.limitedParallelism(1) + } + } + dispatcher.dispatch(kotlin.coroutines.EmptyCoroutineContext, Runnable {}) + return LaneTarget(dispatcher, null) +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherChoiceReporter.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherChoiceReporter.kt new file mode 100644 index 0000000000..3a4d472b63 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherChoiceReporter.kt @@ -0,0 +1,8 @@ +package com.onesignal.common.threading + +import com.onesignal.core.internal.startup.IStartableService + +/** Sends a degraded pool choice to remote telemetry, which init installs before services start. */ +internal class DispatcherChoiceReporter : IStartableService { + override fun start() = OneSignalDispatchers.reportChoiceToTelemetry() +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherPools.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherPools.kt new file mode 100644 index 0000000000..c8d665bf3c --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherPools.kt @@ -0,0 +1,28 @@ +package com.onesignal.common.threading + +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.CoroutineScope +import java.util.concurrent.ThreadPoolExecutor + +internal interface DispatcherLane { + val dispatcher: CoroutineDispatcher + + fun executor(): ThreadPoolExecutor? + + fun status(): String +} + +internal interface DispatcherPools { + val io: DispatcherLane + val defaultLane: DispatcherLane + val serialIO: DispatcherLane + val ingress: DispatcherLane + val ioScope: CoroutineScope + val defaultScope: CoroutineScope + val serialIOScope: CoroutineScope + val ingressScope: CoroutineScope + + fun prewarm() + + fun shutdown() +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/LazyDispatcherPools.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/LazyDispatcherPools.kt new file mode 100644 index 0000000000..5fdd301f96 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/LazyDispatcherPools.kt @@ -0,0 +1,238 @@ +package com.onesignal.common.threading + +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.asCoroutineDispatcher +import kotlinx.coroutines.cancel +import kotlinx.coroutines.launch +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.ThreadFactory +import java.util.concurrent.ThreadPoolExecutor +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicReference + +/** + * Pre-bootstrap pools. The first dispatch constructs the executor on the caller thread. + * [prewarm] tries to do that on a daemon thread, and a faster dispatch can still win. + */ +@Suppress("TooGenericExceptionCaught") +internal class LazyDispatcherPools : DispatcherPools { + private val closed = AtomicBoolean(false) + private val ioLane = + LazyExecutorLane("IO", IO_CORE, IO_MAX, QUEUE_CAPACITY, "$BASE-IO", Thread.NORM_PRIORITY - 1) { + Dispatchers.IO + } + private val defaultExecutorLane = + LazyExecutorLane("DEFAULT", DEFAULT_CORE, DEFAULT_MAX, QUEUE_CAPACITY, "$BASE-Default", Thread.NORM_PRIORITY) { + Dispatchers.Default + } + private val serialLane = + LazyExecutorLane("SERIAL_IO", 1, 1, Int.MAX_VALUE, "$BASE-SerialIO", Thread.NORM_PRIORITY - 1) { + @Suppress("OPT_IN_USAGE") + Dispatchers.IO.limitedParallelism(1) + } + private val ingressLane = + LazyExecutorLane("INGRESS", 1, 1, Int.MAX_VALUE, "$BASE-Ingress", Thread.NORM_PRIORITY - 1) { + @Suppress("OPT_IN_USAGE") + Dispatchers.IO.limitedParallelism(1) + } + + override val io: DispatcherLane = ioLane + override val defaultLane: DispatcherLane = defaultExecutorLane + override val serialIO: DispatcherLane = serialLane + override val ingress: DispatcherLane = ingressLane + + private val ioScopeLazy = lazy { CoroutineScope(SupervisorJob() + ioLane.dispatcher) } + private val defaultScopeLazy = lazy { CoroutineScope(SupervisorJob() + defaultExecutorLane.dispatcher) } + private val serialScopeLazy = lazy { CoroutineScope(SupervisorJob() + serialLane.dispatcher) } + private val ingressScopeLazy = lazy { CoroutineScope(SupervisorJob() + ingressLane.dispatcher) } + + override val ioScope: CoroutineScope get() = ioScopeLazy.value + override val defaultScope: CoroutineScope get() = defaultScopeLazy.value + override val serialIOScope: CoroutineScope get() = serialScopeLazy.value + override val ingressScope: CoroutineScope get() = ingressScopeLazy.value + + private val prewarmThread = AtomicReference(null) + + override fun prewarm() { + if (closed.get()) return + val thread = + Thread( + { + // Ingress first: receivers have a ~10s broadcast budget. + warm("INGRESS") { ingressScope } + warm("IO") { ioScope } + warm("DEFAULT") { defaultScope } + warm("SERIAL_IO") { serialIOScope } + }, + "$BASE-prewarm", + ) + thread.isDaemon = true + // Not lowered: a launch on main blocks behind this thread's lazy builds, so it must not be starved. + thread.priority = Thread.NORM_PRIORITY + if (prewarmThread.compareAndSet(null, thread)) startPrewarm(thread) + } + + private fun startPrewarm(thread: Thread) { + if (closed.get()) { + prewarmThread.compareAndSet(thread, null) + return + } + try { + OneSignalDispatchers.beforeLazyPrewarmStartForTest?.invoke() + thread.start() + } catch (t: Throwable) { + // start() failed after the ref was published. Clear it or every retry hits a failed CAS. + prewarmThread.compareAndSet(thread, null) + throw t + } + } + + override fun shutdown() { + closed.set(true) + prewarmThread.getAndSet(null)?.let { thread -> + thread.interrupt() + runCatching { thread.join(PREWARM_JOIN_MS) } + } + listOf(ioScopeLazy, defaultScopeLazy, serialScopeLazy, ingressScopeLazy).forEach { scope -> + if (scope.isInitialized()) runCatching { scope.value.cancel() } + } + listOf(ioLane, defaultExecutorLane, serialLane, ingressLane).forEach { it.shutdown() } + } + + private fun warm( + lane: String, + resolve: () -> CoroutineScope, + ) { + OneSignalDispatchers.beforeLazyWarmForTest?.invoke(lane) + if (closed.get()) return + val scope = resolveScope(resolve) ?: return + if (closed.get()) { + runCatching { scope.cancel() } + } else { + runCatching { scope.launch { } }.onFailure(::logWarmFailure) + } + } + + private fun resolveScope(resolve: () -> CoroutineScope): CoroutineScope? = + try { + resolve() + } catch (t: Throwable) { + logWarmFailure(t) + null + } + + private fun logWarmFailure(t: Throwable) { + if (closed.get()) return + OneSignalDispatchers.logFailure("lazy prewarm failed", t) + } + + private inner class LazyExecutorLane( + private val lane: String, + private val core: Int, + private val max: Int, + private val queueCapacity: Int, + private val threadName: String, + private val priority: Int, + private val fallback: () -> CoroutineDispatcher, + ) : DispatcherLane { + private val executorLazy: Lazy = lazy { createExecutor() } + + private fun createExecutor(): ThreadPoolExecutor? { + if (closed.get()) return null + return try { + buildExecutor() + } catch (t: Throwable) { + OneSignalDispatchers.logFailure("lazy $lane fell back", t) + OneSignalDispatchers.beforeFallbackCreateForTest?.invoke(lane) + null + } + } + + private fun buildExecutor(): ThreadPoolExecutor? { + OneSignalDispatchers.beforeLaneCreateForTest?.invoke(lane) + if (closed.get()) return null + val created = newExecutor() + OneSignalDispatchers.afterExecutorCreateForTest?.invoke(lane, created) + if (closed.get()) created.shutdownNow() + return created.takeUnless { it.isShutdown } + } + + override val dispatcher: CoroutineDispatcher by lazy { + val executor = executorLazy.value + if (executor == null) { + fallback() + } else { + try { + executor.asCoroutineDispatcher() + } catch (t: Throwable) { + OneSignalDispatchers.logFailure("lazy $lane fell back", t) + fallback() + } + } + } + + override fun executor(): ThreadPoolExecutor? = + if (executorLazy.isInitialized()) executorLazy.value else null + + override fun status(): String = + try { + val executor = if (executorLazy.isInitialized()) executorLazy.value else null + when { + closed.get() -> "Shutdown" + executor == null && executorLazy.isInitialized() -> "Fallback" + executor == null -> "Cold" + executor.isShutdown -> "Shutdown" + else -> "Active" + } + } catch (_: Throwable) { + "Cold" + } + + fun shutdown() { + if (!executorLazy.isInitialized()) return + runCatching { executorLazy.value?.shutdownNow() } + } + + private fun newExecutor(): ThreadPoolExecutor = + ThreadPoolExecutor( + core, + max, + KEEP_ALIVE_SECONDS, + TimeUnit.SECONDS, + LinkedBlockingQueue(queueCapacity), + NamedThreadFactory(threadName, priority), + ).apply { + allowCoreThreadTimeOut(false) + } + } + + private class NamedThreadFactory( + private val namePrefix: String, + private val priority: Int, + ) : ThreadFactory { + private val threadNumber = AtomicInteger(1) + + override fun newThread(r: Runnable): Thread { + val thread = Thread(r, "$namePrefix-${threadNumber.getAndIncrement()}") + thread.isDaemon = true + thread.priority = priority + return thread + } + } + + private companion object { + private const val BASE = OneSignalDispatchers.BASE_THREAD_NAME + private const val IO_CORE = 2 + private const val IO_MAX = 3 + private const val DEFAULT_CORE = 2 + private const val DEFAULT_MAX = 3 + private const val QUEUE_CAPACITY = 200 + private const val KEEP_ALIVE_SECONDS = 30L + private const val PREWARM_JOIN_MS = 250L + } +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt index 738fdc22ae..2904d75739 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt @@ -1,24 +1,18 @@ package com.onesignal.common.threading +import android.content.Context +import android.util.Log +import com.onesignal.common.AppContext +import com.onesignal.core.internal.features.FeatureManager +import com.onesignal.debug.LogLevel import com.onesignal.debug.internal.logging.Logging +import com.onesignal.features.FeatureFlag +import com.onesignal.features.StartupFlagsSource import kotlinx.coroutines.CoroutineDispatcher -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job -import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.asCoroutineDispatcher -import kotlinx.coroutines.cancel import kotlinx.coroutines.isActive import kotlinx.coroutines.launch -import java.util.concurrent.CancellationException -import java.util.concurrent.ConcurrentLinkedQueue -import java.util.concurrent.LinkedBlockingQueue -import java.util.concurrent.ThreadFactory import java.util.concurrent.ThreadPoolExecutor -import java.util.concurrent.TimeUnit -import java.util.concurrent.atomic.AtomicBoolean -import java.util.concurrent.atomic.AtomicInteger -import kotlin.coroutines.CoroutineContext /** * Optimized threading manager for the OneSignal SDK. @@ -35,25 +29,10 @@ import kotlin.coroutines.CoroutineContext */ @Suppress("TooManyFunctions", "StringLiteralDuplication") object OneSignalDispatchers { - // Optimized pool sizes based on CPU cores and workload analysis - private const val IO_CORE_POOL_SIZE = 2 // Increased for better concurrency - private const val IO_MAX_POOL_SIZE = 3 // Increased for better concurrency - private const val DEFAULT_CORE_POOL_SIZE = 2 // Optimal for CPU operations - private const val DEFAULT_MAX_POOL_SIZE = 3 // Slightly larger for CPU operations - private const val KEEP_ALIVE_TIME_SECONDS = - 30L // Keep threads alive longer to reduce recreation - private const val QUEUE_CAPACITY = - 200 // Increased to handle more queued operations during init, while still preventing memory bloat private const val TEST_READY_TIMEOUT_MS = 2_000L private const val TEST_READY_POLL_MS = 5L internal const val BASE_THREAD_NAME = "OneSignal" // Base thread name prefix - private const val IO_THREAD_NAME_PREFIX = - "$BASE_THREAD_NAME-IO" // Thread name prefix for I/O operations - private const val DEFAULT_THREAD_NAME_PREFIX = - "$BASE_THREAD_NAME-Default" // Thread name prefix for CPU operations - private const val SERIAL_IO_THREAD_NAME = - "$BASE_THREAD_NAME-SerialIO" // Single, named thread for order-sensitive work - private const val INGRESS_THREAD_NAME = "$BASE_THREAD_NAME-Ingress" + private const val RELEASE_LOG_TAG = "OneSignal" @Volatile internal var beforeLaneCreateForTest: ((String) -> Unit)? = null @@ -64,400 +43,203 @@ object OneSignalDispatchers { @Volatile internal var beforeFallbackCreateForTest: ((String) -> Unit)? = null - private class OptimizedThreadFactory( - private val namePrefix: String, - private val priority: Int = Thread.NORM_PRIORITY, - ) : ThreadFactory { - private val threadNumber = AtomicInteger(1) - - override fun newThread(r: Runnable): Thread { - val thread = Thread(r, "$namePrefix-${threadNumber.getAndIncrement()}") - thread.isDaemon = true - thread.priority = priority - return thread - } - } - - private enum class Lane { - IO, - DEFAULT, - SERIAL_IO, - INGRESS, - } + /** Lazy prewarm, before the lane lazy runs. Blocking here must not deadlock shutdown. */ + @Volatile + internal var beforeLazyWarmForTest: ((String) -> Unit)? = null - private enum class LaneState { - COLD, - STARTING, - DRAINING, - READY, - CLOSED, - } + /** After the lazy prewarm thread is published, before [Thread.start]. */ + @Volatile + internal var beforeLazyPrewarmStartForTest: (() -> Unit)? = null - private class LaneTarget( - val dispatcher: CoroutineDispatcher, - val executor: ThreadPoolExecutor?, - ) { - fun close() { - executor?.shutdownNow() - } - } + @Volatile + internal var beforePoolChoiceForTest: (() -> Unit)? = null - private data class LaneConfig( - val corePoolSize: Int, - val maxPoolSize: Int, - val threadName: String, - val priority: Int = Thread.NORM_PRIORITY, - val queueCapacity: Int = QUEUE_CAPACITY, - ) + private val inFailureLog = ThreadLocal() - /** - * A stable dispatcher that queues work until its backing executor has been built and - * prestarted on the bootstrap thread. CoroutineScope.launch can therefore return its real Job - * without forcing executor construction or waiting on a lazy lock on the caller thread. - */ - private class GateDispatcher( - private val lane: Lane, - private val requestBootstrap: (GateDispatcher) -> Unit, - ) : CoroutineDispatcher() { - private val lock = Any() - private val pending = ArrayDeque>() - private var state = LaneState.COLD - private var target: LaneTarget? = null - - override fun dispatch( - context: CoroutineContext, - block: Runnable, - ) { - var readyDispatcher: CoroutineDispatcher? = null - var shouldBootstrap = false - var rejected = false - - synchronized(lock) { - when (state) { - LaneState.COLD -> { - pending.addLast(context to block) - state = LaneState.STARTING - shouldBootstrap = true - } - LaneState.STARTING, LaneState.DRAINING -> pending.addLast(context to block) - LaneState.READY -> readyDispatcher = target!!.dispatcher - LaneState.CLOSED -> rejected = true - } - } + @Volatile + private var poolsRef: DispatcherPools? = null + private val poolsLock = Any() - when { - shouldBootstrap -> requestBootstrap(this) - readyDispatcher != null -> dispatchSafely(readyDispatcher!!, context, block) - rejected -> cancelAndComplete(context, block, "OneSignal dispatcher generation is closed") - } - } + /** Inside the choice report handoff, between reading readiness and parking the report. */ + @Volatile + internal var duringChoiceHandoffForTest: (() -> Unit)? = null - fun requestWarmup() { - val shouldBootstrap = - synchronized(lock) { - if (state == LaneState.COLD) { - state = LaneState.STARTING - true - } else { - false - } - } - if (shouldBootstrap) requestBootstrap(this) - } + // Guards only the two fields below. Never report while holding it: reporting can launch. + private val choiceReportLock = Any() + private var telemetryReady = false + private var pendingChoiceReport: PoolChoice? = null - @Suppress("TooGenericExceptionCaught") - fun initialize() { - createTargetOrFallback()?.let(::drainPending) + /** Test override. Null reads the cached flag, and no cache means off. */ + @Volatile + internal var bootstrapEnabledForTest: Boolean? = null + + internal fun resetBootstrapInputsForTest() { + AppContext.resetForTest() + bootstrapEnabledForTest = null + synchronized(choiceReportLock) { + telemetryReady = false + pendingChoiceReport = null } + FeatureManager.resetStartupFlagsForTest() + } - // Catches Throwable because thread creation in prestartAllCoreThreads fails with OutOfMemoryError, - // which is exactly the case the Dispatchers fallback exists for. - @Suppress("TooGenericExceptionCaught") - private fun createTargetOrFallback(): LaneTarget? = - try { - createTarget(lane) - } catch (t: Throwable) { - Logging.warn("OneSignalDispatchers: Using fallback for $lane lane: ${t.message}", t) - try { - createFallbackTarget(lane) - } catch (fallbackError: Throwable) { - Logging.warn("OneSignalDispatchers: Fallback failed for $lane lane: ${fallbackError.message}", fallbackError) - failPending("OneSignal $lane dispatcher fallback failed") - null - } - } - - private fun drainPending(newTarget: LaneTarget) { - while (true) { - val next = - synchronized(lock) { - if (state == LaneState.CLOSED) { - null - } else { - state = LaneState.DRAINING - pending.removeFirstOrNull().also { - if (it == null) { - target = newTarget - state = LaneState.READY - } - } - } - } - - if (next == null) { - val closed = synchronized(lock) { state == LaneState.CLOSED } - if (closed) newTarget.close() - return + private fun currentPools(): DispatcherPools { + poolsRef?.let { return it } + // Flag read does disk I/O. Keep it outside the lock so other launches are not stuck behind it. + val choice = choosePoolsOrLazy() + val published = + synchronized(poolsLock) { + val existing = poolsRef + if (existing != null) { + choice.pools.shutdown() + existing + } else { + poolsRef = choice.pools + choice.pools } - dispatchSafely(newTarget.dispatcher, next.first, next.second) - } - } - - fun close() { - val queued: List> - val oldTarget: LaneTarget? - synchronized(lock) { - if (state == LaneState.CLOSED) return - state = LaneState.CLOSED - queued = pending.toList() - pending.clear() - oldTarget = target - target = null } - queued.forEach { - cancelAndComplete(it.first, it.second, "OneSignal dispatcher generation was reset") - } - oldTarget?.close() - } + if (published === choice.pools) recordChoice(choice) + return published + } - fun bootstrapFailed() { - failPending("OneSignal dispatcher bootstrap thread failed to start") + // Every launch in the process goes through this, so a failed choice must not throw. + // Lazy pools need no flags, and they are what the flag being off already means. + // A failure after an "on" snapshot leaves FeatureManager reporting the flag on over lazy + // pools. The "choice failed" report is what tells those launches apart in telemetry. + @Suppress("TooGenericExceptionCaught") + private fun choosePoolsOrLazy(): PoolChoice = + try { + choosePools() + } catch (t: Throwable) { + PoolChoice(LazyDispatcherPools(), "off, choice failed", warn = true, error = t) } - fun status(): String = - synchronized(lock) { - when (state) { - LaneState.READY -> "Active" - LaneState.CLOSED -> "Shutdown" - else -> state.name.lowercase().replaceFirstChar { it.uppercase() } - } + /** Call only after [choice] is published: reporting can log, and log listeners can launch. */ + private fun recordChoice(choice: PoolChoice) { + if (bootstrapEnabledForTest != null) return + logRelease(choice.summary, choice.error, warn = choice.warn) + if (!choice.warn) return + val reportNow = + synchronized(choiceReportLock) { + val ready = telemetryReady + duringChoiceHandoffForTest?.invoke() + if (!ready) pendingChoiceReport = choice + ready } + if (reportNow) reportChoice(choice) + } - fun executor(): ThreadPoolExecutor? = synchronized(lock) { target?.executor } - - private fun failPending(reason: String) { - val queued = - synchronized(lock) { - val copy = pending.toList() - pending.clear() - // CLOSED is terminal: reopening would let a discarded generation build an executor - // that no later reset tears down. - if (state != LaneState.CLOSED) state = LaneState.COLD - copy - } - queued.forEach { cancelAndComplete(it.first, it.second, reason) } - } - - private fun dispatchSafely( - dispatcher: CoroutineDispatcher, - context: CoroutineContext, - block: Runnable, - ) { - try { - dispatcher.dispatch(context, block) - } catch (e: RuntimeException) { - Logging.error("OneSignalDispatchers: $lane dispatch rejected: ${e.message}", e) - cancelAndComplete(context, block, "OneSignal $lane dispatch rejected") + /** Pools are usually chosen before init installs remote telemetry, so degraded choices wait for it. */ + internal fun reportChoiceToTelemetry() { + val pending = + synchronized(choiceReportLock) { + telemetryReady = true + pendingChoiceReport.also { pendingChoiceReport = null } } - } + pending?.let(::reportChoice) + } - private fun cancelAndComplete( - context: CoroutineContext, - block: Runnable, - reason: String, - ) { - context[Job]?.cancel(CancellationException(reason)) - // Mirrors kotlinx's rejected-execution handling: the caller may be the main thread, - // so the cancelled continuation's catch/finally must not run inline here. - Dispatchers.IO.dispatch(context, block) + @Suppress("SwallowedException") + private fun reportChoice(choice: PoolChoice) { + try { + Logging.warn("OneSignalDispatchers: sdk_dispatcher_bootstrap ${choice.summary}", choice.error) + } catch (_: RuntimeException) { + // A telemetry fault must not fail init or the launch that published the pools. } } - private class BootstrapCoordinator { - private val queue = ConcurrentLinkedQueue() - private val running = AtomicBoolean(false) - - fun request(gate: GateDispatcher) { - queue.add(gate) - startIfNeeded() + private fun choosePools(): PoolChoice { + beforePoolChoiceForTest?.invoke() + val override = bootstrapEnabledForTest + return if (override != null) { + PoolChoice(poolsFor(override), "test", false) + } else { + chooseFromCache() } + } - @Suppress("TooGenericExceptionCaught") - private fun startIfNeeded() { - if (!running.compareAndSet(false, true)) return - try { - Thread( - ::runLoop, - "$BASE_THREAD_NAME-bootstrap", - ).apply { - isDaemon = true - priority = Thread.NORM_PRIORITY - 2 - start() - } - } catch (t: Throwable) { - running.set(false) - Logging.warn("OneSignalDispatchers: Failed to start bootstrap thread: ${t.message}", t) - while (true) { - val gate = queue.poll() ?: break - gate.bootstrapFailed() - } + private fun chooseFromCache(): PoolChoice { + val flags = FeatureManager.startupFlags(AppContext::get) + val enabled = flags.isEnabled(FeatureFlag.SDK_DISPATCHER_BOOTSTRAP) + val detail = + when (flags.source) { + StartupFlagsSource.STORE -> if (enabled) "on" else "off, flag not in cache" + StartupFlagsSource.MISSING -> "off, no cache yet" + StartupFlagsSource.UNREADABLE -> "off, cache unreadable" + StartupFlagsSource.NO_STORE -> "off, no application" } - } - - @Suppress("TooGenericExceptionCaught") - private fun runLoop() { - try { - while (true) { - val gate = queue.poll() ?: return - try { - gate.initialize() - } catch (t: Throwable) { - Logging.warn("OneSignalDispatchers: Bootstrap failed for a lane: ${t.message}", t) - gate.bootstrapFailed() - } - } - } finally { - running.set(false) - if (queue.isNotEmpty()) startIfNeeded() - } - } + val degraded = flags.source == StartupFlagsSource.NO_STORE || flags.source == StartupFlagsSource.UNREADABLE + return PoolChoice(poolsFor(enabled), detail, warn = degraded) } - private class Pools { - private val coordinator = BootstrapCoordinator() - val IO = GateDispatcher(Lane.IO, coordinator::request) - val Default = GateDispatcher(Lane.DEFAULT, coordinator::request) - val SerialIO = GateDispatcher(Lane.SERIAL_IO, coordinator::request) - val Ingress = GateDispatcher(Lane.INGRESS, coordinator::request) - val IOScope = CoroutineScope(SupervisorJob() + IO) - val DefaultScope = CoroutineScope(SupervisorJob() + Default) - val SerialIOScope = CoroutineScope(SupervisorJob() + SerialIO) - val IngressScope = CoroutineScope(SupervisorJob() + Ingress) - - // Ingress bootstraps first because receivers have a ~10s broadcast budget. - private val gates = listOf(Ingress, IO, Default, SerialIO) - private val scopes = listOf(IngressScope, IOScope, DefaultScope, SerialIOScope) - - fun prewarm() { - gates.forEach { it.requestWarmup() } - } + private fun poolsFor(enabled: Boolean): DispatcherPools = if (enabled) BootstrapDispatcherPools() else LazyDispatcherPools() - fun shutdown() { - scopes.forEach { it.cancel() } - gates.forEach { it.close() } - } + private class PoolChoice( + val pools: DispatcherPools, + val detail: String, + val warn: Boolean, + val error: Throwable? = null, + ) { + val summary: String get() = "$detail; pools=${if (pools is LazyDispatcherPools) "lazy" else "bootstrap"}" } - @Suppress("TooGenericExceptionCaught") - private fun createTarget(lane: Lane): LaneTarget { - beforeLaneCreateForTest?.invoke(lane.name) - val config = lane.config() - val executor = - ThreadPoolExecutor( - config.corePoolSize, - config.maxPoolSize, - KEEP_ALIVE_TIME_SECONDS, - TimeUnit.SECONDS, - LinkedBlockingQueue(config.queueCapacity), - OptimizedThreadFactory(config.threadName, config.priority), - ) + /** Logcat at the customer's SDK log level, not through [Logging]: Logging can dispatch. */ + @Suppress("SwallowedException") + internal fun logRelease( + message: String, + error: Throwable? = null, + warn: Boolean = error != null, + ) { + val line = "sdk_dispatcher_bootstrap: $message" try { - executor.allowCoreThreadTimeOut(false) - afterExecutorCreateForTest?.invoke(lane.name, executor) - executor.prestartAllCoreThreads() - } catch (t: Throwable) { - executor.shutdownNow() - throw t + if (!Logging.atLogLevel(if (warn) LogLevel.WARN else LogLevel.INFO)) return + if (warn) Log.w(RELEASE_LOG_TAG, line, error) else Log.i(RELEASE_LOG_TAG, line) + } catch (_: RuntimeException) { + // JVM unit tests throw from android.util.Log. A log must not fail the latch. } - return LaneTarget(executor.asCoroutineDispatcher(), executor) } - private fun Lane.config(): LaneConfig = - when (this) { - Lane.IO -> - LaneConfig( - IO_CORE_POOL_SIZE, - IO_MAX_POOL_SIZE, - IO_THREAD_NAME_PREFIX, - Thread.NORM_PRIORITY - 1, - ) - Lane.DEFAULT -> - LaneConfig( - DEFAULT_CORE_POOL_SIZE, - DEFAULT_MAX_POOL_SIZE, - DEFAULT_THREAD_NAME_PREFIX, - ) - Lane.SERIAL_IO -> - LaneConfig( - 1, - 1, - SERIAL_IO_THREAD_NAME, - Thread.NORM_PRIORITY - 1, - // Unbounded like Executors.newSingleThreadExecutor; rejecting would drop ordered lifecycle work. - Int.MAX_VALUE, - ) - // Kept separate from IO so receiver handoffs never queue behind SDK HTTP. - Lane.INGRESS -> - LaneConfig( - 1, - 1, - INGRESS_THREAD_NAME, - Thread.NORM_PRIORITY - 1, - Int.MAX_VALUE, - ) + /** Lane and pool failures. Also sent through [Logging] so they reach remote telemetry. */ + @Suppress("TooGenericExceptionCaught", "SwallowedException") + internal fun logFailure( + message: String, + error: Throwable, + ) { + // Log listeners and visual logging can launch onto the failing lane, which fails and logs again. + if (inFailureLog.get() == true) return logRelease(message, error) + inFailureLog.set(true) + try { + Logging.warn("OneSignalDispatchers: $message", error) + } catch (_: RuntimeException) { + logRelease(message, error) + } finally { + inFailureLog.remove() } - - private fun createFallbackTarget(lane: Lane): LaneTarget { - beforeFallbackCreateForTest?.invoke(lane.name) - val dispatcher = - when (lane) { - Lane.IO -> Dispatchers.IO - Lane.DEFAULT -> Dispatchers.Default - Lane.SERIAL_IO, Lane.INGRESS -> { - @Suppress("OPT_IN_USAGE") - Dispatchers.IO.limitedParallelism(1) - } - } - dispatcher.dispatch(kotlin.coroutines.EmptyCoroutineContext, Runnable {}) - return LaneTarget(dispatcher, null) } - @Volatile - private var pools = Pools() - // Dispatchers and scopes delegate to the current generation so [resetForTest] can swap them. - val IO: CoroutineDispatcher get() = pools.IO + val IO: CoroutineDispatcher get() = currentPools().io.dispatcher - val Default: CoroutineDispatcher get() = pools.Default + val Default: CoroutineDispatcher get() = currentPools().defaultLane.dispatcher - val SerialIO: CoroutineDispatcher get() = pools.SerialIO + val SerialIO: CoroutineDispatcher get() = currentPools().serialIO.dispatcher fun launchOnIO(block: suspend () -> Unit): Job { - return pools.IOScope.launch { block() } + return currentPools().ioScope.launch { block() } } fun launchOnDefault(block: suspend () -> Unit): Job { - return pools.DefaultScope.launch { block() } + return currentPools().defaultScope.launch { block() } } /** Launches [block] on the single-thread serial IO dispatcher (FIFO across all callers). */ fun launchOnSerialIO(block: suspend () -> Unit): Job { - return pools.SerialIOScope.launch { block() } + return currentPools().serialIOScope.launch { block() } } /** Launches short durable-ingress work on a lane isolated from general SDK I/O. */ fun launchOnIngress(block: suspend () -> Unit): Job { - return pools.IngressScope.launch { block() } + return currentPools().ingressScope.launch { block() } } @Volatile @@ -465,22 +247,22 @@ object OneSignalDispatchers { private val prewarmLock = Any() /** - * Requests asynchronous initialization and worker prestart for every lane. Dispatch correctness - * does not depend on this call: a cold first dispatch queues behind the same bootstrap gate and - * returns without constructing or waiting for a pool on its caller thread. + * Notes [context] first, since the pools latch from the startup flags file, then warms every lane off + * the caller when it can. Bootstrap queues construction. Lazy construction runs on the caller, or blocks behind prewarm. */ @Suppress("TooGenericExceptionCaught") - fun prewarm() { + fun prewarm(context: Context?) { + context?.let(AppContext::note) if (prewarmStarted) return synchronized(prewarmLock) { if (prewarmStarted) return prewarmStarted = true } try { - pools.prewarm() + currentPools().prewarm() } catch (t: Throwable) { synchronized(prewarmLock) { prewarmStarted = false } - Logging.warn("OneSignalDispatchers.prewarm failed: ${t.message}", t) + logFailure("prewarm failed", t) } } @@ -515,10 +297,16 @@ object OneSignalDispatchers { */ @Suppress("TooGenericExceptionCaught") internal fun resetForTest() { - val old = pools - pools = Pools() + FeatureManager.resetStartupFlagsForTest() + val choice = choosePoolsOrLazy() + val old: DispatcherPools? + synchronized(poolsLock) { + old = poolsRef + poolsRef = choice.pools + } + recordChoice(choice) try { - old.shutdown() + old?.shutdown() } catch (e: Exception) { Logging.error("OneSignalDispatchers.resetForTest failed: ${e.message}", e) } @@ -526,15 +314,19 @@ object OneSignalDispatchers { beforeLaneCreateForTest = null afterExecutorCreateForTest = null beforeFallbackCreateForTest = null + beforeLazyWarmForTest = null + beforeLazyPrewarmStartForTest = null + beforePoolChoiceForTest = null + duringChoiceHandoffForTest = null } @Suppress("ComplexMethod") internal fun getPerformanceMetrics(): String { - val current = pools - val io = current.IO.executor() - val default = current.Default.executor() - val serial = current.SerialIO.executor() - val ingress = current.Ingress.executor() + val current = currentPools() + val io = current.io.executor() + val default = current.defaultLane.executor() + val serial = current.serialIO.executor() + val ingress = current.ingress.executor() return """ OneSignalDispatchers Performance Metrics: - IO Pool: ${io?.let { "${it.activeCount}/${it.corePoolSize}" } ?: "n/a"} active/core threads @@ -549,25 +341,25 @@ object OneSignalDispatchers { } internal fun getStatus(): String { - val current = pools + val current = currentPools() return """ OneSignalDispatchers Status: - - IO Executor: ${current.IO.status()} - - Default Executor: ${current.Default.status()} - - SerialIO Executor: ${current.SerialIO.status()} - - Ingress Executor: ${current.Ingress.status()} - - IO Scope: ${scopeStatus("IOScope") { current.IOScope.isActive }} - - Default Scope: ${scopeStatus("DefaultScope") { current.DefaultScope.isActive }} - - SerialIO Scope: ${scopeStatus("SerialIOScope") { current.SerialIOScope.isActive }} - - Ingress Scope: ${scopeStatus("IngressScope") { current.IngressScope.isActive }} + - IO Executor: ${current.io.status()} + - Default Executor: ${current.defaultLane.status()} + - SerialIO Executor: ${current.serialIO.status()} + - Ingress Executor: ${current.ingress.status()} + - IO Scope: ${scopeStatus("IOScope") { current.ioScope.isActive }} + - Default Scope: ${scopeStatus("DefaultScope") { current.defaultScope.isActive }} + - SerialIO Scope: ${scopeStatus("SerialIOScope") { current.serialIOScope.isActive }} + - Ingress Scope: ${scopeStatus("IngressScope") { current.ingressScope.isActive }} """.trimIndent() } internal fun awaitReadyForTest(timeoutMs: Long = TEST_READY_TIMEOUT_MS): Boolean { val deadline = System.currentTimeMillis() + timeoutMs while (System.currentTimeMillis() < deadline) { - val current = pools - if (listOf(current.IO, current.Default, current.SerialIO, current.Ingress).all { it.status() == "Active" }) { + val current = currentPools() + if (listOf(current.io, current.defaultLane, current.serialIO, current.ingress).all { it.status() in READY_STATUSES }) { return true } Thread.sleep(TEST_READY_POLL_MS) @@ -597,6 +389,7 @@ object OneSignalDispatchers { "$name $NOT_INITIALIZED ${e.message ?: UNKNOWN_ERROR}" } + private val READY_STATUSES = setOf("Active", "Fallback") private const val NOT_INITIALIZED = "Not initialized" private const val UNKNOWN_ERROR = "Unknown error" } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt index e20d9b44ee..fc59121a26 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/CoreModule.kt @@ -2,6 +2,7 @@ package com.onesignal.core import com.onesignal.common.modules.IModule import com.onesignal.common.services.ServiceBuilder +import com.onesignal.common.threading.DispatcherChoiceReporter import com.onesignal.core.internal.application.IApplicationService import com.onesignal.core.internal.application.impl.ApplicationService import com.onesignal.core.internal.backend.IFeatureFlagsBackendService @@ -86,6 +87,7 @@ internal class CoreModule : IModule { builder.register().provides() builder.register().provides() builder.register().provides() + builder.register().provides() builder.register().provides() builder.register() diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/application/impl/ActivityLifecycleInitializer.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/application/impl/ActivityLifecycleInitializer.kt index e910e4fa67..4c27318c47 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/application/impl/ActivityLifecycleInitializer.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/application/impl/ActivityLifecycleInitializer.kt @@ -3,7 +3,6 @@ package com.onesignal.core.internal.application.impl import android.app.Application import android.content.Context import androidx.startup.Initializer -import com.onesignal.common.AppContext import com.onesignal.common.threading.OneSignalDispatchers import com.onesignal.core.internal.application.IApplicationService @@ -22,8 +21,7 @@ import com.onesignal.core.internal.application.IApplicationService */ class ActivityLifecycleInitializer : Initializer { override fun create(context: Context): IApplicationService { - AppContext.note(context) - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(context) val applicationService = ApplicationService.getInstance() (context.applicationContext as? Application)?.let { application -> applicationService.attachToApplication(application) diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/features/README.md b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/features/README.md index 43e4c331a0..cd7f091391 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/features/README.md +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/features/README.md @@ -4,6 +4,7 @@ `OneSignalStartupFlags`, so code that runs before SDK init can read them without loading the whole OneSignal prefs file (operations queue included) on the main thread. Typical pre-init readers make a choice that lives for the whole process, such as which thread pools or logging pipeline to build. +`OneSignalDispatchers` is one: it picks its pools from `sdk_dispatcher_bootstrap`. ## Adding a startup flag @@ -69,8 +70,9 @@ process-wide holder (`StartupFlagsCache`). ## Tests - `FeatureManager.resetStartupFlagsForTest()` forgets the snapshot, as a new process would. - `OneSignalTestProjectConfig` calls it around every spec. + `OneSignalTestProjectConfig` calls it around every spec, and `OneSignalDispatchers.resetForTest()` + calls it before building a new pool generation. - `StartupFlagsCache.setSnapshotForTest(...)` installs a fixed snapshot; `readFile` reads the file without touching the snapshot. - `FeatureManager.localFeatureOverrides` does not reach the snapshot. Pre-init readers need their own - test hooks. + test hooks (for example `OneSignalDispatchers.bootstrapEnabledForTest`). diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt index 3b686d249c..f89b2189f3 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt @@ -59,7 +59,7 @@ class SyncJobService : JobService() { // short-lived daemon thread; idempotent, so it's harmless when initWithContext already // ran. Must happen BEFORE suspendifyOnIO; calling it inside the dispatched block would // be too late because the cold-init cost has already been paid on entry to the helper. - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(this) val run = JobRun(jobParameters) activeRun.getAndSet(run)?.let { previous -> diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/internal/OneSignalImp.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/internal/OneSignalImp.kt index 56f11bd106..adbc36708c 100644 --- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/internal/OneSignalImp.kt +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/internal/OneSignalImp.kt @@ -356,7 +356,7 @@ internal class OneSignalImp : IOneSignal, // of suspendifyOnIO / launchOnSerialIO doesn't pay the ThreadPoolExecutor + dispatcher + // scope construction cost on the main thread (observed as 5-20s main-thread blocks at the // first foreground/background lifecycle event). prewarm() is idempotent and fire-and-forget. - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(context) synchronized(initLock) { if (initState.isSDKAccessible()) { @@ -765,7 +765,7 @@ internal class OneSignalImp : IOneSignal, // (e.g. SyncJobService.onStartJob -> suspendifyOnIO -> initWithContext(context)) pays the // cold-init cost on the dispatcher used to enter [withContext] below, so warm // OneSignalDispatchers on a background thread before we touch [ioDispatcher]. - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(context) // Observing a terminal SUCCESS must never require an IO worker (see waitForInit). if (initState == InitState.SUCCESS) { diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/OneSignalTestProjectConfig.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/OneSignalTestProjectConfig.kt index b4bc385edd..5adcad7739 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/OneSignalTestProjectConfig.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/OneSignalTestProjectConfig.kt @@ -1,7 +1,6 @@ package com.onesignal import com.onesignal.common.threading.OneSignalDispatchers -import com.onesignal.core.internal.features.FeatureManager import io.kotest.core.config.AbstractProjectConfig import io.kotest.core.extensions.Extension import io.kotest.core.listeners.AfterSpecListener @@ -31,12 +30,17 @@ class OneSignalTestProjectConfig : AbstractProjectConfig() { private object OneSignalDispatchersResetListener : BeforeSpecListener, AfterSpecListener { override suspend fun beforeSpec(spec: Spec) { - OneSignalDispatchers.resetForTest() - FeatureManager.resetStartupFlagsForTest() + resetDispatchers() } override suspend fun afterSpec(spec: Spec) { - OneSignalDispatchers.resetForTest() - FeatureManager.resetStartupFlagsForTest() + resetDispatchers() } } + +// Inputs first: the fresh generation latches its pools from them, and a spec's mock +// Application must not decide the pools of the next spec. +private fun resetDispatchers() { + OneSignalDispatchers.resetBootstrapInputsForTest() + OneSignalDispatchers.resetForTest() +} diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/DispatcherBootstrapFlagTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/DispatcherBootstrapFlagTests.kt new file mode 100644 index 0000000000..54e0cf1cd8 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/DispatcherBootstrapFlagTests.kt @@ -0,0 +1,270 @@ +package com.onesignal.common.threading + +import android.content.Context +import androidx.test.core.app.ApplicationProvider +import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest +import com.onesignal.common.AppContext +import com.onesignal.core.internal.features.FeatureManager +import com.onesignal.core.internal.features.StartupFlagsCache +import com.onesignal.debug.ILogListener +import com.onesignal.debug.LogLevel +import com.onesignal.debug.internal.logging.Logging +import com.onesignal.features.FeatureFlag +import com.onesignal.mocks.MockHelper +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.shouldBe +import io.kotest.matchers.string.shouldContain +import io.mockk.every +import io.mockk.mockk +import java.util.concurrent.CopyOnWriteArrayList +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicLong + +/** How [OneSignalDispatchers] picks pools from the startup flags snapshot. */ +@RobolectricTest +class DispatcherBootstrapFlagTests : FunSpec({ + val bootstrap = FeatureFlag.SDK_DISPATCHER_BOOTSTRAP + val key = bootstrap.key + + afterTest { + reportListener?.let(Logging::removeListener) + reportListener = null + StartupFlagsCache.write(appContext()) { emptyList() } + OneSignalDispatchers.resetBootstrapInputsForTest() + OneSignalDispatchers.resetForTest() + } + + test("each APP_STARTUP flag is mirrored on its own") { + val context = appContext() + val logging = FeatureFlag.SDK_CUSTOM_LOGGING + StartupFlagsCache.write(context) { listOf(logging.key) } + StartupFlagsCache.readFile(context).isEnabled(logging) shouldBe true + StartupFlagsCache.readFile(context).isEnabled(bootstrap) shouldBe false + + StartupFlagsCache.write(context) { listOf(logging.key, key) } + StartupFlagsCache.readFile(context).isEnabled(logging) shouldBe true + StartupFlagsCache.readFile(context).isEnabled(bootstrap) shouldBe true + } + + test("a mirrored key selects bootstrap on the next pool generation") { + val context = appContext() + StartupFlagsCache.write(context) { listOf(key) } + latchFrom(context) + + builtOnCaller() shouldBe false + } + + test("no mirrored key keeps construction on the caller") { + val context = appContext() + StartupFlagsCache.write(context) { emptyList() } + latchFrom(context) + + builtOnCaller() shouldBe true + } + + test("an unreadable file keeps construction on the caller") { + latchFrom(unreadableContext()) + + builtOnCaller() shouldBe true + } + + test("a pool choice failure falls back to lazy pools instead of throwing") { + val context = appContext() + StartupFlagsCache.write(context) { listOf(key) } + OneSignalDispatchers.beforePoolChoiceForTest = { throw IllegalStateException("choice failed") } + latchFrom(context) + + builtOnCaller() shouldBe true + } + + test("a choice failure after an on snapshot runs lazy pools while FeatureManager reports on") { + val context = appContext() + StartupFlagsCache.write(context) { listOf(key) } + OneSignalDispatchers.beforePoolChoiceForTest = { + FeatureManager.startupFlags { context } + throw IllegalStateException("choice failed") + } + val reports = captureChoiceReports() + latchFrom(context) + + builtOnCaller() shouldBe true + FeatureManager(MockHelper.configModelStore(), applicationServiceFor(context)).isEnabled(bootstrap) shouldBe true + OneSignalDispatchers.reportChoiceToTelemetry() + reports.single() shouldContain "off, choice failed; pools=lazy" + } + + test("an unreadable file reaches telemetry once init has installed it") { + val reports = captureChoiceReports() + latchFrom(unreadableContext()) + reports shouldBe emptyList() + + OneSignalDispatchers.reportChoiceToTelemetry() + reports.single() shouldContain "off, cache unreadable; pools=lazy" + OneSignalDispatchers.reportChoiceToTelemetry() + reports.size shouldBe 1 + } + + test("a degraded choice after telemetry is up is reported right away") { + val reports = captureChoiceReports() + OneSignalDispatchers.resetBootstrapInputsForTest() + OneSignalDispatchers.reportChoiceToTelemetry() + AppContext.note(unreadableContext()) + OneSignalDispatchers.resetForTest() + + reports.single() shouldContain "off, cache unreadable" + } + + test("a degraded choice racing telemetry readiness is still reported") { + val reports = captureChoiceReports() + OneSignalDispatchers.resetBootstrapInputsForTest() + val inHandoff = CountDownLatch(1) + val release = CountDownLatch(1) + OneSignalDispatchers.duringChoiceHandoffForTest = { + inHandoff.countDown() + release.await(2, TimeUnit.SECONDS) + } + val chooser = Thread { OneSignalDispatchers.resetForTest() }.apply { start() } + inHandoff.await(2, TimeUnit.SECONDS) shouldBe true + + val reporter = Thread { OneSignalDispatchers.reportChoiceToTelemetry() }.apply { start() } + awaitBlockedOrDone(reporter) + release.countDown() + chooser.join(2_000) + reporter.join(2_000) + + reports.single() shouldContain "off, no application; pools=lazy" + } + + test("a healthy choice is not reported") { + val context = appContext() + StartupFlagsCache.write(context) { listOf(key) } + val reports = captureChoiceReports() + latchFrom(context) + OneSignalDispatchers.reportChoiceToTelemetry() + + reports shouldBe emptyList() + } + + test("the bootstrap thread runs at normal priority") { + OneSignalDispatchers.bootstrapEnabledForTest = true + OneSignalDispatchers.resetForTest() + val priority = AtomicInteger(-1) + val seen = CountDownLatch(1) + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "IO") { + priority.set(Thread.currentThread().priority) + seen.countDown() + } + } + + OneSignalDispatchers.launchOnIO { } + + seen.await(2, TimeUnit.SECONDS) shouldBe true + priority.get() shouldBe Thread.NORM_PRIORITY + } + + test("the lazy prewarm thread runs at normal priority") { + OneSignalDispatchers.bootstrapEnabledForTest = false + OneSignalDispatchers.resetForTest() + val priority = AtomicInteger(-1) + val seen = CountDownLatch(1) + OneSignalDispatchers.beforeLazyWarmForTest = { + if (priority.compareAndSet(-1, Thread.currentThread().priority)) seen.countDown() + } + + OneSignalDispatchers.prewarm(null) + + seen.await(2, TimeUnit.SECONDS) shouldBe true + priority.get() shouldBe Thread.NORM_PRIORITY + } + + test("the pools latch from the context passed to prewarm when startup did not note one") { + StartupFlagsCache.write(appContext()) { listOf(key) } + OneSignalDispatchers.resetBootstrapInputsForTest() + + OneSignalDispatchers.prewarm(appContext()) + OneSignalDispatchers.resetForTest() + + builtOnCaller() shouldBe false + } + + test("with no context noted the pools stay lazy instead of guessing one") { + StartupFlagsCache.write(appContext()) { listOf(key) } + OneSignalDispatchers.resetBootstrapInputsForTest() + OneSignalDispatchers.resetForTest() + + builtOnCaller() shouldBe true + } + + test("FeatureManager reports the pools the process actually latched when store and file disagree") { + val context = appContext() + StartupFlagsCache.write(context) { emptyList() } + latchFrom(context) + builtOnCaller() shouldBe true + + val store = MockHelper.configModelStore { it.sdkRemoteFeatureFlags = listOf(key) } + val applicationService = MockHelper.applicationService() + every { applicationService.appContext } returns context + val manager = FeatureManager(store, applicationService) + + manager.isEnabled(bootstrap) shouldBe false + manager.enabledFeatureKeys() shouldBe emptyList() + } +}) + +private fun appContext(): Context = ApplicationProvider.getApplicationContext() + +private var reportListener: ILogListener? = null + +/** Warnings [OneSignalDispatchers] sends through [Logging] about its pool choice. */ +private fun captureChoiceReports(): List { + val reports = CopyOnWriteArrayList() + val listener = ILogListener { if (it.level == LogLevel.WARN && "sdk_dispatcher_bootstrap" in it.entry) reports.add(it.entry) } + reportListener = listener + Logging.addListener(listener) + return reports +} + +private fun unreadableContext(): Context { + val broken = mockk() + every { broken.applicationContext } returns broken + every { broken.getSharedPreferences(any(), any()) } throws SecurityException("prefs unavailable") + return broken +} + +private fun applicationServiceFor(context: Context) = MockHelper.applicationService().also { every { it.appContext } returns context } + +private fun awaitBlockedOrDone(thread: Thread) { + val deadline = System.currentTimeMillis() + 2_000 + while (thread.state != Thread.State.BLOCKED && thread.state != Thread.State.TERMINATED) { + if (System.currentTimeMillis() > deadline) return + Thread.sleep(5) + } +} + +private fun latchFrom(context: Context) { + OneSignalDispatchers.resetBootstrapInputsForTest() + AppContext.note(context) + OneSignalDispatchers.resetForTest() +} + +/** True when the IO lane is built on the thread that first launched work. */ +private fun builtOnCaller(): Boolean { + val callerThreadId = AtomicLong() + val createThreadId = AtomicLong() + val created = CountDownLatch(1) + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "IO") { + createThreadId.set(Thread.currentThread().id) + created.countDown() + } + } + Thread { + callerThreadId.set(Thread.currentThread().id) + OneSignalDispatchers.launchOnIO { } + }.start() + created.await(2, TimeUnit.SECONDS) shouldBe true + return createThreadId.get() == callerThreadId.get() +} diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/DispatcherFailureLogTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/DispatcherFailureLogTests.kt new file mode 100644 index 0000000000..ed34141650 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/DispatcherFailureLogTests.kt @@ -0,0 +1,85 @@ +package com.onesignal.common.threading + +import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest +import com.onesignal.debug.ILogListener +import com.onesignal.debug.LogLevel +import com.onesignal.debug.internal.logging.Logging +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.ints.shouldBeLessThanOrEqual +import io.kotest.matchers.shouldBe +import org.robolectric.shadows.ShadowLog +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger + +/** Lane failures go through [Logging], whose listeners can launch straight back onto the failing lane. */ +@RobolectricTest +class DispatcherFailureLogTests : FunSpec({ + // Top-level tests only: the Robolectric extension deadlocks on nested containers. + fun usePools(bootstrap: Boolean) { + Logging.logLevel = LogLevel.NONE + OneSignalDispatchers.bootstrapEnabledForTest = bootstrap + OneSignalDispatchers.resetForTest() + } + + afterEach { + OneSignalDispatchers.bootstrapEnabledForTest = null + OneSignalDispatchers.resetForTest() + } + + test("logcat lines follow the customer's SDK log level") { + ShadowLog.clear() + try { + Logging.logLevel = LogLevel.NONE + OneSignalDispatchers.logRelease("probe at none", IllegalStateException("x")) + Logging.logLevel = LogLevel.INFO + OneSignalDispatchers.logRelease("probe at info") + } finally { + Logging.logLevel = LogLevel.WARN + } + + val lines = ShadowLog.getLogsForTag("OneSignal").map { it.msg } + lines.any { "probe at none" in it } shouldBe false + lines.any { "probe at info" in it } shouldBe true + } + + listOf(true to "bootstrap", false to "lazy").forEach { (bootstrap, mode) -> + test("$mode pools: a lane failure reaches log listeners") { + usePools(bootstrap) + val logged = CountDownLatch(1) + val listener = ILogListener { if ("IO fell back" in it.entry) logged.countDown() } + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "IO") throw IllegalStateException("no threads") + } + Logging.addListener(listener) + try { + OneSignalDispatchers.launchOnIO { } + logged.await(2, TimeUnit.SECONDS) shouldBe true + } finally { + Logging.removeListener(listener) + } + } + + test("$mode pools: a listener that launches work does not re-enter lane creation") { + usePools(bootstrap) + val attempts = AtomicInteger() + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "IO") { + attempts.incrementAndGet() + throw IllegalStateException("no threads") + } + } + val listener = ILogListener { OneSignalDispatchers.launchOnIO { } } + Logging.addListener(listener) + try { + val ran = CountDownLatch(1) + OneSignalDispatchers.launchOnIO { ran.countDown() } + ran.await(2, TimeUnit.SECONDS) shouldBe true + } finally { + Logging.removeListener(listener) + } + + attempts.get() shouldBeLessThanOrEqual 2 + } + } +}) diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/DispatcherLaneBehaviorTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/DispatcherLaneBehaviorTests.kt new file mode 100644 index 0000000000..bcc1c91bb3 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/DispatcherLaneBehaviorTests.kt @@ -0,0 +1,338 @@ +package com.onesignal.common.threading + +import com.onesignal.debug.LogLevel +import com.onesignal.debug.internal.logging.Logging +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.shouldBe +import io.kotest.matchers.shouldNotBe +import io.kotest.matchers.string.shouldContain +import kotlinx.coroutines.delay +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicLong + +/** Lane behavior both pool types must share. Each test runs once per value of the flag. */ +class DispatcherLaneBehaviorTests : FunSpec({ + beforeAny { + Logging.logLevel = LogLevel.NONE + } + + listOf(true to "bootstrap", false to "lazy").forEach { (bootstrap, mode) -> + context("$mode pools") { + beforeEach { + OneSignalDispatchers.bootstrapEnabledForTest = bootstrap + OneSignalDispatchers.resetForTest() + } + + afterEach { + OneSignalDispatchers.resetBootstrapInputsForTest() + OneSignalDispatchers.resetForTest() + } + + test("builds lanes on the thread this mode promises") { + val callerThreadId = AtomicLong() + val createThreadId = AtomicLong() + val created = CountDownLatch(1) + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "IO") { + createThreadId.set(Thread.currentThread().id) + created.countDown() + } + } + Thread { + callerThreadId.set(Thread.currentThread().id) + OneSignalDispatchers.launchOnIO { } + }.start() + + created.await(2, TimeUnit.SECONDS) shouldBe true + (createThreadId.get() == callerThreadId.get()) shouldBe !bootstrap + } + + test("OneSignalDispatchers should be properly initialized") { + // Access dispatchers to trigger initialization + OneSignalDispatchers.IO shouldNotBe null + OneSignalDispatchers.Default shouldNotBe null + } + + test("ingress work runs while every IO worker is blocked") { + OneSignalDispatchers.resetForTest() + val releaseIo = CountDownLatch(1) + val ioWorkersBlocked = CountDownLatch(2) + val ingressRan = CountDownLatch(1) + var ingressThreadName: String? = null + + repeat(2) { + OneSignalDispatchers.launchOnIO { + ioWorkersBlocked.countDown() + releaseIo.await() + } + } + ioWorkersBlocked.await(1, TimeUnit.SECONDS) shouldBe true + + OneSignalDispatchers.launchOnIngress { + ingressThreadName = Thread.currentThread().name + ingressRan.countDown() + } + + try { + ingressRan.await(1, TimeUnit.SECONDS) shouldBe true + ingressThreadName shouldContain "OneSignal-Ingress" + } finally { + releaseIo.countDown() + OneSignalDispatchers.resetForTest() + } + } + + test("IO dispatcher should execute work on background thread") { + val mainThreadId = Thread.currentThread().id + var backgroundThreadId: Long? = null + + runBlocking { + withContext(OneSignalDispatchers.IO) { + backgroundThreadId = Thread.currentThread().id + } + } + + backgroundThreadId shouldNotBe null + backgroundThreadId shouldNotBe mainThreadId + } + + test("Default dispatcher should execute work on background thread") { + val mainThreadId = Thread.currentThread().id + var backgroundThreadId: Long? = null + + runBlocking { + withContext(OneSignalDispatchers.Default) { + backgroundThreadId = Thread.currentThread().id + } + } + + backgroundThreadId shouldNotBe null + backgroundThreadId shouldNotBe mainThreadId + } + + test("IOScope should launch coroutines asynchronously") { + var completed = false + + OneSignalDispatchers.launchOnIO { + Thread.sleep(100) + completed = true + } + + Thread.sleep(50) + completed shouldBe false + } + + test("DefaultScope should launch coroutines asynchronously") { + var completed = false + + OneSignalDispatchers.launchOnDefault { + Thread.sleep(100) + completed = true + } + + Thread.sleep(50) + completed shouldBe false + } + + test("getStatus should return meaningful status information") { + OneSignalDispatchers.prewarm(null) + OneSignalDispatchers.awaitReadyForTest() shouldBe true + val status = OneSignalDispatchers.getStatus() + + status shouldContain "OneSignalDispatchers Status:" + status shouldContain "IO Executor: Active" + status shouldContain "Default Executor: Active" + status shouldContain "SerialIO Executor: Active" + status shouldContain "IO Scope: Active" + status shouldContain "Default Scope: Active" + status shouldContain "SerialIO Scope: Active" + status shouldContain "Ingress Executor: Active" + status shouldContain "Ingress Scope: Active" + } + + test("getPerformanceMetrics should include SerialIO queue and total completed task counters") { + // Trigger lazy init of the SerialIO executor before asking for metrics so its queue + // line resolves to a concrete value instead of the "n/a" fallback path. + OneSignalDispatchers.SerialIO shouldNotBe null + + val metrics = OneSignalDispatchers.getPerformanceMetrics() + + metrics shouldContain "OneSignalDispatchers Performance Metrics:" + metrics shouldContain "SerialIO Queue:" + metrics shouldContain "Total completed tasks:" + } + + test("SerialIO dispatcher executes work on a background thread") { + val callerThreadId = Thread.currentThread().id + var serialThreadId: Long? = null + + runBlocking { + withContext(OneSignalDispatchers.SerialIO) { + serialThreadId = Thread.currentThread().id + } + } + + serialThreadId shouldNotBe null + serialThreadId shouldNotBe callerThreadId + } + + test("launchOnSerialIO runs tasks on a single thread in submission order") { + // SerialIO's contract: submission order on the caller thread == execution order on + // the worker thread. We submit N tasks with a small sleep so they queue up, then + // assert the recorded order matches submission order and that all observations + // came from one thread. + val taskCount = 5 + val observedOrder = mutableListOf() + val observedThreads = mutableSetOf() + val latch = CountDownLatch(taskCount) + + repeat(taskCount) { i -> + OneSignalDispatchers.launchOnSerialIO { + Thread.sleep(5) + synchronized(observedOrder) { + observedOrder.add(i) + observedThreads.add(Thread.currentThread().id) + } + latch.countDown() + } + } + + latch.await() + observedOrder shouldBe (0 until taskCount).toList() + observedThreads.size shouldBe 1 + } + + test("exceptions in a SerialIO task do not stop subsequent tasks from running") { + // Mirrors the parallel "exceptions in one task should not affect others" case for + // launchOnIO. A thrown exception in one serial task must not poison the dispatcher + // for the rest of the queue. + val latch = CountDownLatch(3) + val successCount = AtomicInteger(0) + val errorCount = AtomicInteger(0) + + repeat(3) { i -> + OneSignalDispatchers.launchOnSerialIO { + try { + if (i == 1) { + throw RuntimeException("Test error") + } + successCount.incrementAndGet() + } catch (e: Exception) { + errorCount.incrementAndGet() + } finally { + latch.countDown() + } + } + } + + latch.await() + successCount.get() shouldBe 2 + errorCount.get() shouldBe 1 + } + + test("dispatchers should handle concurrent operations") { + val results = mutableListOf() + val expectedResults = (1..5).toList() + + runBlocking { + (1..5).forEach { i -> + OneSignalDispatchers.launchOnIO { + Thread.sleep(10) + synchronized(results) { + results.add(i) + } + } + } + + Thread.sleep(100) + } + + results.sorted() shouldBe expectedResults + } + + test("multiple concurrent launches should not cause issues") { + val latch = CountDownLatch(5) // Reduced from 20 to 5 + val completed = AtomicInteger(0) + + repeat(5) { i -> // Reduced from 20 to 5 + OneSignalDispatchers.launchOnIO { + delay(10) // Use coroutine delay instead of Thread.sleep + completed.incrementAndGet() + latch.countDown() + } + } + + latch.await() + completed.get() shouldBe 5 // Updated expectation + } + + test("mixed IO and computation tasks should work together") { + val latch = CountDownLatch(10) + val ioCount = AtomicInteger(0) + val compCount = AtomicInteger(0) + + repeat(5) { i -> + OneSignalDispatchers.launchOnIO { + Thread.sleep(20) + ioCount.incrementAndGet() + latch.countDown() + } + + OneSignalDispatchers.launchOnDefault { + Thread.sleep(20) + compCount.incrementAndGet() + latch.countDown() + } + } + + latch.await() + ioCount.get() shouldBe 5 + compCount.get() shouldBe 5 + } + + test("exceptions in one task should not affect others") { + val latch = CountDownLatch(5) + val successCount = AtomicInteger(0) + val errorCount = AtomicInteger(0) + + repeat(5) { i -> + OneSignalDispatchers.launchOnIO { + try { + if (i == 2) { + throw RuntimeException("Test error") + } + Thread.sleep(10) + successCount.incrementAndGet() + } catch (e: Exception) { + errorCount.incrementAndGet() + } finally { + latch.countDown() + } + } + } + + latch.await() + successCount.get() shouldBe 4 + errorCount.get() shouldBe 1 + } + + test("prewarm returns immediately and warms IO / Default / SerialIO dispatchers on a background thread") { + OneSignalDispatchers.resetForTest() + OneSignalDispatchers.prewarm(null) + + OneSignalDispatchers.awaitReadyForTest() shouldBe true + val status = OneSignalDispatchers.getStatus() + status shouldContain "IO Executor: Active" + status shouldContain "Default Executor: Active" + status shouldContain "SerialIO Executor: Active" + status shouldContain "IO Scope: Active" + status shouldContain "Default Scope: Active" + status shouldContain "SerialIO Scope: Active" + } + } + } +}) diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt index 1e8556b8b9..fbd2b07a28 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt @@ -5,18 +5,15 @@ import com.onesignal.debug.internal.logging.Logging import io.kotest.core.spec.style.FunSpec import io.kotest.matchers.shouldBe import io.kotest.matchers.shouldNotBe -import io.kotest.matchers.string.shouldContain import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.delay import kotlinx.coroutines.launch -import kotlinx.coroutines.runBlocking -import kotlinx.coroutines.withContext import java.util.Collections import java.util.concurrent.CountDownLatch import java.util.concurrent.ThreadPoolExecutor import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicLong +import java.util.concurrent.atomic.AtomicReference class OneSignalDispatchersTests : FunSpec({ @@ -24,39 +21,14 @@ class OneSignalDispatchersTests : FunSpec({ Logging.logLevel = LogLevel.NONE } - test("OneSignalDispatchers should be properly initialized") { - // Access dispatchers to trigger initialization - OneSignalDispatchers.IO shouldNotBe null - OneSignalDispatchers.Default shouldNotBe null + beforeTest { + OneSignalDispatchers.bootstrapEnabledForTest = true + OneSignalDispatchers.resetForTest() } - test("ingress work runs while every IO worker is blocked") { + afterTest { + OneSignalDispatchers.resetBootstrapInputsForTest() OneSignalDispatchers.resetForTest() - val releaseIo = CountDownLatch(1) - val ioWorkersBlocked = CountDownLatch(2) - val ingressRan = CountDownLatch(1) - var ingressThreadName: String? = null - - repeat(2) { - OneSignalDispatchers.launchOnIO { - ioWorkersBlocked.countDown() - releaseIo.await() - } - } - ioWorkersBlocked.await(1, TimeUnit.SECONDS) shouldBe true - - OneSignalDispatchers.launchOnIngress { - ingressThreadName = Thread.currentThread().name - ingressRan.countDown() - } - - try { - ingressRan.await(1, TimeUnit.SECONDS) shouldBe true - ingressThreadName shouldContain "OneSignal-Ingress" - } finally { - releaseIo.countDown() - OneSignalDispatchers.resetForTest() - } } test("prewarm bootstraps the ingress lane before the other lanes") { @@ -65,7 +37,7 @@ class OneSignalDispatchersTests : FunSpec({ OneSignalDispatchers.beforeLaneCreateForTest = { created += it } try { - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(null) OneSignalDispatchers.awaitReadyForTest() shouldBe true created.first() shouldBe "INGRESS" } finally { @@ -203,126 +175,6 @@ class OneSignalDispatchersTests : FunSpec({ OneSignalDispatchers.resetForTest() } - test("IO dispatcher should execute work on background thread") { - val mainThreadId = Thread.currentThread().id - var backgroundThreadId: Long? = null - - runBlocking { - withContext(OneSignalDispatchers.IO) { - backgroundThreadId = Thread.currentThread().id - } - } - - backgroundThreadId shouldNotBe null - backgroundThreadId shouldNotBe mainThreadId - } - - test("Default dispatcher should execute work on background thread") { - val mainThreadId = Thread.currentThread().id - var backgroundThreadId: Long? = null - - runBlocking { - withContext(OneSignalDispatchers.Default) { - backgroundThreadId = Thread.currentThread().id - } - } - - backgroundThreadId shouldNotBe null - backgroundThreadId shouldNotBe mainThreadId - } - - test("IOScope should launch coroutines asynchronously") { - var completed = false - - OneSignalDispatchers.launchOnIO { - Thread.sleep(100) - completed = true - } - - Thread.sleep(50) - completed shouldBe false - } - - test("DefaultScope should launch coroutines asynchronously") { - var completed = false - - OneSignalDispatchers.launchOnDefault { - Thread.sleep(100) - completed = true - } - - Thread.sleep(50) - completed shouldBe false - } - - test("getStatus should return meaningful status information") { - OneSignalDispatchers.prewarm() - OneSignalDispatchers.awaitReadyForTest() shouldBe true - val status = OneSignalDispatchers.getStatus() - - status shouldContain "OneSignalDispatchers Status:" - status shouldContain "IO Executor: Active" - status shouldContain "Default Executor: Active" - status shouldContain "SerialIO Executor: Active" - status shouldContain "IO Scope: Active" - status shouldContain "Default Scope: Active" - status shouldContain "SerialIO Scope: Active" - status shouldContain "Ingress Executor: Active" - status shouldContain "Ingress Scope: Active" - } - - test("getPerformanceMetrics should include SerialIO queue and total completed task counters") { - // Trigger lazy init of the SerialIO executor before asking for metrics so its queue - // line resolves to a concrete value instead of the "n/a" fallback path. - OneSignalDispatchers.SerialIO shouldNotBe null - - val metrics = OneSignalDispatchers.getPerformanceMetrics() - - metrics shouldContain "OneSignalDispatchers Performance Metrics:" - metrics shouldContain "SerialIO Queue:" - metrics shouldContain "Total completed tasks:" - } - - test("SerialIO dispatcher executes work on a background thread") { - val callerThreadId = Thread.currentThread().id - var serialThreadId: Long? = null - - runBlocking { - withContext(OneSignalDispatchers.SerialIO) { - serialThreadId = Thread.currentThread().id - } - } - - serialThreadId shouldNotBe null - serialThreadId shouldNotBe callerThreadId - } - - test("launchOnSerialIO runs tasks on a single thread in submission order") { - // SerialIO's contract: submission order on the caller thread == execution order on - // the worker thread. We submit N tasks with a small sleep so they queue up, then - // assert the recorded order matches submission order and that all observations - // came from one thread. - val taskCount = 5 - val observedOrder = mutableListOf() - val observedThreads = mutableSetOf() - val latch = CountDownLatch(taskCount) - - repeat(taskCount) { i -> - OneSignalDispatchers.launchOnSerialIO { - Thread.sleep(5) - synchronized(observedOrder) { - observedOrder.add(i) - observedThreads.add(Thread.currentThread().id) - } - latch.countDown() - } - } - - latch.await() - observedOrder shouldBe (0 until taskCount).toList() - observedThreads.size shouldBe 1 - } - test("executorStatus returns 'Active' / 'Shutdown' on the happy path and the Not initialized message when the underlying check throws") { // Happy paths (Shutdown / Active) are exercised indirectly via getStatus(); this // case pins down the defensive catch branch, which fires when the underlying @@ -351,140 +203,178 @@ class OneSignalDispatchersTests : FunSpec({ } shouldBe "IOScope Not initialized Unknown error" } - test("exceptions in a SerialIO task do not stop subsequent tasks from running") { - // Mirrors the parallel "exceptions in one task should not affect others" case for - // launchOnIO. A thrown exception in one serial task must not poison the dispatcher - // for the rest of the queue. - val latch = CountDownLatch(3) - val successCount = AtomicInteger(0) - val errorCount = AtomicInteger(0) - - repeat(3) { i -> - OneSignalDispatchers.launchOnSerialIO { - try { - if (i == 1) { - throw RuntimeException("Test error") - } - successCount.incrementAndGet() - } catch (e: Exception) { - errorCount.incrementAndGet() - } finally { - latch.countDown() - } - } + test("flag off constructs the IO executor on the calling thread") { + OneSignalDispatchers.bootstrapEnabledForTest = false + OneSignalDispatchers.resetForTest() + val callerThreadId = AtomicLong() + val createThreadId = AtomicLong() + val finished = CountDownLatch(1) + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "IO") createThreadId.set(Thread.currentThread().id) } - latch.await() - successCount.get() shouldBe 2 - errorCount.get() shouldBe 1 - } + Thread { + callerThreadId.set(Thread.currentThread().id) + OneSignalDispatchers.launchOnIO { } + finished.countDown() + }.start() - test("dispatchers should handle concurrent operations") { - val results = mutableListOf() - val expectedResults = (1..5).toList() - - runBlocking { - (1..5).forEach { i -> - OneSignalDispatchers.launchOnIO { - Thread.sleep(10) - synchronized(results) { - results.add(i) - } - } - } + finished.await(2, TimeUnit.SECONDS) shouldBe true + createThreadId.get() shouldBe callerThreadId.get() + createThreadId.get() shouldNotBe 0L + } - Thread.sleep(100) + test("flag off prewarm warms ingress before the other lanes") { + OneSignalDispatchers.bootstrapEnabledForTest = false + OneSignalDispatchers.resetForTest() + val order = Collections.synchronizedList(mutableListOf()) + val created = CountDownLatch(4) + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + order.add(lane) + created.countDown() } - - results.sorted() shouldBe expectedResults + OneSignalDispatchers.prewarm(null) + created.await(2, TimeUnit.SECONDS) shouldBe true + order.first() shouldBe "INGRESS" } - test("multiple concurrent launches should not cause issues") { - val latch = CountDownLatch(5) // Reduced from 20 to 5 - val completed = AtomicInteger(0) + test("flag off prewarm does not build executors after reset") { + OneSignalDispatchers.bootstrapEnabledForTest = false + OneSignalDispatchers.resetForTest() + val entered = CountDownLatch(1) + val release = CountDownLatch(1) + // Shutdown interrupts prewarm. Ignore it so the closed checks, not a dead thread, stop the build. + OneSignalDispatchers.beforeLazyWarmForTest = { + entered.countDown() + awaitIgnoringInterrupts(release) + } + OneSignalDispatchers.prewarm(null) + entered.await(2, TimeUnit.SECONDS) shouldBe true + OneSignalDispatchers.resetForTest() + val builtAfterReset = AtomicInteger() + OneSignalDispatchers.afterExecutorCreateForTest = { _, _ -> builtAfterReset.incrementAndGet() } + release.countDown() + Thread.sleep(400) + builtAfterReset.get() shouldBe 0 + } - repeat(5) { i -> // Reduced from 20 to 5 - OneSignalDispatchers.launchOnIO { - delay(10) // Use coroutine delay instead of Thread.sleep - completed.incrementAndGet() - latch.countDown() + test("flag off prewarm does not build an executor when reset lands mid-build") { + OneSignalDispatchers.bootstrapEnabledForTest = false + OneSignalDispatchers.resetForTest() + val entered = CountDownLatch(1) + val release = CountDownLatch(1) + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "INGRESS") { + entered.countDown() + awaitIgnoringInterrupts(release) } } - - latch.await() - completed.get() shouldBe 5 // Updated expectation + OneSignalDispatchers.prewarm(null) + entered.await(2, TimeUnit.SECONDS) shouldBe true + OneSignalDispatchers.resetForTest() + val builtAfterReset = AtomicInteger() + OneSignalDispatchers.afterExecutorCreateForTest = { _, _ -> builtAfterReset.incrementAndGet() } + release.countDown() + Thread.sleep(400) + builtAfterReset.get() shouldBe 0 } - test("mixed IO and computation tasks should work together") { - val latch = CountDownLatch(10) - val ioCount = AtomicInteger(0) - val compCount = AtomicInteger(0) - - repeat(5) { i -> - OneSignalDispatchers.launchOnIO { - Thread.sleep(20) - ioCount.incrementAndGet() - latch.countDown() + test("flag off prewarm shuts down an executor built after reset") { + OneSignalDispatchers.bootstrapEnabledForTest = false + OneSignalDispatchers.resetForTest() + val entered = CountDownLatch(1) + val release = CountDownLatch(1) + val built = AtomicReference() + OneSignalDispatchers.afterExecutorCreateForTest = { lane, executor -> + if (lane == "INGRESS") { + built.set(executor) + entered.countDown() + awaitIgnoringInterrupts(release) } + } + OneSignalDispatchers.prewarm(null) + entered.await(2, TimeUnit.SECONDS) shouldBe true + OneSignalDispatchers.resetForTest() + release.countDown() + Thread.sleep(400) + built.get().isShutdown shouldBe true + } - OneSignalDispatchers.launchOnDefault { - Thread.sleep(20) - compCount.incrementAndGet() - latch.countDown() - } + test("flag off reports Fallback for a lane that fell back") { + OneSignalDispatchers.bootstrapEnabledForTest = false + OneSignalDispatchers.resetForTest() + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "IO") throw IllegalStateException("boom") } + val ran = CountDownLatch(1) + OneSignalDispatchers.launchOnIO { ran.countDown() } + ran.await(2, TimeUnit.SECONDS) shouldBe true - latch.await() - ioCount.get() shouldBe 5 - compCount.get() shouldBe 5 + awaitStatus("IO Executor: Fallback") shouldBe true } - test("exceptions in one task should not affect others") { - val latch = CountDownLatch(5) - val successCount = AtomicInteger(0) - val errorCount = AtomicInteger(0) - - repeat(5) { i -> - OneSignalDispatchers.launchOnIO { - try { - if (i == 2) { - throw RuntimeException("Test error") - } - Thread.sleep(10) - successCount.incrementAndGet() - } catch (e: Exception) { - errorCount.incrementAndGet() - } finally { - latch.countDown() - } - } + test("bootstrap reports Fallback for a lane that fell back") { + OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + if (lane == "IO") throw IllegalStateException("boom") } + val ran = CountDownLatch(1) + OneSignalDispatchers.launchOnIO { ran.countDown() } + ran.await(2, TimeUnit.SECONDS) shouldBe true - latch.await() - successCount.get() shouldBe 4 - errorCount.get() shouldBe 1 + awaitStatus("IO Executor: Fallback") shouldBe true } - test("prewarm returns immediately and warms IO / Default / SerialIO dispatchers on a background thread") { + test("flag off falls back when executor construction throws") { + OneSignalDispatchers.bootstrapEnabledForTest = false OneSignalDispatchers.resetForTest() - OneSignalDispatchers.prewarm() + val fallbacks = AtomicInteger() + OneSignalDispatchers.beforeLaneCreateForTest = { throw IllegalStateException("boom") } + OneSignalDispatchers.beforeFallbackCreateForTest = { fallbacks.incrementAndGet() } + val ran = CountDownLatch(1) + OneSignalDispatchers.launchOnIO { ran.countDown() } + ran.await(2, TimeUnit.SECONDS) shouldBe true + fallbacks.get() shouldBe 1 + } - OneSignalDispatchers.awaitReadyForTest() shouldBe true - val status = OneSignalDispatchers.getStatus() - status shouldContain "IO Executor: Active" - status shouldContain "Default Executor: Active" - status shouldContain "SerialIO Executor: Active" - status shouldContain "IO Scope: Active" - status shouldContain "Default Scope: Active" - status shouldContain "SerialIO Scope: Active" + test("flag off prewarm retries when the prewarm thread fails to start") { + OneSignalDispatchers.bootstrapEnabledForTest = false + OneSignalDispatchers.resetForTest() + OneSignalDispatchers.beforeLazyPrewarmStartForTest = { throw IllegalStateException("start failed") } + OneSignalDispatchers.prewarm(null) + + OneSignalDispatchers.beforeLazyPrewarmStartForTest = null + val created = CountDownLatch(4) + OneSignalDispatchers.beforeLaneCreateForTest = { created.countDown() } + OneSignalDispatchers.prewarm(null) + created.await(2, TimeUnit.SECONDS) shouldBe true } test("prewarm is idempotent: a second call is a no-op and does not spawn a second prewarm thread") { OneSignalDispatchers.resetForTest() - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(null) OneSignalDispatchers.awaitReadyForTest() shouldBe true val statusBefore = OneSignalDispatchers.getStatus() - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(null) OneSignalDispatchers.getStatus() shouldBe statusBefore } }) + +private fun awaitStatus(line: String): Boolean { + val deadline = System.currentTimeMillis() + 2_000 + while (System.currentTimeMillis() < deadline) { + if (line in OneSignalDispatchers.getStatus()) return true + Thread.sleep(5) + } + return false +} + +private fun awaitIgnoringInterrupts(latch: CountDownLatch) { + val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2) + while (latch.count > 0 && System.nanoTime() < deadline) { + try { + latch.await(deadline - System.nanoTime(), TimeUnit.NANOSECONDS) + } catch (_: InterruptedException) { + // Keep waiting: the test needs the thread alive after shutdown. + } + } +} diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/impl/ActivityLifecycleInitializerTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/impl/ActivityLifecycleInitializerTests.kt index d23c11fd49..ef6a4f9164 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/impl/ActivityLifecycleInitializerTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/application/impl/ActivityLifecycleInitializerTests.kt @@ -2,10 +2,8 @@ package com.onesignal.core.internal.application.impl import android.app.Application import android.content.Context -import com.onesignal.common.AppContext import com.onesignal.common.threading.OneSignalDispatchers import io.kotest.core.spec.style.FunSpec -import io.kotest.matchers.shouldBe import io.mockk.every import io.mockk.mockk import io.mockk.mockkObject @@ -27,15 +25,13 @@ import io.mockk.verifyOrder class ActivityLifecycleInitializerTests : FunSpec({ beforeTest { mockkObject(OneSignalDispatchers) - every { OneSignalDispatchers.prewarm() } returns Unit + every { OneSignalDispatchers.prewarm(any()) } returns Unit resetSharedApplicationService() - AppContext.resetForTest() } afterTest { unmockkObject(OneSignalDispatchers) resetSharedApplicationService() - AppContext.resetForTest() } test("create prewarms before registering activity lifecycle callbacks") { @@ -44,32 +40,23 @@ class ActivityLifecycleInitializerTests : FunSpec({ ActivityLifecycleInitializer().create(application) - verify(exactly = 1) { OneSignalDispatchers.prewarm() } + verify(exactly = 1) { OneSignalDispatchers.prewarm(application) } verify(exactly = 1) { application.registerActivityLifecycleCallbacks(any()) } verifyOrder { - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(application) application.registerActivityLifecycleCallbacks(any()) } } - test("create notes the application context for code that runs before init") { - val application = mockk(relaxed = true) - every { application.applicationContext } returns application - - ActivityLifecycleInitializer().create(application) - - AppContext.get() shouldBe application - } - test("create prewarms even when application context is unavailable") { val context = mockk(relaxed = true) every { context.applicationContext } returns context ActivityLifecycleInitializer().create(context) - verify(exactly = 1) { OneSignalDispatchers.prewarm() } + verify(exactly = 1) { OneSignalDispatchers.prewarm(context) } } }) diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/services/SyncJobServiceTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/services/SyncJobServiceTests.kt index a58e02207b..53f41657c0 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/services/SyncJobServiceTests.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/services/SyncJobServiceTests.kt @@ -66,11 +66,11 @@ class SyncJobServiceTests : FunSpec({ mocks.syncJobService.onStartJob(mocks.jobParameters) - verify(exactly = 1) { OneSignalDispatchers.prewarm() } + verify(exactly = 1) { OneSignalDispatchers.prewarm(any()) } // prewarm() must run before the first suspendifyOnIO dispatch, otherwise the cold-init cost // would already be paid on the caller (main) thread by the time the helper is entered. verifyOrder { - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(any()) OneSignalDispatchers.launchOnIO(any Unit>()) } } diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/NotificationOpenedActivityHMS.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/NotificationOpenedActivityHMS.kt index 8d65f73713..ddd3c56e00 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/NotificationOpenedActivityHMS.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/NotificationOpenedActivityHMS.kt @@ -78,7 +78,7 @@ class NotificationOpenedActivityHMS : private fun processIntent() { // HMS notification-open trampoline runs on the main thread and can cold-start the process; // warm dispatchers before the first suspendifyOnDefault dispatch. - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(this) suspendifyOnDefault { try { if (!OneSignal.initWithContext(applicationContext)) { diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/activities/NotificationOpenedActivityBase.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/activities/NotificationOpenedActivityBase.kt index 2b3beb5b6c..b0a281a75a 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/activities/NotificationOpenedActivityBase.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/activities/NotificationOpenedActivityBase.kt @@ -56,7 +56,7 @@ abstract class NotificationOpenedActivityBase : internal open fun processIntent() { // Notification-open trampoline runs on the main thread and can cold-start the process; // warm dispatchers before the first suspendifyOnDefault dispatch. - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(this) suspendifyOnDefault { try { if (!OneSignal.initWithContext(applicationContext)) { diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/bridges/OneSignalHmsEventBridge.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/bridges/OneSignalHmsEventBridge.kt index 4b6c64833b..7baa44fd4b 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/bridges/OneSignalHmsEventBridge.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/bridges/OneSignalHmsEventBridge.kt @@ -41,7 +41,7 @@ object OneSignalHmsEventBridge { if (firstToken.compareAndSet(true, false)) { Logging.info("OneSignalHmsEventBridge onNewToken - HMS token: $token Bundle: $bundle") // HMS can cold-start the process before initWithContext; warm dispatchers first. - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(context) suspendifyOnIO { val registerer = OneSignal.getService() registerer.fireCallback(token) @@ -68,7 +68,7 @@ object OneSignalHmsEventBridge { message: RemoteMessage, ) { // HMS can cold-start the process before initWithContext; warm dispatchers first. - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(context) suspendifyOnDefault { if (!OneSignal.initWithContext(context)) { return@suspendifyOnDefault diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt index 4aa8b9a454..a2229cc5d3 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BootUpReceiver.kt @@ -38,6 +38,7 @@ class BootUpReceiver : BroadcastReceiver() { ) { // Persist the reconstructible restore request without waiting for full SDK initialization. runIngressHandoff( + context, "BootUpReceiver", BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, ) { diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt index 16f35e00ce..79459f786a 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/BroadcastCompletion.kt @@ -1,6 +1,7 @@ package com.onesignal.notifications.receivers import android.content.BroadcastReceiver +import android.content.Context import android.os.SystemClock import com.onesignal.common.threading.OneSignalDispatchers import com.onesignal.common.threading.suspendifyOnIngress @@ -10,19 +11,21 @@ import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean internal fun BroadcastReceiver.runIngressHandoff( + context: Context, receiverName: String, timeoutMs: Long, block: suspend () -> Unit, ) { - runOrderedIngressHandoff(receiverName, timeoutMs) { _, _ -> block() } + runOrderedIngressHandoff(context, receiverName, timeoutMs) { _, _ -> block() } } internal fun BroadcastReceiver.runOrderedIngressHandoff( + context: Context, receiverName: String, timeoutMs: Long, block: suspend (ordered: Boolean, pendingResult: BroadcastReceiver.PendingResult?) -> Unit, ) { - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(context) // goAsync() nulls the pending result, and isOrderedBroadcast is then false. val ordered = isOrderedBroadcast val pendingResult = goAsync() diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt index e6023ee6fb..babd9ca0a5 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/FCMBroadcastReceiver.kt @@ -23,6 +23,7 @@ class FCMBroadcastReceiver : BroadcastReceiver() { } runOrderedIngressHandoff( + context, "FCMBroadcastReceiver", BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, ) { ordered, pendingResult -> diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt index f5b5c74dbe..525279cda7 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/NotificationDismissReceiver.kt @@ -35,6 +35,7 @@ class NotificationDismissReceiver : BroadcastReceiver() { intent: Intent, ) { runIngressHandoff( + context, "NotificationDismissReceiver", BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, ) { diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt index 9788beb5de..1b9eb79b8c 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/receivers/UpgradeReceiver.kt @@ -47,6 +47,7 @@ class UpgradeReceiver : BroadcastReceiver() { // Persist the reconstructible restore request without waiting for full SDK initialization. runIngressHandoff( + context, "UpgradeReceiver", BroadcastCompletion.RECONSTRUCTIBLE_WORK_TIMEOUT_MS, ) { diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/services/ADMMessageHandler.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/services/ADMMessageHandler.kt index 71039eb318..523765d388 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/services/ADMMessageHandler.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/services/ADMMessageHandler.kt @@ -17,7 +17,7 @@ class ADMMessageHandler : ADMMessageHandlerBase("ADMMessageHandler") { val bundle = intent.extras ?: return // ADM can cold-start the process before initWithContext; warm dispatchers first. - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(applicationContext) suspendifyOnIO { if (!OneSignal.initWithContext(context)) { Logging.warn("onMessage skipped due to failed OneSignal init") @@ -32,7 +32,7 @@ class ADMMessageHandler : ADMMessageHandlerBase("ADMMessageHandler") { override fun onRegistered(newRegistrationId: String) { Logging.info("ADM registration ID: $newRegistrationId") - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(applicationContext) suspendifyOnIO { val registerer = OneSignal.getService() registerer.fireCallback(newRegistrationId) @@ -48,7 +48,7 @@ class ADMMessageHandler : ADMMessageHandlerBase("ADMMessageHandler") { ) } - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(applicationContext) suspendifyOnIO { val registerer = OneSignal.getService() registerer.fireCallback(null) diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/services/ADMMessageHandlerJob.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/services/ADMMessageHandlerJob.kt index 5b94d82bea..98309e0e83 100644 --- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/services/ADMMessageHandlerJob.kt +++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/services/ADMMessageHandlerJob.kt @@ -24,7 +24,7 @@ class ADMMessageHandlerJob : ADMMessageHandlerJobBase() { val safeContext = context.applicationContext // ADM can cold-start the process before initWithContext; warm dispatchers first. - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(context) suspendifyOnIO { if (!OneSignal.initWithContext(safeContext)) { Logging.warn("onMessage skipped due to failed OneSignal init") @@ -42,7 +42,7 @@ class ADMMessageHandlerJob : ADMMessageHandlerJobBase() { ) { Logging.info("ADM registration ID: $newRegistrationId") - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(context) suspendifyOnIO { val registerer = OneSignal.getService() registerer.fireCallback(newRegistrationId) @@ -67,7 +67,7 @@ class ADMMessageHandlerJob : ADMMessageHandlerJobBase() { ) } - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(context) suspendifyOnIO { val registerer = OneSignal.getService() registerer.fireCallback(null) diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/activities/NotificationOpenedActivityBaseTest.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/activities/NotificationOpenedActivityBaseTest.kt index 30ce6a5070..f39c85c873 100644 --- a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/activities/NotificationOpenedActivityBaseTest.kt +++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/activities/NotificationOpenedActivityBaseTest.kt @@ -55,7 +55,7 @@ class TestNotificationOpenedActivity : NotificationOpenedActivityBase() { @RobolectricTest class NotificationOpenedActivityTest : FunSpec({ - // processIntent() now calls OneSignalDispatchers.prewarm(); IOMockHelper stubs it to a no-op so + // processIntent() now calls OneSignalDispatchers.prewarm(any()); IOMockHelper stubs it to a no-op so // setup() doesn't spawn the real OneSignal-prewarm daemon + executors that persist across the JVM. listener(IOMockHelper) @@ -66,9 +66,9 @@ class NotificationOpenedActivityTest : FunSpec({ test("processIntent calls prewarm before suspendifyOnDefault") { Robolectric.buildActivity(TestNotificationOpenedActivity::class.java).setup() - verify(atLeast = 1) { OneSignalDispatchers.prewarm() } + verify(atLeast = 1) { OneSignalDispatchers.prewarm(any()) } verifyOrder { - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(any()) suspendifyOnDefault(any Unit>()) } } diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt index 59b3898dc2..13f27db59d 100644 --- a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt +++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/FCMBroadcastReceiverTests.kt @@ -44,7 +44,7 @@ class FCMBroadcastReceiverTests : FunSpec({ } test("FCMBroadcastReceiver.onReceive makes the explicit prewarm() head-start call before dispatch for a normal push") { - // Scope of this test: it asserts the explicit `OneSignalDispatchers.prewarm()` call in + // Scope of this test: it asserts the explicit `OneSignalDispatchers.prewarm(any())` call in // onReceive (the goAsync() head start) happens before the ingress dispatch. // IOMockHelper stubs `suspendifyOnIngress` (run inline) and prewarm(), so this verifies // placement/ordering, not end-to-end cold-init behavior. @@ -57,9 +57,9 @@ class FCMBroadcastReceiverTests : FunSpec({ FCMBroadcastReceiver().onReceive(context, intent) - verify(exactly = 1) { OneSignalDispatchers.prewarm() } + verify(exactly = 1) { OneSignalDispatchers.prewarm(any()) } verifyOrder { - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(any()) suspendifyOnIngress(any Unit>(), any<() -> Unit>()) } } @@ -73,7 +73,7 @@ class FCMBroadcastReceiverTests : FunSpec({ FCMBroadcastReceiver().onReceive(context, intent) - verify(exactly = 0) { OneSignalDispatchers.prewarm() } + verify(exactly = 0) { OneSignalDispatchers.prewarm(any()) } } test("FCMBroadcastReceiver does not claim payloads rejected by durable ingress") { diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt index c35ede7098..fdaa85e8fe 100644 --- a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt +++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/receivers/PrewarmEntryPointTests.kt @@ -54,9 +54,9 @@ class PrewarmEntryPointTests : FunSpec({ test("BootUpReceiver.onReceive prewarms before dispatch") { val context = ApplicationProvider.getApplicationContext() BootUpReceiver().onReceive(context, Intent(Intent.ACTION_BOOT_COMPLETED)) - verify(exactly = 1) { OneSignalDispatchers.prewarm() } + verify(exactly = 1) { OneSignalDispatchers.prewarm(any()) } verifyOrder { - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(any()) suspendifyOnIngress(any Unit>(), any<() -> Unit>()) } } @@ -64,9 +64,9 @@ class PrewarmEntryPointTests : FunSpec({ test("UpgradeReceiver.onReceive prewarms before dispatch") { val context = ApplicationProvider.getApplicationContext() UpgradeReceiver().onReceive(context, Intent(Intent.ACTION_MY_PACKAGE_REPLACED)) - verify(exactly = 1) { OneSignalDispatchers.prewarm() } + verify(exactly = 1) { OneSignalDispatchers.prewarm(any()) } verifyOrder { - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(any()) suspendifyOnIngress(any Unit>(), any<() -> Unit>()) } } @@ -74,9 +74,9 @@ class PrewarmEntryPointTests : FunSpec({ test("NotificationDismissReceiver.onReceive prewarms before dispatch") { val context = ApplicationProvider.getApplicationContext() NotificationDismissReceiver().onReceive(context, Intent()) - verify(exactly = 1) { OneSignalDispatchers.prewarm() } + verify(exactly = 1) { OneSignalDispatchers.prewarm(any()) } verifyOrder { - OneSignalDispatchers.prewarm() + OneSignalDispatchers.prewarm(any()) suspendifyOnIngress(any Unit>(), any<() -> Unit>()) } } diff --git a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt index 977fd18961..a1a3a23efd 100644 --- a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt +++ b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt @@ -90,7 +90,7 @@ object IOMockHelper : BeforeSpecListener, AfterSpecListener, BeforeTestListener, // before dispatching. We mock launch*/suspendify* below to run blocks inline, so the real // prewarm()'s background daemon is pure overhead in tests — stub it to a no-op so specs // don't spawn OneSignal-prewarm threads through that entry-point code. - every { OneSignalDispatchers.prewarm() } returns Unit + every { OneSignalDispatchers.prewarm(any()) } returns Unit // Helper function to track async work (suspendifyOnIO, launchOnIO, launchOnDefault) // Note: We use runBlocking with Dispatchers.Unconfined to execute synchronously and deterministically