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 index be66d1c6e..0d3d38b56 100644 --- 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 @@ -10,53 +10,10 @@ 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, -} +import kotlin.coroutines.EmptyCoroutineContext private enum class LaneState { COLD, @@ -75,18 +32,9 @@ private class LaneTarget( } } -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. + * Queues work until the bootstrap thread has built and prestarted the lane's executor, + * so `launch` returns its Job without building anything or blocking on the caller. */ @Suppress("TooManyFunctions") private class GateDispatcher( @@ -100,34 +48,48 @@ private class GateDispatcher( private var state = LaneState.COLD private var target: LaneTarget? = null + // Built once per READY transition: dispatch runs on every resume, so it must not allocate. + private var run: GateDecision.Run? = 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 + val decision = + synchronized(lock) { + when (state) { + LaneState.COLD -> { + pending.addLast(context to block) + state = LaneState.STARTING + GateDecision.Start + } + LaneState.STARTING, LaneState.DRAINING -> { + pending.addLast(context to block) + GateDecision.Queued + } + LaneState.READY -> run!! + LaneState.CLOSED -> GateDecision.Rejected } - 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") + when (decision) { + GateDecision.Queued -> Unit + GateDecision.Start -> requestBootstrap(this) + is GateDecision.Run -> dispatchSafely(decision.dispatcher, context, block) + GateDecision.Rejected -> cancelAndComplete(context, block, "OneSignal dispatcher generation is closed") } } + private sealed interface GateDecision { + object Queued : GateDecision + + object Start : GateDecision + + class Run(val dispatcher: CoroutineDispatcher) : GateDecision + + object Rejected : GateDecision + } + fun requestWarmup() { val shouldBootstrap = synchronized(lock) { @@ -174,6 +136,7 @@ private class GateDispatcher( pending.removeFirstOrNull().also { if (it == null) { target = newTarget + run = GateDecision.Run(newTarget.dispatcher) state = LaneState.READY } } @@ -199,6 +162,7 @@ private class GateDispatcher( pending.clear() oldTarget = target target = null + run = null } queued.forEach { cancelAndComplete(it.first, it.second, "OneSignal dispatcher generation was reset") @@ -274,7 +238,7 @@ private class BootstrapCoordinator { try { Thread( ::runLoop, - "$BASE-bootstrap", + "$BASE_THREAD_NAME-bootstrap", ).apply { isDaemon = true // Not lowered: main can block waiting for init, which queues behind these builds. @@ -341,20 +305,10 @@ internal class BootstrapDispatcherPools : DispatcherPools { @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), - ) + DispatcherTestHooks.beforeLaneCreate?.invoke(lane.name) + val executor = newLaneExecutor(lane) try { - executor.allowCoreThreadTimeOut(false) - OneSignalDispatchers.afterExecutorCreateForTest?.invoke(lane.name, executor) + DispatcherTestHooks.afterExecutorCreate?.invoke(lane.name, executor) executor.prestartAllCoreThreads() } catch (t: Throwable) { executor.shutdownNow() @@ -363,52 +317,9 @@ private fun createTarget(lane: Lane): LaneTarget { 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 {}) + DispatcherTestHooks.beforeFallbackCreate?.invoke(lane.name) + val dispatcher = lane.fallbackDispatcher() + dispatcher.dispatch(EmptyCoroutineContext, Runnable {}) return LaneTarget(dispatcher, null) } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherDiagnostics.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherDiagnostics.kt new file mode 100644 index 000000000..4d53e403f --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherDiagnostics.kt @@ -0,0 +1,81 @@ +package com.onesignal.common.threading + +import kotlinx.coroutines.isActive + +private const val TEST_READY_TIMEOUT_MS = 2_000L +private const val TEST_READY_POLL_MS = 5L +private const val NOT_INITIALIZED = "Not initialized" +private const val UNKNOWN_ERROR = "Unknown error" +private const val NOT_AVAILABLE = "n/a" +private const val STACK_KB_PER_THREAD = 1024 +private val READY_STATUSES = setOf("Active", "Fallback") + +internal fun performanceMetrics(current: DispatcherPools): String { + val io = current.io.executor() + val default = current.defaultLane.executor() + val serial = current.serialIO.executor() + val ingress = current.ingress.executor() + val built = listOfNotNull(io, default, serial, ingress) + return """ + OneSignalDispatchers Performance Metrics: + - IO Pool: ${io?.let { "${it.activeCount}/${it.corePoolSize}" } ?: NOT_AVAILABLE} active/core threads + - IO Queue: ${io?.queue?.size ?: NOT_AVAILABLE} pending tasks + - Default Pool: ${default?.let { "${it.activeCount}/${it.corePoolSize}" } ?: NOT_AVAILABLE} active/core threads + - Default Queue: ${default?.queue?.size ?: NOT_AVAILABLE} pending tasks + - SerialIO Queue: ${serial?.queue?.size ?: NOT_AVAILABLE} pending tasks + - Ingress Queue: ${ingress?.queue?.size ?: NOT_AVAILABLE} pending tasks + - Total completed tasks: ${built.sumOf { it.completedTaskCount }} + - Memory usage: ~${built.sumOf { it.activeCount } * STACK_KB_PER_THREAD}KB (thread stacks, ~1MB each) + """.trimIndent() +} + +internal fun dispatcherStatus(current: DispatcherPools): String { + return """ + OneSignalDispatchers Status: + - 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 awaitPoolsReady( + pools: () -> DispatcherPools, + timeoutMs: Long = TEST_READY_TIMEOUT_MS, +): Boolean { + val deadline = System.currentTimeMillis() + timeoutMs + while (System.currentTimeMillis() < deadline) { + val current = pools() + if (listOf(current.io, current.defaultLane, current.serialIO, current.ingress).all { it.status() in READY_STATUSES }) { + return true + } + Thread.sleep(TEST_READY_POLL_MS) + } + return false +} + +@Suppress("TooGenericExceptionCaught") +internal fun executorStatus( + name: String, + isShutdown: () -> Boolean, +): String = + try { + if (isShutdown()) "Shutdown" else "Active" + } catch (e: Exception) { + "$name $NOT_INITIALIZED ${e.message ?: UNKNOWN_ERROR}" + } + +@Suppress("TooGenericExceptionCaught") +internal fun scopeStatus( + name: String, + isActive: () -> Boolean, +): String = + try { + if (isActive()) "Active" else "Cancelled" + } catch (e: Exception) { + "$name $NOT_INITIALIZED ${e.message ?: UNKNOWN_ERROR}" + } diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherLanes.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherLanes.kt new file mode 100644 index 000000000..6e2e01028 --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherLanes.kt @@ -0,0 +1,67 @@ +package com.onesignal.common.threading + +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.Dispatchers +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.AtomicInteger + +internal const val BASE_THREAD_NAME = "OneSignal" +private const val POOL_MAX = 3 +private const val BOUNDED_QUEUE = 200 +private const val KEEP_ALIVE_SECONDS = 30L + +internal enum class Lane( + val core: Int, + val max: Int, + val queueCapacity: Int, + val threadName: String, + val priority: Int, +) { + IO(2, POOL_MAX, BOUNDED_QUEUE, "$BASE_THREAD_NAME-IO", Thread.NORM_PRIORITY - 1), + DEFAULT(2, POOL_MAX, BOUNDED_QUEUE, "$BASE_THREAD_NAME-Default", Thread.NORM_PRIORITY), + + // Unbounded like Executors.newSingleThreadExecutor; rejecting would drop ordered lifecycle work. + SERIAL_IO(1, 1, Int.MAX_VALUE, "$BASE_THREAD_NAME-SerialIO", Thread.NORM_PRIORITY - 1), + + // Kept separate from IO so receiver handoffs never queue behind SDK HTTP. + INGRESS(1, 1, Int.MAX_VALUE, "$BASE_THREAD_NAME-Ingress", Thread.NORM_PRIORITY - 1), + ; + + fun fallbackDispatcher(): CoroutineDispatcher = + when (this) { + IO -> Dispatchers.IO + DEFAULT -> Dispatchers.Default + SERIAL_IO, INGRESS -> { + @Suppress("OPT_IN_USAGE") + Dispatchers.IO.limitedParallelism(1) + } + } +} + +/** Threads start on first task unless the caller prestarts them. */ +internal fun newLaneExecutor(lane: Lane): ThreadPoolExecutor = + ThreadPoolExecutor( + lane.core, + lane.max, + KEEP_ALIVE_SECONDS, + TimeUnit.SECONDS, + LinkedBlockingQueue(lane.queueCapacity), + LaneThreadFactory(lane), + ).apply { + allowCoreThreadTimeOut(false) + } + +private class LaneThreadFactory( + private val lane: Lane, +) : ThreadFactory { + private val threadNumber = AtomicInteger(1) + + override fun newThread(r: Runnable): Thread = + Thread(r, "${lane.threadName}-${threadNumber.getAndIncrement()}").apply { + isDaemon = true + priority = lane.priority + } +} diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherTestHooks.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherTestHooks.kt new file mode 100644 index 000000000..b0da2fe9e --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/DispatcherTestHooks.kt @@ -0,0 +1,40 @@ +package com.onesignal.common.threading + +import java.util.concurrent.ThreadPoolExecutor + +/** Seams for dispatcher tests. Lane hooks receive [Lane.name]. Null in production. */ +internal object DispatcherTestHooks { + @Volatile + var beforeLaneCreate: ((String) -> Unit)? = null + + @Volatile + var afterExecutorCreate: ((String, ThreadPoolExecutor) -> Unit)? = null + + @Volatile + var beforeFallbackCreate: ((String) -> Unit)? = null + + /** Lazy prewarm, before the lane lazy runs. Blocking here must not deadlock shutdown. */ + @Volatile + var beforeLazyWarm: ((String) -> Unit)? = null + + /** After the lazy prewarm thread is published, before [Thread.start]. */ + @Volatile + var beforeLazyPrewarmStart: (() -> Unit)? = null + + @Volatile + var beforePoolChoice: (() -> Unit)? = null + + /** Inside the choice report handoff, between reading readiness and parking the report. */ + @Volatile + var duringChoiceHandoff: (() -> Unit)? = null + + fun reset() { + beforeLaneCreate = null + afterExecutorCreate = null + beforeFallbackCreate = null + beforeLazyWarm = null + beforeLazyPrewarmStart = null + beforePoolChoice = null + duringChoiceHandoff = null + } +} 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 index 5fdd301f9..39e6d52e1 100644 --- 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 @@ -2,17 +2,12 @@ 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 /** @@ -22,24 +17,10 @@ import java.util.concurrent.atomic.AtomicReference @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) - } + private val ioLane = LazyExecutorLane(Lane.IO) + private val defaultExecutorLane = LazyExecutorLane(Lane.DEFAULT) + private val serialLane = LazyExecutorLane(Lane.SERIAL_IO) + private val ingressLane = LazyExecutorLane(Lane.INGRESS) override val io: DispatcherLane = ioLane override val defaultLane: DispatcherLane = defaultExecutorLane @@ -64,12 +45,12 @@ internal class LazyDispatcherPools : DispatcherPools { Thread( { // Ingress first: receivers have a ~10s broadcast budget. - warm("INGRESS") { ingressScope } - warm("IO") { ioScope } - warm("DEFAULT") { defaultScope } - warm("SERIAL_IO") { serialIOScope } + warm(Lane.INGRESS) { ingressScope } + warm(Lane.IO) { ioScope } + warm(Lane.DEFAULT) { defaultScope } + warm(Lane.SERIAL_IO) { serialIOScope } }, - "$BASE-prewarm", + "$BASE_THREAD_NAME-prewarm", ) thread.isDaemon = true // Not lowered: a launch on main blocks behind this thread's lazy builds, so it must not be starved. @@ -83,7 +64,7 @@ internal class LazyDispatcherPools : DispatcherPools { return } try { - OneSignalDispatchers.beforeLazyPrewarmStartForTest?.invoke() + DispatcherTestHooks.beforeLazyPrewarmStart?.invoke() thread.start() } catch (t: Throwable) { // start() failed after the ref was published. Clear it or every retry hits a failed CAS. @@ -105,10 +86,10 @@ internal class LazyDispatcherPools : DispatcherPools { } private fun warm( - lane: String, + lane: Lane, resolve: () -> CoroutineScope, ) { - OneSignalDispatchers.beforeLazyWarmForTest?.invoke(lane) + DispatcherTestHooks.beforeLazyWarm?.invoke(lane.name) if (closed.get()) return val scope = resolveScope(resolve) ?: return if (closed.get()) { @@ -132,13 +113,7 @@ internal class LazyDispatcherPools : DispatcherPools { } 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, + private val lane: Lane, ) : DispatcherLane { private val executorLazy: Lazy = lazy { createExecutor() } @@ -148,16 +123,16 @@ internal class LazyDispatcherPools : DispatcherPools { buildExecutor() } catch (t: Throwable) { OneSignalDispatchers.logFailure("lazy $lane fell back", t) - OneSignalDispatchers.beforeFallbackCreateForTest?.invoke(lane) + DispatcherTestHooks.beforeFallbackCreate?.invoke(lane.name) null } } private fun buildExecutor(): ThreadPoolExecutor? { - OneSignalDispatchers.beforeLaneCreateForTest?.invoke(lane) + DispatcherTestHooks.beforeLaneCreate?.invoke(lane.name) if (closed.get()) return null - val created = newExecutor() - OneSignalDispatchers.afterExecutorCreateForTest?.invoke(lane, created) + val created = newLaneExecutor(lane) + DispatcherTestHooks.afterExecutorCreate?.invoke(lane.name, created) if (closed.get()) created.shutdownNow() return created.takeUnless { it.isShutdown } } @@ -165,13 +140,13 @@ internal class LazyDispatcherPools : DispatcherPools { override val dispatcher: CoroutineDispatcher by lazy { val executor = executorLazy.value if (executor == null) { - fallback() + lane.fallbackDispatcher() } else { try { executor.asCoroutineDispatcher() } catch (t: Throwable) { OneSignalDispatchers.logFailure("lazy $lane fell back", t) - fallback() + lane.fallbackDispatcher() } } } @@ -197,42 +172,9 @@ internal class LazyDispatcherPools : DispatcherPools { 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 2904d7573..4e2a5513f 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 @@ -6,64 +6,24 @@ 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.Job -import kotlinx.coroutines.isActive import kotlinx.coroutines.launch -import java.util.concurrent.ThreadPoolExecutor /** - * Optimized threading manager for the OneSignal SDK. - * - * Performance optimizations: - * - Lazy initialization to reduce startup overhead - * - Custom thread pools for both IO and Default operations - * - Optimized thread pool configuration (smaller pools) - * - Small bounded queues (10 tasks) to prevent memory bloat - * - Reduced context switching overhead - * - Efficient thread management with controlled resource usage - * - * Made public to allow mocking in tests via IOMockHelper. + * The SDK's coroutine lanes. Pools are chosen once per process from the startup flags: + * bootstrap builds executors off the caller, lazy builds them on first use. */ -@Suppress("TooManyFunctions", "StringLiteralDuplication") +@Suppress("TooManyFunctions") object OneSignalDispatchers { - 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 RELEASE_LOG_TAG = "OneSignal" - @Volatile - internal var beforeLaneCreateForTest: ((String) -> Unit)? = null - - @Volatile - internal var afterExecutorCreateForTest: ((String, ThreadPoolExecutor) -> Unit)? = null - - @Volatile - internal var beforeFallbackCreateForTest: ((String) -> Unit)? = null - - /** Lazy prewarm, before the lane lazy runs. Blocking here must not deadlock shutdown. */ - @Volatile - internal var beforeLazyWarmForTest: ((String) -> Unit)? = null - - /** After the lazy prewarm thread is published, before [Thread.start]. */ - @Volatile - internal var beforeLazyPrewarmStartForTest: (() -> Unit)? = null - - @Volatile - internal var beforePoolChoiceForTest: (() -> Unit)? = null - private val inFailureLog = ThreadLocal() @Volatile private var poolsRef: DispatcherPools? = null private val poolsLock = Any() - /** Inside the choice report handoff, between reading readiness and parking the report. */ - @Volatile - internal var duringChoiceHandoffForTest: (() -> Unit)? = null - // Guards only the two fields below. Never report while holding it: reporting can launch. private val choiceReportLock = Any() private var telemetryReady = false @@ -83,46 +43,42 @@ object OneSignalDispatchers { FeatureManager.resetStartupFlagsForTest() } - 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 = + private fun currentPools(): DispatcherPools = poolsRef ?: publishFirstPools() + + // A launch from the choice re-enters here (poolsLock is reentrant) and publishes first; keep those pools. + // Only test hooks re-enter today: the production choice reads a prefs file and builds pools without launching. + // Logging can launch, so only the outermost call records, after publishing and outside poolsLock. + private fun publishFirstPools(): DispatcherPools { + var published: DispatcherPools? = null + val choice = synchronized(poolsLock) { - val existing = poolsRef - if (existing != null) { - choice.pools.shutdown() - existing - } else { - poolsRef = choice.pools - choice.pools + poolsRef?.let { return it } + choosePools(bootstrapEnabledForTest, AppContext::get).also { choice -> + published = poolsRef + if (published == null) poolsRef = choice.pools else choice.pools.shutdown() } } - if (published === choice.pools) recordChoice(choice) - return published + if (!Thread.holdsLock(poolsLock)) recordChoice(choice) + return published ?: choice.pools } - // 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) - } + internal fun holdsPoolsLockForTest(): Boolean = Thread.holdsLock(poolsLock) + + internal fun getPerformanceMetrics(): String = performanceMetrics(currentPools()) + + internal fun getStatus(): String = dispatcherStatus(currentPools()) + + internal fun awaitReadyForTest(): Boolean = awaitPoolsReady(::currentPools) /** 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) + logRelease(choice.summary, choice.failure, warn = choice.warn) if (!choice.warn) return val reportNow = synchronized(choiceReportLock) { val ready = telemetryReady - duringChoiceHandoffForTest?.invoke() + DispatcherTestHooks.duringChoiceHandoff?.invoke() if (!ready) pendingChoiceReport = choice ready } @@ -142,47 +98,12 @@ object OneSignalDispatchers { @Suppress("SwallowedException") private fun reportChoice(choice: PoolChoice) { try { - Logging.warn("OneSignalDispatchers: sdk_dispatcher_bootstrap ${choice.summary}", choice.error) + Logging.warn("OneSignalDispatchers: sdk_dispatcher_bootstrap ${choice.summary}", choice.failure) } catch (_: RuntimeException) { // A telemetry fault must not fail init or the launch that published the pools. } } - private fun choosePools(): PoolChoice { - beforePoolChoiceForTest?.invoke() - val override = bootstrapEnabledForTest - return if (override != null) { - PoolChoice(poolsFor(override), "test", false) - } else { - chooseFromCache() - } - } - - 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" - } - val degraded = flags.source == StartupFlagsSource.NO_STORE || flags.source == StartupFlagsSource.UNREADABLE - return PoolChoice(poolsFor(enabled), detail, warn = degraded) - } - - private fun poolsFor(enabled: Boolean): DispatcherPools = if (enabled) BootstrapDispatcherPools() else LazyDispatcherPools() - - 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"}" - } - /** Logcat at the customer's SDK log level, not through [Logging]: Logging can dispatch. */ @Suppress("SwallowedException") internal fun logRelease( @@ -266,10 +187,6 @@ object OneSignalDispatchers { } } - /** - * Test-only hook to reset [prewarmStarted] so different specs can exercise the - * "first call wins" branch independently. Not part of any public contract. - */ internal fun resetPrewarmForTest() { synchronized(prewarmLock) { prewarmStarted = false @@ -277,28 +194,13 @@ object OneSignalDispatchers { } /** - * Test-only hook to clear all in-flight and queued work between specs. - * - * Every `suspendify*` / `launchOn*` now routes unconditionally through this single - * process-wide object (bounded pools: IO core=2/max=3, Default core=2/max=3, queue cap 200). - * In the full unit-test suite, background work launched by one spec can outlive it and - * accumulate, saturating the pools and `LinkedBlockingQueue`. Worse, a spec can leave a worker - * thread parked in a non-cancellable JVM wait (e.g. a `CountDownLatch.await()` that never gets - * released because the test asserted/failed first); a plain coroutine cancellation cannot free - * such a thread, so it permanently starves the small pool. Later specs that use the real pool - * (e.g. HttpClientTests' `launchOnIO {…}.join()`, OperationRepo, LoggerIdResolver) then see - * `launchOnIO` rejected/cancelled and observe null results. - * - * This atomically swaps in a fresh [Pools] generation and tears the old one down — - * cancelling its scopes and calling `shutdownNow()` to interrupt parked workers. Safe to call - * even when this object is `mockkObject`-mocked: the affected helpers stub only - * `prewarm`/`launchOn*`/`suspendify*`, so this real method still runs. Not part of any public - * contract. + * Swaps in a fresh pool generation and `shutdownNow()`s the old one, so a worker a spec left + * parked in a blocking wait cannot starve the small pools for later specs. Clears test hooks. */ @Suppress("TooGenericExceptionCaught") internal fun resetForTest() { FeatureManager.resetStartupFlagsForTest() - val choice = choosePoolsOrLazy() + val choice = choosePools(bootstrapEnabledForTest, AppContext::get) val old: DispatcherPools? synchronized(poolsLock) { old = poolsRef @@ -311,85 +213,12 @@ object OneSignalDispatchers { Logging.error("OneSignalDispatchers.resetForTest failed: ${e.message}", e) } resetPrewarmForTest() - beforeLaneCreateForTest = null - afterExecutorCreateForTest = null - beforeFallbackCreateForTest = null - beforeLazyWarmForTest = null - beforeLazyPrewarmStartForTest = null - beforePoolChoiceForTest = null - duringChoiceHandoffForTest = null - } - - @Suppress("ComplexMethod") - internal fun getPerformanceMetrics(): String { - 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 - - IO Queue: ${io?.queue?.size ?: "n/a"} pending tasks - - Default Pool: ${default?.let { "${it.activeCount}/${it.corePoolSize}" } ?: "n/a"} active/core threads - - Default Queue: ${default?.queue?.size ?: "n/a"} pending tasks - - SerialIO Queue: ${serial?.queue?.size ?: "n/a"} pending tasks - - Ingress Queue: ${ingress?.queue?.size ?: "n/a"} pending tasks - - Total completed tasks: ${listOfNotNull(io, default, serial, ingress).sumOf { it.completedTaskCount }} - - Memory usage: ~${listOfNotNull(io, default, serial, ingress).sumOf { it.activeCount } * 1024}KB (thread stacks, ~1MB each) - """.trimIndent() - } - - internal fun getStatus(): String { - val current = currentPools() - return """ - OneSignalDispatchers Status: - - 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() + DispatcherTestHooks.reset() } - internal fun awaitReadyForTest(timeoutMs: Long = TEST_READY_TIMEOUT_MS): Boolean { - val deadline = System.currentTimeMillis() + timeoutMs - while (System.currentTimeMillis() < deadline) { - 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) - } - return false + /** Back to the first-use state, which [resetForTest] never leaves, so a spec can drive the first choice. */ + internal fun unpublishPoolsForTest() { + val old = synchronized(poolsLock) { poolsRef.also { poolsRef = null } } + old?.shutdown() } - - // internal so tests can exercise the failure branch (when `isShutdown()` itself throws, - // which happens when the lazy initializer threw and re-throws on every access). - internal fun executorStatus( - name: String, - isShutdown: () -> Boolean, - ): String = - try { - if (isShutdown()) "Shutdown" else "Active" - } catch (e: Exception) { - "$name $NOT_INITIALIZED ${e.message ?: UNKNOWN_ERROR}" - } - - internal fun scopeStatus( - name: String, - isActive: () -> Boolean, - ): String = - try { - if (isActive()) "Active" else "Cancelled" - } catch (e: Exception) { - "$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/common/threading/PoolChoice.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/PoolChoice.kt new file mode 100644 index 000000000..519582eae --- /dev/null +++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/PoolChoice.kt @@ -0,0 +1,47 @@ +package com.onesignal.common.threading + +import android.content.Context +import com.onesignal.core.internal.features.FeatureManager +import com.onesignal.features.FeatureFlag +import com.onesignal.features.StartupFlagsSource + +internal class PoolChoice( + val pools: DispatcherPools, + val detail: String, + val warn: Boolean, + val failure: Throwable? = null, +) { + val summary: String get() = "$detail; pools=${if (pools is LazyDispatcherPools) "lazy" else "bootstrap"}" +} + +// 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") +internal fun choosePools( + override: Boolean?, + context: () -> Context?, +): PoolChoice = + try { + DispatcherTestHooks.beforePoolChoice?.invoke() + if (override != null) PoolChoice(poolsFor(override), "test", warn = false) else chooseFromCache(context) + } catch (t: Throwable) { + PoolChoice(LazyDispatcherPools(), "off, choice failed", warn = true, failure = t) + } + +private fun chooseFromCache(context: () -> Context?): PoolChoice { + val flags = FeatureManager.startupFlags(context) + 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" + } + val degraded = flags.source == StartupFlagsSource.NO_STORE || flags.source == StartupFlagsSource.UNREADABLE + return PoolChoice(poolsFor(enabled), detail, warn = degraded) +} + +private fun poolsFor(enabled: Boolean): DispatcherPools = if (enabled) BootstrapDispatcherPools() else LazyDispatcherPools() 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 5adcad773..adfd15b6e 100644 --- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/OneSignalTestProjectConfig.kt +++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/OneSignalTestProjectConfig.kt @@ -1,5 +1,6 @@ package com.onesignal +import com.onesignal.common.threading.DispatcherTestHooks import com.onesignal.common.threading.OneSignalDispatchers import io.kotest.core.config.AbstractProjectConfig import io.kotest.core.extensions.Extension @@ -41,6 +42,7 @@ private object OneSignalDispatchersResetListener : BeforeSpecListener, AfterSpec // 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() { + DispatcherTestHooks.reset() 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 index 54e0cf1cd..775012f99 100644 --- 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 @@ -31,6 +31,7 @@ class DispatcherBootstrapFlagTests : FunSpec({ afterTest { reportListener?.let(Logging::removeListener) reportListener = null + DispatcherTestHooks.reset() StartupFlagsCache.write(appContext()) { emptyList() } OneSignalDispatchers.resetBootstrapInputsForTest() OneSignalDispatchers.resetForTest() @@ -73,7 +74,7 @@ class DispatcherBootstrapFlagTests : FunSpec({ 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") } + DispatcherTestHooks.beforePoolChoice = { throw IllegalStateException("choice failed") } latchFrom(context) builtOnCaller() shouldBe true @@ -82,7 +83,7 @@ class DispatcherBootstrapFlagTests : FunSpec({ 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 = { + DispatcherTestHooks.beforePoolChoice = { FeatureManager.startupFlags { context } throw IllegalStateException("choice failed") } @@ -121,7 +122,7 @@ class DispatcherBootstrapFlagTests : FunSpec({ OneSignalDispatchers.resetBootstrapInputsForTest() val inHandoff = CountDownLatch(1) val release = CountDownLatch(1) - OneSignalDispatchers.duringChoiceHandoffForTest = { + DispatcherTestHooks.duringChoiceHandoff = { inHandoff.countDown() release.await(2, TimeUnit.SECONDS) } @@ -152,7 +153,7 @@ class DispatcherBootstrapFlagTests : FunSpec({ OneSignalDispatchers.resetForTest() val priority = AtomicInteger(-1) val seen = CountDownLatch(1) - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") { priority.set(Thread.currentThread().priority) seen.countDown() @@ -170,7 +171,7 @@ class DispatcherBootstrapFlagTests : FunSpec({ OneSignalDispatchers.resetForTest() val priority = AtomicInteger(-1) val seen = CountDownLatch(1) - OneSignalDispatchers.beforeLazyWarmForTest = { + DispatcherTestHooks.beforeLazyWarm = { if (priority.compareAndSet(-1, Thread.currentThread().priority)) seen.countDown() } @@ -255,7 +256,7 @@ private fun builtOnCaller(): Boolean { val callerThreadId = AtomicLong() val createThreadId = AtomicLong() val created = CountDownLatch(1) - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") { createThreadId.set(Thread.currentThread().id) created.countDown() 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 index ed3414165..9ba98bcb9 100644 --- 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 @@ -7,6 +7,8 @@ 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 io.kotest.matchers.types.shouldBeSameInstanceAs +import kotlinx.coroutines.CoroutineDispatcher import org.robolectric.shadows.ShadowLog import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit @@ -23,6 +25,7 @@ class DispatcherFailureLogTests : FunSpec({ } afterEach { + DispatcherTestHooks.reset() OneSignalDispatchers.bootstrapEnabledForTest = null OneSignalDispatchers.resetForTest() } @@ -48,7 +51,7 @@ class DispatcherFailureLogTests : FunSpec({ usePools(bootstrap) val logged = CountDownLatch(1) val listener = ILogListener { if ("IO fell back" in it.entry) logged.countDown() } - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") throw IllegalStateException("no threads") } Logging.addListener(listener) @@ -63,7 +66,7 @@ class DispatcherFailureLogTests : FunSpec({ test("$mode pools: a listener that launches work does not re-enter lane creation") { usePools(bootstrap) val attempts = AtomicInteger() - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") { attempts.incrementAndGet() throw IllegalStateException("no threads") @@ -82,4 +85,77 @@ class DispatcherFailureLogTests : FunSpec({ attempts.get() shouldBeLessThanOrEqual 2 } } + + test("a listener that launches on a pool choice failure reuses the published pools") { + Logging.logLevel = LogLevel.NONE + OneSignalDispatchers.resetBootstrapInputsForTest() + OneSignalDispatchers.reportChoiceToTelemetry() + val choices = AtomicInteger() + DispatcherTestHooks.beforePoolChoice = { + choices.incrementAndGet() + throw IllegalStateException("choice failed") + } + val reported = CountDownLatch(1) + val listener = + ILogListener { + if ("choice failed" in it.entry) { + reported.countDown() + OneSignalDispatchers.launchOnIO { } + } + } + Logging.addListener(listener) + try { + OneSignalDispatchers.unpublishPoolsForTest() + val ran = CountDownLatch(1) + OneSignalDispatchers.launchOnIO { ran.countDown() } + ran.await(2, TimeUnit.SECONDS) shouldBe true + reported.await(2, TimeUnit.SECONDS) shouldBe true + } finally { + Logging.removeListener(listener) + OneSignalDispatchers.resetBootstrapInputsForTest() + } + + choices.get() shouldBe 1 + } + + test("a choice that re-enters keeps the pools the inner call published") { + OneSignalDispatchers.bootstrapEnabledForTest = false + var inner: CoroutineDispatcher? = null + DispatcherTestHooks.beforePoolChoice = { + if (inner == null) { + DispatcherTestHooks.beforePoolChoice = null + inner = OneSignalDispatchers.IO + } + } + OneSignalDispatchers.unpublishPoolsForTest() + + val outer = OneSignalDispatchers.IO + + outer shouldBeSameInstanceAs inner + OneSignalDispatchers.IO shouldBeSameInstanceAs inner + } + + test("a re-entrant choice is recorded once, after the outer call releases the pool lock") { + Logging.logLevel = LogLevel.NONE + OneSignalDispatchers.resetBootstrapInputsForTest() + val lockHeldWhileRecording = mutableListOf() + DispatcherTestHooks.duringChoiceHandoff = { + lockHeldWhileRecording += OneSignalDispatchers.holdsPoolsLockForTest() + } + var reentered = false + DispatcherTestHooks.beforePoolChoice = { + if (!reentered) { + reentered = true + OneSignalDispatchers.IO + } + } + try { + OneSignalDispatchers.unpublishPoolsForTest() + OneSignalDispatchers.IO + } finally { + OneSignalDispatchers.resetBootstrapInputsForTest() + } + + lockHeldWhileRecording shouldBe listOf(false) + } }) 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 index bcc1c91bb..3b279a730 100644 --- 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 @@ -28,6 +28,7 @@ class DispatcherLaneBehaviorTests : FunSpec({ } afterEach { + DispatcherTestHooks.reset() OneSignalDispatchers.resetBootstrapInputsForTest() OneSignalDispatchers.resetForTest() } @@ -36,7 +37,7 @@ class DispatcherLaneBehaviorTests : FunSpec({ val callerThreadId = AtomicLong() val createThreadId = AtomicLong() val created = CountDownLatch(1) - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") { createThreadId.set(Thread.currentThread().id) created.countDown() 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 fbd2b07a2..481a4d0b8 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 @@ -27,6 +27,7 @@ class OneSignalDispatchersTests : FunSpec({ } afterTest { + DispatcherTestHooks.reset() OneSignalDispatchers.resetBootstrapInputsForTest() OneSignalDispatchers.resetForTest() } @@ -34,7 +35,7 @@ class OneSignalDispatchersTests : FunSpec({ test("prewarm bootstraps the ingress lane before the other lanes") { OneSignalDispatchers.resetForTest() val created = Collections.synchronizedList(mutableListOf()) - OneSignalDispatchers.beforeLaneCreateForTest = { created += it } + DispatcherTestHooks.beforeLaneCreate = { created += it } try { OneSignalDispatchers.prewarm(null) @@ -54,7 +55,7 @@ class OneSignalDispatchersTests : FunSpec({ val createThreadId = AtomicLong() val callerThreadId = AtomicLong() - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") { createThreadId.set(Thread.currentThread().id) createStarted.countDown() @@ -75,7 +76,7 @@ class OneSignalDispatchersTests : FunSpec({ createThreadId.get() shouldNotBe callerThreadId.get() allowCreate.countDown() workRan.await(1, TimeUnit.SECONDS) shouldBe true - OneSignalDispatchers.beforeLaneCreateForTest = null + DispatcherTestHooks.beforeLaneCreate = null } test("cold queued launch completes when its generation is reset") { @@ -83,7 +84,7 @@ class OneSignalDispatchersTests : FunSpec({ val createStarted = CountDownLatch(1) val allowCreate = CountDownLatch(1) val completed = CountDownLatch(1) - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") { createStarted.countDown() allowCreate.await() @@ -102,18 +103,18 @@ class OneSignalDispatchersTests : FunSpec({ OneSignalDispatchers.resetForTest() val failedJobCompleted = CountDownLatch(1) val defaultWorkRan = CountDownLatch(1) - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") throw IllegalStateException("primary failed") } - OneSignalDispatchers.beforeFallbackCreateForTest = { lane -> + DispatcherTestHooks.beforeFallbackCreate = { lane -> if (lane == "IO") throw AssertionError("fallback failed") } OneSignalDispatchers.launchOnIO {}.invokeOnCompletion { failedJobCompleted.countDown() } failedJobCompleted.await(1, TimeUnit.SECONDS) shouldBe true - OneSignalDispatchers.beforeLaneCreateForTest = null - OneSignalDispatchers.beforeFallbackCreateForTest = null + DispatcherTestHooks.beforeLaneCreate = null + DispatcherTestHooks.beforeFallbackCreate = null OneSignalDispatchers.launchOnDefault { defaultWorkRan.countDown() } defaultWorkRan.await(1, TimeUnit.SECONDS) shouldBe true } @@ -122,7 +123,7 @@ class OneSignalDispatchersTests : FunSpec({ OneSignalDispatchers.resetForTest() val workRan = CountDownLatch(1) var partial: ThreadPoolExecutor? = null - OneSignalDispatchers.afterExecutorCreateForTest = { lane, executor -> + DispatcherTestHooks.afterExecutorCreate = { lane, executor -> if (lane == "IO") { partial = executor executor.prestartCoreThread() @@ -143,7 +144,7 @@ class OneSignalDispatchersTests : FunSpec({ val allowCreate = CountDownLatch(1) val bootstrapFinished = CountDownLatch(1) val ioCreateCount = AtomicInteger() - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") { ioCreateCount.incrementAndGet() createStarted.countDown() @@ -151,7 +152,7 @@ class OneSignalDispatchersTests : FunSpec({ throw IllegalStateException("primary failed") } } - OneSignalDispatchers.beforeFallbackCreateForTest = { lane -> + DispatcherTestHooks.beforeFallbackCreate = { lane -> if (lane == "IO") { bootstrapFinished.countDown() throw AssertionError("fallback failed") @@ -161,10 +162,10 @@ class OneSignalDispatchersTests : FunSpec({ OneSignalDispatchers.launchOnIO {} createStarted.await(1, TimeUnit.SECONDS) shouldBe true - val hooks = OneSignalDispatchers.beforeLaneCreateForTest to OneSignalDispatchers.beforeFallbackCreateForTest + val hooks = DispatcherTestHooks.beforeLaneCreate to DispatcherTestHooks.beforeFallbackCreate OneSignalDispatchers.resetForTest() - OneSignalDispatchers.beforeLaneCreateForTest = hooks.first - OneSignalDispatchers.beforeFallbackCreateForTest = hooks.second + DispatcherTestHooks.beforeLaneCreate = hooks.first + DispatcherTestHooks.beforeFallbackCreate = hooks.second allowCreate.countDown() bootstrapFinished.await(1, TimeUnit.SECONDS) shouldBe true @@ -182,23 +183,23 @@ class OneSignalDispatchersTests : FunSpec({ // Executors.newSingleThreadExecutor refused to construct) and every isShutdown // access re-throws. Without this, the catch is unreachable from unit tests because // ThreadPoolExecutor.isShutdown does not normally throw. - OneSignalDispatchers.executorStatus("ioExecutor") { false } shouldBe "Active" - OneSignalDispatchers.executorStatus("ioExecutor") { true } shouldBe "Shutdown" - OneSignalDispatchers.executorStatus("ioExecutor") { + executorStatus("ioExecutor") { false } shouldBe "Active" + executorStatus("ioExecutor") { true } shouldBe "Shutdown" + executorStatus("ioExecutor") { throw RuntimeException("init failure") } shouldBe "ioExecutor Not initialized init failure" - OneSignalDispatchers.executorStatus("ioExecutor") { + executorStatus("ioExecutor") { throw RuntimeException() } shouldBe "ioExecutor Not initialized Unknown error" } test("scopeStatus returns 'Active' / 'Cancelled' on the happy path and the Not initialized message when the underlying check throws") { - OneSignalDispatchers.scopeStatus("IOScope") { true } shouldBe "Active" - OneSignalDispatchers.scopeStatus("IOScope") { false } shouldBe "Cancelled" - OneSignalDispatchers.scopeStatus("IOScope") { + scopeStatus("IOScope") { true } shouldBe "Active" + scopeStatus("IOScope") { false } shouldBe "Cancelled" + scopeStatus("IOScope") { throw RuntimeException("cancelled supervisor") } shouldBe "IOScope Not initialized cancelled supervisor" - OneSignalDispatchers.scopeStatus("IOScope") { + scopeStatus("IOScope") { throw RuntimeException() } shouldBe "IOScope Not initialized Unknown error" } @@ -209,7 +210,7 @@ class OneSignalDispatchersTests : FunSpec({ val callerThreadId = AtomicLong() val createThreadId = AtomicLong() val finished = CountDownLatch(1) - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") createThreadId.set(Thread.currentThread().id) } @@ -229,7 +230,7 @@ class OneSignalDispatchersTests : FunSpec({ OneSignalDispatchers.resetForTest() val order = Collections.synchronizedList(mutableListOf()) val created = CountDownLatch(4) - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> order.add(lane) created.countDown() } @@ -244,7 +245,7 @@ class OneSignalDispatchersTests : FunSpec({ 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 = { + DispatcherTestHooks.beforeLazyWarm = { entered.countDown() awaitIgnoringInterrupts(release) } @@ -252,7 +253,7 @@ class OneSignalDispatchersTests : FunSpec({ entered.await(2, TimeUnit.SECONDS) shouldBe true OneSignalDispatchers.resetForTest() val builtAfterReset = AtomicInteger() - OneSignalDispatchers.afterExecutorCreateForTest = { _, _ -> builtAfterReset.incrementAndGet() } + DispatcherTestHooks.afterExecutorCreate = { _, _ -> builtAfterReset.incrementAndGet() } release.countDown() Thread.sleep(400) builtAfterReset.get() shouldBe 0 @@ -263,7 +264,7 @@ class OneSignalDispatchersTests : FunSpec({ OneSignalDispatchers.resetForTest() val entered = CountDownLatch(1) val release = CountDownLatch(1) - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "INGRESS") { entered.countDown() awaitIgnoringInterrupts(release) @@ -273,7 +274,7 @@ class OneSignalDispatchersTests : FunSpec({ entered.await(2, TimeUnit.SECONDS) shouldBe true OneSignalDispatchers.resetForTest() val builtAfterReset = AtomicInteger() - OneSignalDispatchers.afterExecutorCreateForTest = { _, _ -> builtAfterReset.incrementAndGet() } + DispatcherTestHooks.afterExecutorCreate = { _, _ -> builtAfterReset.incrementAndGet() } release.countDown() Thread.sleep(400) builtAfterReset.get() shouldBe 0 @@ -285,7 +286,7 @@ class OneSignalDispatchersTests : FunSpec({ val entered = CountDownLatch(1) val release = CountDownLatch(1) val built = AtomicReference() - OneSignalDispatchers.afterExecutorCreateForTest = { lane, executor -> + DispatcherTestHooks.afterExecutorCreate = { lane, executor -> if (lane == "INGRESS") { built.set(executor) entered.countDown() @@ -303,7 +304,7 @@ class OneSignalDispatchersTests : FunSpec({ test("flag off reports Fallback for a lane that fell back") { OneSignalDispatchers.bootstrapEnabledForTest = false OneSignalDispatchers.resetForTest() - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") throw IllegalStateException("boom") } val ran = CountDownLatch(1) @@ -314,7 +315,7 @@ class OneSignalDispatchersTests : FunSpec({ } test("bootstrap reports Fallback for a lane that fell back") { - OneSignalDispatchers.beforeLaneCreateForTest = { lane -> + DispatcherTestHooks.beforeLaneCreate = { lane -> if (lane == "IO") throw IllegalStateException("boom") } val ran = CountDownLatch(1) @@ -328,8 +329,8 @@ class OneSignalDispatchersTests : FunSpec({ OneSignalDispatchers.bootstrapEnabledForTest = false OneSignalDispatchers.resetForTest() val fallbacks = AtomicInteger() - OneSignalDispatchers.beforeLaneCreateForTest = { throw IllegalStateException("boom") } - OneSignalDispatchers.beforeFallbackCreateForTest = { fallbacks.incrementAndGet() } + DispatcherTestHooks.beforeLaneCreate = { throw IllegalStateException("boom") } + DispatcherTestHooks.beforeFallbackCreate = { fallbacks.incrementAndGet() } val ran = CountDownLatch(1) OneSignalDispatchers.launchOnIO { ran.countDown() } ran.await(2, TimeUnit.SECONDS) shouldBe true @@ -339,12 +340,12 @@ class OneSignalDispatchersTests : FunSpec({ test("flag off prewarm retries when the prewarm thread fails to start") { OneSignalDispatchers.bootstrapEnabledForTest = false OneSignalDispatchers.resetForTest() - OneSignalDispatchers.beforeLazyPrewarmStartForTest = { throw IllegalStateException("start failed") } + DispatcherTestHooks.beforeLazyPrewarmStart = { throw IllegalStateException("start failed") } OneSignalDispatchers.prewarm(null) - OneSignalDispatchers.beforeLazyPrewarmStartForTest = null + DispatcherTestHooks.beforeLazyPrewarmStart = null val created = CountDownLatch(4) - OneSignalDispatchers.beforeLaneCreateForTest = { created.countDown() } + DispatcherTestHooks.beforeLaneCreate = { created.countDown() } OneSignalDispatchers.prewarm(null) created.await(2, TimeUnit.SECONDS) shouldBe true }