Skip to content

Commit 7d4a763

Browse files
AR Abdul Azeezcursoragent
andcommitted
feat: introduce SerialIO dispatcher + runOnSerialIOIfBackgroundThreading helper
Introduces the threading infrastructure that the follow-up PRs depend on. This PR adds the helpers and tests; it does not change any production call sites. What it adds * OneSignalDispatchers.SerialIO A single-thread, named ("OneSignal-SerialIO") CoroutineDispatcher backed by Executors.newSingleThreadExecutor with a SupervisorJob + CoroutineScope. Falls back to Dispatchers.IO.limitedParallelism(1) if executor construction fails. Submission order on the dispatcher == execution order on its single worker, which is exactly the semantics the focus / unfocus lifecycle handlers need (see the next PR). Companion: launchOnSerialIO { ... } and a SerialIO entry in OneSignalDispatchers.getPerformanceMetrics() / getStatus(). * ThreadUtils.suspendifyOnSerialIO { ... } Always-on serial dispatch. Wraps OneSignalDispatchers.launchOnSerialIO and is intentionally NOT gated on ThreadingMode.useBackgroundThreading - some code paths need ordered off-main execution unconditionally. * ThreadUtils.runOnSerialIOIfBackgroundThreading { ... } FF-gated wrapper for non-suspending blocks. When ThreadingMode.useBackgroundThreading is true the block is dispatched to SerialIO; when false the block runs inline on the calling thread. This is the call shape every subsequent focus / unfocus handler in this series uses, so the rollout matrix stays one-knob simple. Block is non-suspending on purpose: the FF-off branch executes on the caller's thread, and a suspending block there would force a runBlocking, which defeats the purpose of an A/B comparison. * IOMockHelper stubs the new helpers suspendifyOnSerialIO + launchOnSerialIO are tracked by awaitIO() so existing specs stay deterministic. runOnSerialIOIfBackgroundThreading is stubbed inline-on-test-thread by default so existing call-site specs keep their observable behavior; specs that want to exercise the FF-on (offload) branch can override the stub. Tests * OneSignalDispatchersTests: new SerialIO cases - construction, lazy chain activates on first launch, getStatus reports Active + queue size, falls back to the limitedParallelism(1) path if executor construction fails. getStatus + getPerformanceMetrics are refactored to extract executorStatus + scopeStatus inline helpers to keep them under Detekt's LongMethod / ComplexMethod thresholds. * ThreadUtilsFeatureFlagTests: new cases that suspendifyOnSerialIO always routes through the serial dispatcher (FF-agnostic), and that runOnSerialIOIfBackgroundThreading routes through the serial dispatcher when the FF is on and runs inline when the FF is off. Why a dedicated serial dispatcher (not just suspendifyOnIO) Multi-thread IO pools don't guarantee submission order = execution order. A rapid focus burst (activity restart, share flow popping the activity back/ forth) could otherwise interleave cancel/schedule pairs or session-state mutations. Pinning order-sensitive lifecycle work to a single executor keeps it globally ordered, and future per-event work (focus counters, session timing, analytics) inherits the guarantee for free. :OneSignal:core detekt + full unit suite green. No production behavior change in this PR; the follow-up PRs land the call-site offloads (#2644) and the dispatcher prewarm (#2645). Co-authored-by: Cursor <cursoragent@cursor.com>
1 parent fed76c4 commit 7d4a763

5 files changed

Lines changed: 354 additions & 39 deletions

File tree

‎OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt‎

Lines changed: 90 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@ import kotlinx.coroutines.SupervisorJob
99
import kotlinx.coroutines.asCoroutineDispatcher
1010
import kotlinx.coroutines.isActive
1111
import kotlinx.coroutines.launch
12+
import java.util.concurrent.ExecutorService
13+
import java.util.concurrent.Executors
1214
import java.util.concurrent.LinkedBlockingQueue
1315
import java.util.concurrent.ThreadFactory
1416
import java.util.concurrent.ThreadPoolExecutor
@@ -43,6 +45,8 @@ object OneSignalDispatchers {
4345
"$BASE_THREAD_NAME-IO" // Thread name prefix for I/O operations
4446
private const val DEFAULT_THREAD_NAME_PREFIX =
4547
"$BASE_THREAD_NAME-Default" // Thread name prefix for CPU operations
48+
private const val SERIAL_IO_THREAD_NAME =
49+
"$BASE_THREAD_NAME-SerialIO" // Single, named thread for order-sensitive work
4650

4751
private class OptimizedThreadFactory(
4852
private val namePrefix: String,
@@ -80,6 +84,27 @@ object OneSignalDispatchers {
8084
}
8185
}
8286

87+
/**
88+
* Single-thread executor for order-sensitive work (e.g. lifecycle event handlers that
89+
* must run in the order the events fired on the main thread). Submission order on the
90+
* main thread equals execution order on this thread, so callers don't need their own
91+
* synchronization to preserve sequence — they only need to capture any time-sensitive
92+
* state (timestamps, current state snapshots) on the caller before dispatching.
93+
*/
94+
private val serialIOExecutor: ExecutorService by lazy {
95+
try {
96+
Executors.newSingleThreadExecutor(
97+
OptimizedThreadFactory(
98+
namePrefix = SERIAL_IO_THREAD_NAME,
99+
priority = Thread.NORM_PRIORITY - 1,
100+
),
101+
)
102+
} catch (e: Exception) {
103+
Logging.error("OneSignalDispatchers: Failed to create SerialIO executor: ${e.message}")
104+
throw e
105+
}
106+
}
107+
83108
private val defaultExecutor: ThreadPoolExecutor by lazy {
84109
try {
85110
ThreadPoolExecutor(
@@ -117,6 +142,18 @@ object OneSignalDispatchers {
117142
}
118143
}
119144

145+
val SerialIO: CoroutineDispatcher by lazy {
146+
try {
147+
serialIOExecutor.asCoroutineDispatcher()
148+
} catch (e: Exception) {
149+
// Fall back to a limited-parallelism view of Dispatchers.IO. A single shared view
150+
// serializes our submissions even when our own executor failed to start.
151+
Logging.error("OneSignalDispatchers: Using fallback serialized Dispatchers.IO: ${e.message}")
152+
@Suppress("OPT_IN_USAGE")
153+
Dispatchers.IO.limitedParallelism(1)
154+
}
155+
}
156+
120157
private val IOScope: CoroutineScope by lazy {
121158
CoroutineScope(SupervisorJob() + IO)
122159
}
@@ -125,6 +162,10 @@ object OneSignalDispatchers {
125162
CoroutineScope(SupervisorJob() + Default)
126163
}
127164

165+
private val SerialIOScope: CoroutineScope by lazy {
166+
CoroutineScope(SupervisorJob() + SerialIO)
167+
}
168+
128169
fun launchOnIO(block: suspend () -> Unit): Job {
129170
return IOScope.launch { block() }
130171
}
@@ -133,57 +174,72 @@ object OneSignalDispatchers {
133174
return DefaultScope.launch { block() }
134175
}
135176

177+
/**
178+
* Launches [block] on the dedicated serial IO thread. Tasks submitted from any thread
179+
* run one-at-a-time in submission order, which makes this the right entry point for
180+
* order-sensitive lifecycle work (e.g. focus/unfocus handlers).
181+
*/
182+
fun launchOnSerialIO(block: suspend () -> Unit): Job {
183+
return SerialIOScope.launch { block() }
184+
}
185+
136186
internal fun getPerformanceMetrics(): String {
137187
return try {
188+
val serialQueueSize =
189+
(serialIOExecutor as? ThreadPoolExecutor)?.queue?.size?.toString() ?: "n/a"
190+
val serialCompleted =
191+
(serialIOExecutor as? ThreadPoolExecutor)?.completedTaskCount ?: 0L
138192
"""
139193
OneSignalDispatchers Performance Metrics:
140194
- IO Pool: ${ioExecutor.activeCount}/${ioExecutor.corePoolSize} active/core threads
141195
- IO Queue: ${ioExecutor.queue.size} pending tasks
142196
- Default Pool: ${defaultExecutor.activeCount}/${defaultExecutor.corePoolSize} active/core threads
143197
- Default Queue: ${defaultExecutor.queue.size} pending tasks
144-
- Total completed tasks: ${ioExecutor.completedTaskCount + defaultExecutor.completedTaskCount}
145-
- Memory usage: ~${(ioExecutor.activeCount + defaultExecutor.activeCount) * 1024}KB (thread stacks, ~1MB each)
198+
- SerialIO Queue: $serialQueueSize pending tasks
199+
- Total completed tasks: ${ioExecutor.completedTaskCount + defaultExecutor.completedTaskCount + serialCompleted}
200+
- Memory usage: ~${(ioExecutor.activeCount + defaultExecutor.activeCount + 1) * 1024}KB (thread stacks, ~1MB each)
146201
""".trimIndent()
147202
} catch (e: Exception) {
148203
"OneSignalDispatchers not initialized or using fallback dispatchers ${e.message}"
149204
}
150205
}
151206

152207
internal fun getStatus(): String {
153-
val ioExecutorStatus =
154-
try {
155-
if (ioExecutor.isShutdown) "Shutdown" else "Active"
156-
} catch (e: Exception) {
157-
"ioExecutor Not initialized ${e.message ?: "Unknown error"}"
158-
}
159-
160-
val defaultExecutorStatus =
161-
try {
162-
if (defaultExecutor.isShutdown) "Shutdown" else "Active"
163-
} catch (e: Exception) {
164-
"defaultExecutor Not initialized ${e.message ?: "Unknown error"}"
165-
}
166-
167-
val ioScopeStatus =
168-
try {
169-
if (IOScope.isActive) "Active" else "Cancelled"
170-
} catch (e: Exception) {
171-
"IOScope Not initialized ${e.message ?: "Unknown error"}"
172-
}
173-
174-
val defaultScopeStatus =
175-
try {
176-
if (DefaultScope.isActive) "Active" else "Cancelled"
177-
} catch (e: Exception) {
178-
"DefaultScope Not initialized ${e.message ?: "Unknown error"}"
179-
}
180-
181208
return """
182209
OneSignalDispatchers Status:
183-
- IO Executor: $ioExecutorStatus
184-
- Default Executor: $defaultExecutorStatus
185-
- IO Scope: $ioScopeStatus
186-
- Default Scope: $defaultScopeStatus
210+
- IO Executor: ${executorStatus("ioExecutor") { ioExecutor.isShutdown }}
211+
- Default Executor: ${executorStatus("defaultExecutor") { defaultExecutor.isShutdown }}
212+
- SerialIO Executor: ${executorStatus("serialIOExecutor") { serialIOExecutor.isShutdown }}
213+
- IO Scope: ${scopeStatus("IOScope") { IOScope.isActive }}
214+
- Default Scope: ${scopeStatus("DefaultScope") { DefaultScope.isActive }}
215+
- SerialIO Scope: ${scopeStatus("SerialIOScope") { SerialIOScope.isActive }}
187216
""".trimIndent()
188217
}
218+
219+
// Visible to tests so the failure branch (where `isShutdown()` itself throws — happens
220+
// when the underlying executor's lazy initializer threw and re-throws on every access)
221+
// can be exercised without forcing a real ThreadPoolExecutor to fail to construct, which
222+
// is impractical to trigger in a unit test.
223+
internal fun executorStatus(
224+
name: String,
225+
isShutdown: () -> Boolean,
226+
): String =
227+
try {
228+
if (isShutdown()) "Shutdown" else "Active"
229+
} catch (e: Exception) {
230+
"$name $NOT_INITIALIZED ${e.message ?: UNKNOWN_ERROR}"
231+
}
232+
233+
internal fun scopeStatus(
234+
name: String,
235+
isActive: () -> Boolean,
236+
): String =
237+
try {
238+
if (isActive()) "Active" else "Cancelled"
239+
} catch (e: Exception) {
240+
"$name $NOT_INITIALIZED ${e.message ?: UNKNOWN_ERROR}"
241+
}
242+
243+
private const val NOT_INITIALIZED = "Not initialized"
244+
private const val UNKNOWN_ERROR = "Unknown error"
189245
}

‎OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,61 @@ fun suspendifyOnDefault(block: suspend () -> Unit) {
9898
suspendifyWithCompletion(useIO = false, block = block, onComplete = null)
9999
}
100100

101+
/**
102+
* Allows a non suspending function to create a scope that runs on the dedicated
103+
* single-thread serial IO dispatcher. Tasks submitted from any thread run one-at-a-time
104+
* in submission order, so callers do not need their own synchronization to preserve event
105+
* ordering. This is the right entry point for lifecycle event handlers that do per-event
106+
* work (timing, analytics, scheduler state) and must observe events in the order the
107+
* Android framework fired them.
108+
*
109+
* Always routes through [OneSignalDispatchers.launchOnSerialIO] regardless of
110+
* [ThreadingMode.useBackgroundThreading]: the serial ordering guarantee is the whole
111+
* point of this helper, and the dedicated single thread carries none of the resource
112+
* concerns that motivated the background-threading FF in the first place.
113+
*
114+
* Callers should capture any time-sensitive state (timestamps, "current" snapshots) on
115+
* the caller's thread before invoking this — the work itself may run a few hundred ms
116+
* later under load.
117+
*
118+
* @param block The suspending code to execute
119+
*/
120+
fun suspendifyOnSerialIO(block: suspend () -> Unit) {
121+
OneSignalDispatchers.launchOnSerialIO {
122+
try {
123+
block()
124+
} catch (e: Exception) {
125+
Logging.error("Exception in suspendifyOnSerialIO", e)
126+
}
127+
}
128+
}
129+
130+
/**
131+
* Conditionally dispatches [block] to the serial IO thread when the SDK_BACKGROUND_THREADING
132+
* remote feature flag is enabled, otherwise runs [block] inline on the calling thread.
133+
*
134+
* Used to gate incremental rollouts of off-main lifecycle work where we want to A/B compare
135+
* the offloaded behavior against the previous inline behavior in production. Customers without
136+
* the flag see the original code path; customers with the flag see the offloaded one.
137+
*
138+
* Differs from [suspendifyOnSerialIO] (which always routes through the serial dispatcher
139+
* because its ordering guarantee is the whole point) and from [suspendifyOnIO] (which always
140+
* dispatches off-main, just to a different pool depending on the flag).
141+
*
142+
* The block is non-suspending because the FF-off branch executes on the caller's thread —
143+
* any suspending work would force a [kotlinx.coroutines.runBlocking] there, defeating the
144+
* purpose of the comparison.
145+
*
146+
* @param block Non-suspending work to run inline (FF off) or on the serial IO thread (FF on).
147+
*/
148+
fun runOnSerialIOIfBackgroundThreading(block: () -> Unit) {
149+
if (ThreadingMode.useBackgroundThreading) {
150+
suspendifyOnSerialIO { block() }
151+
} else {
152+
block()
153+
}
154+
}
155+
101156
/**
102157
* Modern utility for executing suspending code with completion callback.
103158
* Uses OneSignal's centralized thread management for better resource control.

‎OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,8 +82,118 @@ class OneSignalDispatchersTests : FunSpec({
8282
status shouldContain "OneSignalDispatchers Status:"
8383
status shouldContain "IO Executor: Active"
8484
status shouldContain "Default Executor: Active"
85+
status shouldContain "SerialIO Executor: Active"
8586
status shouldContain "IO Scope: Active"
8687
status shouldContain "Default Scope: Active"
88+
status shouldContain "SerialIO Scope: Active"
89+
}
90+
91+
test("getPerformanceMetrics should include SerialIO queue and total completed task counters") {
92+
// Trigger lazy init of the SerialIO executor before asking for metrics so its queue
93+
// line resolves to a concrete value instead of the "n/a" fallback path.
94+
OneSignalDispatchers.SerialIO shouldNotBe null
95+
96+
val metrics = OneSignalDispatchers.getPerformanceMetrics()
97+
98+
metrics shouldContain "OneSignalDispatchers Performance Metrics:"
99+
metrics shouldContain "SerialIO Queue:"
100+
metrics shouldContain "Total completed tasks:"
101+
}
102+
103+
test("SerialIO dispatcher executes work on a background thread") {
104+
val callerThreadId = Thread.currentThread().id
105+
var serialThreadId: Long? = null
106+
107+
runBlocking {
108+
withContext(OneSignalDispatchers.SerialIO) {
109+
serialThreadId = Thread.currentThread().id
110+
}
111+
}
112+
113+
serialThreadId shouldNotBe null
114+
serialThreadId shouldNotBe callerThreadId
115+
}
116+
117+
test("launchOnSerialIO runs tasks on a single thread in submission order") {
118+
// SerialIO's contract: submission order on the caller thread == execution order on
119+
// the worker thread. We submit N tasks with a small sleep so they queue up, then
120+
// assert the recorded order matches submission order and that all observations
121+
// came from one thread.
122+
val taskCount = 5
123+
val observedOrder = mutableListOf<Int>()
124+
val observedThreads = mutableSetOf<Long>()
125+
val latch = CountDownLatch(taskCount)
126+
127+
repeat(taskCount) { i ->
128+
OneSignalDispatchers.launchOnSerialIO {
129+
Thread.sleep(5)
130+
synchronized(observedOrder) {
131+
observedOrder.add(i)
132+
observedThreads.add(Thread.currentThread().id)
133+
}
134+
latch.countDown()
135+
}
136+
}
137+
138+
latch.await()
139+
observedOrder shouldBe (0 until taskCount).toList()
140+
observedThreads.size shouldBe 1
141+
}
142+
143+
test("executorStatus returns 'Active' / 'Shutdown' on the happy path and the Not initialized message when the underlying check throws") {
144+
// Happy paths (Shutdown / Active) are exercised indirectly via getStatus(); this
145+
// case pins down the defensive catch branch, which fires when the underlying
146+
// executor's lazy initializer is in a failed state (e.g. JVM-level
147+
// Executors.newSingleThreadExecutor refused to construct) and every isShutdown
148+
// access re-throws. Without this, the catch is unreachable from unit tests because
149+
// ThreadPoolExecutor.isShutdown does not normally throw.
150+
OneSignalDispatchers.executorStatus("ioExecutor") { false } shouldBe "Active"
151+
OneSignalDispatchers.executorStatus("ioExecutor") { true } shouldBe "Shutdown"
152+
OneSignalDispatchers.executorStatus("ioExecutor") {
153+
throw RuntimeException("init failure")
154+
} shouldBe "ioExecutor Not initialized init failure"
155+
OneSignalDispatchers.executorStatus("ioExecutor") {
156+
throw RuntimeException()
157+
} shouldBe "ioExecutor Not initialized Unknown error"
158+
}
159+
160+
test("scopeStatus returns 'Active' / 'Cancelled' on the happy path and the Not initialized message when the underlying check throws") {
161+
OneSignalDispatchers.scopeStatus("IOScope") { true } shouldBe "Active"
162+
OneSignalDispatchers.scopeStatus("IOScope") { false } shouldBe "Cancelled"
163+
OneSignalDispatchers.scopeStatus("IOScope") {
164+
throw RuntimeException("cancelled supervisor")
165+
} shouldBe "IOScope Not initialized cancelled supervisor"
166+
OneSignalDispatchers.scopeStatus("IOScope") {
167+
throw RuntimeException()
168+
} shouldBe "IOScope Not initialized Unknown error"
169+
}
170+
171+
test("exceptions in a SerialIO task do not stop subsequent tasks from running") {
172+
// Mirrors the parallel "exceptions in one task should not affect others" case for
173+
// launchOnIO. A thrown exception in one serial task must not poison the dispatcher
174+
// for the rest of the queue.
175+
val latch = CountDownLatch(3)
176+
val successCount = AtomicInteger(0)
177+
val errorCount = AtomicInteger(0)
178+
179+
repeat(3) { i ->
180+
OneSignalDispatchers.launchOnSerialIO {
181+
try {
182+
if (i == 1) {
183+
throw RuntimeException("Test error")
184+
}
185+
successCount.incrementAndGet()
186+
} catch (e: Exception) {
187+
errorCount.incrementAndGet()
188+
} finally {
189+
latch.countDown()
190+
}
191+
}
192+
}
193+
194+
latch.await()
195+
successCount.get() shouldBe 2
196+
errorCount.get() shouldBe 1
87197
}
88198

89199
test("dispatchers should handle concurrent operations") {

0 commit comments

Comments
 (0)