WIP Add Kotlin coroutine examples#172
Conversation
WalkthroughThis PR introduces a comprehensive coroutine-based task execution framework including a generic BaseTask abstract class, 13 specialized task implementations (Prime calculation, sorting, encryption, compression, neural networks, etc.), a TaskViewModel for managing concurrent task execution, progress and status tracking via StateFlow, and extensive test coverage. Changes
Sequence Diagram(s)sequenceDiagram
participant Main as Client<br/>(main())
participant VM as TaskViewModel
participant Scope as CoroutineScope<br/>(SupervisorJob)
participant Task as BaseTask<br/>(execute)
participant Flow as StateFlow<br/>(progress/status)
Main->>VM: addTask(task1...14)
Main->>VM: runAllTasks()
VM->>Scope: launch runTask(task) x14
loop For each task
Scope->>Task: run() [suspend]
Task->>Flow: emit(RUNNING)
Task->>Task: execute() [suspend]
par Progress Updates
Task->>Flow: emit(progress: 0.0..1.0)
and Execution
Task->>Task: compute workload
end
alt Success
Task->>Flow: emit(COMPLETED)
Scope->>VM: addResult(TaskResult)
else Cancelled
Scope->>Task: cancel()
Task->>Flow: emit(CANCELLED)
Scope->>VM: addResult(TaskResult)
else Error
Task->>Flow: emit(ERROR)
Scope->>VM: addResult(TaskResult)
end
end
par Monitor Progress
Main->>VM: observe tasks Flow
VM-->>Main: emit task list updates
and Monitor Results
Main->>VM: observe results Flow
VM-->>Main: emit result updates
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
🧹 Nitpick comments (83)
src/test/kotlin/dev/shtanko/collections/concurrent/ArrayBlockingQueueLinearizabilityTest.kt (1)
30-33: Prefer conditional disable over blanket @disabled to keep local runs activeUsing
@Disabledat class level prevents the test from running anywhere, which makes theassumeTrueCI guard redundant. If the intent is to skip only on CI, use JUnit 5 conditional annotations.Apply this diff to scope skipping to CI environments:
-import org.junit.jupiter.api.Disabled +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable @@ -@Disabled("Time consuming test") +@DisabledIfEnvironmentVariable(named = "CI", matches = "true")Follow-up: with this change, the
assumeTrue(System.getenv("CI") != "true")block becomes redundant and can be removed to reduce noise.src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/TaskStatus.kt (1)
1-5: Add license header and brief KDoc for consistency and clarityOther Kotlin files carry the Apache 2.0 header; add it here for consistency. Also, a short KDoc describing each state improves readability and API discoverability.
Apply this diff to align with project conventions:
+/* + * Designed and developed by 2022 ashtanko (Oleksii Shtanko) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package dev.shtanko.concurrency.coroutines.tasks +/** + * Lifecycle states for coroutine-based tasks. + * + * IDLE: task not started + * RUNNING: task executing + * COMPLETED: finished successfully + * CANCELLED: cancelled by user or system + * ERROR: finished with an unrecoverable error + */ enum class TaskStatus { IDLE, RUNNING, COMPLETED, CANCELLED, ERROR }Note: spelling of CANCELLED vs CANCELED varies by locale; verify it’s consistent across the new API (Task, BaseTask, TaskViewModel, tests).
README.md (2)
23-27: Confirm metrics are auto-generated to avoid driftThese numbers tend to drift quickly. If you already have a generator, consider linking it here or wiring a CI job to refresh on main merges.
I can provide a Gradle task or simple script to regenerate and update this block automatically. Want me to draft it?
33-38: Same for complexity stats: ensure there’s a reproducible generation stepLock in a single source of truth (e.g., CI artifact) to prevent stale complexity data in README.
If these are produced by an internal tool, document the command in CONTRIBUTING.md or reference it here for maintainers.
src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/TaskResult.kt (1)
3-8: Consider richer typing and diagnostics for resultsExecution time as
Longis opaque; usingkotlin.time.Durationis more expressive. Also, capturing an optional error cause improves debuggability forERRORstatus.Apply this diff to improve the API:
data class TaskResult<T>( val taskName: String, val result: T?, - val executionTime: Long, + val executionTime: kotlin.time.Duration, - val status: TaskStatus, + val status: TaskStatus, + val exception: Throwable? = null, )Outside this hunk, add the import at the top of the file:
import kotlin.time.DurationOptional: if results are emitted over Flows to UI or persisted, consider adding Kotlinx Serialization support:
@Serializable data class TaskResult<@Contextual T>(...)Also, align header/licensing with other files (Apache 2.0 header) for consistency.
src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/ImageProcessingTaskTest.kt (1)
21-34: Strengthen assertions for different image sizesAsserting only non-null is weak; validate the format and numeric bounds like in the first test to catch regressions in output formatting or value range.
for ((width, height) in sizes) { val task = ImageProcessingTask(width, height) val result = task.run() - assertNotNull(result) + assertNotNull(result) + assertTrue(result.contains("Avg brightness:")) + val brightness = result.substringAfter("Avg brightness:").trim().toDoubleOrNull() + assertNotNull(brightness) { "Could not parse brightness from: '$result' (size=${width}x$height)" } + val value = requireNotNull(brightness) + assertTrue(value in 0.0..255.0) }src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/Main.kt (2)
11-25: Extremely heavy default workloads may make the demo impractical to runSeveral tasks use very large inputs (e.g., 10k×10k Mandelbrot, 50M–100M iterations) which will take a long time and can exhaust memory/CPU when run via the default main. Consider gating heavy sizes behind an env flag or CLI arg and providing lighter defaults for casual runs.
-fun main(): Unit = runBlocking { +fun main(): Unit = runBlocking { val viewModel = TaskViewModel() - viewModel.addTask(PrimeCalculationTask(5000_0000)) - viewModel.addTask(MatrixMultiplicationTask(2000)) - viewModel.addTask(MandelbrotTask(10000, 10000)) - viewModel.addTask(SortingTask(100_000)) - viewModel.addTask(HashComputationTask(10000_0000)) + val heavy = System.getenv("HEAVY_DEMO")?.toBoolean() == true + viewModel.addTask(PrimeCalculationTask(if (heavy) 50_000_000 else 50_000)) + viewModel.addTask(MatrixMultiplicationTask(if (heavy) 2000 else 300)) + viewModel.addTask(MandelbrotTask(if (heavy) 10_000 else 600, if (heavy) 10_000 else 400)) + viewModel.addTask(SortingTask(if (heavy) 100_000 else 10_000)) + viewModel.addTask(HashComputationTask(if (heavy) 100_000_000 else 1_000_000)) - viewModel.addTask(CryptographicTask(50000)) - viewModel.addTask(CompressionTask(50_0000)) - viewModel.addTask(BinaryTreeTask(30_0000)) - viewModel.addTask(GraphAlgorithmsTask(5000)) - viewModel.addTask(StringMatchingTask(500_0000, 50)) - viewModel.addTask(ImageProcessingTask(600, 400)) - viewModel.addTask(NeuralNetworkTask(epochs = 5)) - viewModel.addTask(MergeSortTask(900_000)) - viewModel.addTask(QuickSortTask(900_000)) + viewModel.addTask(CryptographicTask(if (heavy) 50_000 else 5_000)) + viewModel.addTask(CompressionTask(if (heavy) 500_000 else 50_000)) + viewModel.addTask(BinaryTreeTask(if (heavy) 300_000 else 30_000)) + viewModel.addTask(GraphAlgorithmsTask(if (heavy) 5_000 else 500)) + viewModel.addTask(StringMatchingTask(if (heavy) 5_000_000 else 500_000, 50)) + viewModel.addTask(ImageProcessingTask(600, 400)) + viewModel.addTask(NeuralNetworkTask(epochs = if (heavy) 5 else 2)) + viewModel.addTask(MergeSortTask(if (heavy) 900_000 else 90_000)) + viewModel.addTask(QuickSortTask(if (heavy) 900_000 else 90_000))If you prefer CLI args, I can sketch a tiny parser instead.
27-37: Graceful shutdown of collectorsCollectors are fine for a short-lived program, but if viewModel.results/tasks are hot flows that never complete, they’ll keep the process alive if additional work is added later. Consider scoping and canceling them explicitly after runAllTasks() completes, or using takeWhile/timeout to auto-complete in demos.
- launch { - viewModel.tasks.collect { - println("TASK: ${it}") - } - } + val tasksJob = launch { + viewModel.tasks.collect { println("TASK: $it") } + } - launch { - viewModel.results.collect { - println("RES: ${it}") - } - } + val resultsJob = launch { + viewModel.results.collect { println("RES: $it") } + } viewModel.runAllTasks() + // In case flows do not complete on their own: + tasksJob.cancel() + resultsJob.cancel()src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/PrimeCalculationTaskTest.kt (2)
40-51: Join the runner and cancel the collector to avoid leaked coroutines in runTestWithout waiting for the launched run() or canceling the collector, runTest can report unfinished coroutines or produce flakiness. Join the worker and cancelAndJoin the collector after advancing.
- val progressValues = mutableListOf<Float>() - val job = launch { - task.progress.collect { progressValues.add(it) } - } - - launch { task.run() } - advanceUntilIdle() - job.cancel() + val progressValues = mutableListOf<Float>() + val collector = launch { + task.progress.collect { progressValues.add(it) } + } + val worker = launch { task.run() } + advanceUntilIdle() + worker.join() + collector.cancelAndJoin()
87-98: Duplicated isPrime implementation in testThe helper mirrors production logic. Consider importing a tested, shared implementation or placing this helper in a shared test utility to avoid divergence over time.
src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/NeuralNetworkTaskTest.kt (1)
26-49: Avoid leaving the training job running; ensure cleanup with TurbineYou launch task.run() but break collection at 0.9f, which can leave the runner active. Ensure cleanup by canceling or joining the runner and canceling the Turbine collection to avoid leaked coroutines under runTest.
- task.progress.test { + task.progress.test { assertEquals(0f, awaitItem()) - launch { task.run() } + val runner = launch { task.run() } // Collect progress updates val updates = mutableListOf<Float>() while (true) { val item = awaitItem() updates.add(item) - if (item >= 0.9f) break + if (item >= 0.9f) break } assertTrue(updates.size > 2) // Progress should increase for (i in 1 until updates.size) { assertTrue(updates[i] >= updates[i - 1]) } + // Ensure coroutines are cleaned up for runTest + runner.cancelAndJoin() + cancelAndIgnoreRemainingEvents() }src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/MandelbrotTask.kt (1)
7-14: Minor: validate inputs to fail fastNegative or zero width/height/maxIterations silently produce odd behavior. Consider require() guards.
class MandelbrotTask( private val width: Int = 400, private val height: Int = 300, private val maxIterations: Int = 256, ) : BaseTask<Int>( name = "Mandelbrot Set", description = "Calculating ${width}x$height fractal", ) { + init { + require(width > 0 && height > 0) { "width and height must be positive" } + require(maxIterations > 0) { "maxIterations must be positive" } + }src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/CompressionTaskTest.kt (2)
9-18: Observe status before invoking run() to assert full lifecycle and avoid missed emissionsSubscribing to
statusafterrun()returns only validates the terminal state. Asserting the full sequence (IDLE → RUNNING → COMPLETED) makes the test stronger and avoids race/flakiness.Apply this diff to restructure the test:
@@ - fun `should compress data and return ratio`() = runTest { - val task = CompressionTask(dataSize = 1000) - val ratio = task.run() - task.status.test { - assertTrue(awaitItem() == TaskStatus.COMPLETED) - assertTrue(ratio > 0) - assertTrue(ratio < 1.3) // Compression should reduce size - } - } + fun `should compress data and return ratio`() = runTest { + val task = CompressionTask(dataSize = 1000) + task.status.test { + // Initial state + assertEquals(TaskStatus.IDLE, awaitItem()) + + // Start work + val ratio = task.run() + + // Lifecycle states + assertEquals(TaskStatus.RUNNING, awaitItem()) + assertEquals(TaskStatus.COMPLETED, awaitItem()) + + // Result assertions + assertTrue(ratio > 0) + assertTrue(ratio < 1.3) // Compression should reduce size + + cancelAndIgnoreRemainingEvents() + } + }
20-31: The test name says “should vary” but the assertions don’t check variationYou only check bounds, not that ratios differ across sizes. Add a minimal check that at least two ratios differ to reflect the test’s intent.
@@ - assertTrue(ratios.all { it > 0 && it <= 2 }) + assertTrue(ratios.all { it > 0 && it <= 2 }) + // Ensure there is some variation across sizes + assertTrue(ratios.toSet().size > 1, "Expected ratios to vary across data sizes")src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/MatrixMultiplicationTask.kt (3)
28-31: Progress may never reach 100%If
size*sizeisn’t a multiple of 100, the last update may be < 1.0. Emit a final progress update on completion.completed++ - if (completed % 100 == 0) { + if (completed % 100 == 0) { updateProgress(completed.toFloat() / totalOperations) } }Add a final update before returning (see next comment for exact placement).
34-36: Finalize progress to 1.0 before returningGuarantees consistent UI semantics irrespective of intermediate update cadence.
// Return sum of diagonal elements as result - return (0 until size).sumOf { result[it][it] } + updateProgress(1f) + return (0 until size).sumOf { result[it][it] }
6-11: Avoid magic numbers and make progress cadence configurableExtract
100into a constant to improve readability and maintainability. Also consider usingLongfortotalOperationsin casesizegrows, to avoid overflow in intermediate math.- val totalOperations = size * size + val totalOperations = size.toLong() * size.toLong() var completed = 0Additionally, add this companion object and use it in the modulo check (outside the selected lines):
// Place inside class body private companion object { const val PROGRESS_UPDATE_STEP = 100 }And replace the modulo check (in the loop) accordingly:
- if (completed % 100 == 0) { + if (completed % PROGRESS_UPDATE_STEP == 0) {src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/FlowTestingWithTurbine.kt (3)
30-31: Prefer cancelling the Turbine subscription to assert completion
expectNoEvents()is fine but leaving the collector open can hide leaks in more complex scenarios. Explicitly cancel to be safe.- expectNoEvents() + cancelAndIgnoreRemainingEvents()
44-49: Avoid asserting on Turbine’s error message textString messages may change across versions. Asserting the exception type is enough and more stable.
- val error = assertThrows<AssertionError> { - awaitItem() - } - assertTrue(error.message?.contains("No value produced") == true) + assertThrows<AssertionError> { + awaitItem() + }
35-43: Optional: make timing fully deterministic under runTestDelays inside
TestTimeoutTaskrely on virtual time semantics. If you ever switch to real-time timeouts or upgrade libs, consider advancing virtual time explicitly or using aStandardTestDispatcher-backedTaskViewModelto eliminate flakiness.Would you like me to introduce a
TestCoroutineDispatcher/StandardTestDispatcherinjection intoTaskViewModeland update these tests accordingly?src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/PrimeCalculationTask.kt (3)
17-33: Progress not finalized to 1.0If
limit % 1000 != 0, the last update may reflect < 100%. Emit a final update after the loop.for (num in 2..limit) { if (num % 1000 == 0) { yield() updateProgress(num.toFloat() / limit) } if (isPrime(num)) { primes.add(num) } } - return primes + updateProgress(1f) + return primes
39-44: Nit: guard against potential overflow in the loop conditionFor very large
n,i * ican overflow. Usingi <= n / iavoids that. Not an issue for current defaults, but it’s a safer idiom.- while (i * i <= n) { + while (i <= n / i) {
20-29: Optional micro-optimization: skip even numbers in the main loopYou already weed out non-primes in
isPrime, but iterating only odds halves the number of calls.- for (num in 2..limit) { + // Handle 2 explicitly, then only check odd numbers + if (limit >= 2) primes.add(2) + for (num in 3..limit step 2) {Remember to adjust the progress updates accordingly if you adopt this.
src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/CryptographicTask.kt (2)
25-31: Progress may not reach 100%When
messageCount % 50 != 0, final progress can be < 1.0. Emit a final update after the loop.for (i in 0 until messageCount) { if (i % 50 == 0) { yield() updateProgress(i.toFloat() / messageCount) } val message = Random.nextInt(1, n) @@ - return encryptedCount + updateProgress(1f) + return encryptedCount
59-66: Replace brute force modular inverse with Extended Euclidean algorithm and fail fast if no inverseThe current fallback
return 1silently masks errors. Use EEA for O(log m) complexity and throw if no inverse exists.- private fun modInverse(a: Int, m: Int): Int { - for (x in 1 until m) { - if ((a * x) % m == 1) { - return x - } - } - return 1 - } + private fun modInverse(a: Int, m: Int): Int { + var t = 0L + var newT = 1L + var r = m.toLong() + var newR = (a % m + m) % m.toLong() + while (newR != 0L) { + val q = r / newR + val tmpT = t - q * newT + t = newT + newT = tmpT + val tmpR = r - q * newR + r = newR + newR = tmpR + } + if (r != 1L) { + throw IllegalArgumentException("No modular inverse for a=$a mod m=$m") + } + if (t < 0) t += m + return t.toInt() + }src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/NeuralNetworkTask.kt (4)
1-6: Suppress MagicNumber to keep ktlint/detekt quiet (consistent with other tasks).You’re using several literals (100, 50, etc.). Other files suppress MagicNumber at file level; do the same here for consistency and to avoid lint noise.
+@file:Suppress("MagicNumber") package dev.shtanko.concurrency.coroutines.tasks import kotlin.math.exp import kotlin.random.Random import kotlinx.coroutines.yield
22-31: Make progress computation overflow-safe and slightly more precise.When epochs and/or samples become large, (epochs * samples) and (epoch * samples + sample) can overflow Int. Compute on Long and convert to Float at the end. Also reuse a single “step” counter to avoid recomputation.
// Training samples val samples = 100 var totalLoss = 0.0 - for (epoch in 0 until epochs) { + val totalSteps = epochs.toLong() * samples + for (epoch in 0 until epochs) { for (sample in 0 until samples) { - if ((epoch * samples + sample) % 50 == 0) { - yield() - updateProgress((epoch * samples + sample).toFloat() / (epochs * samples)) - } + val step = epoch.toLong() * samples + sample + if (step % 50L == 0L) { + yield() + updateProgress((step.toDouble() / totalSteps.toDouble()).toFloat()) + }
63-64: Ensure progress reaches 100% and guard divide-by-zero when epochs or samples are zero.This avoids progress stopping below 100% and prevents a crash when totalSteps == 0.
- return totalLoss / (epochs * samples) + // Finalize progress and compute average loss safely + updateProgress(1f) + val totalSteps = epochs.toLong() * 100 // matches `samples` above + return if (totalSteps == 0L) 0.0 else totalLoss / totalSteps.toDouble()Note: If you make
samplesconfigurable later, recompute totalSteps accordingly or lift it outside for reuse.
13-15: Clarify task description (no weight updates = no “training”).Weights are never updated (no backprop/gradient step). Either rename the description or implement a minimal learning step. Renaming is simpler and avoids misleading users.
) : BaseTask<Double>( name = "Neural Network Training", - description = "Training ${hiddenSize}-node network for $epochs epochs", + description = "Simulating forward passes on a ${hiddenSize}-node network for $epochs epochs", )If you want a tiny “training” touch, I can suggest a simple SGD step for the HO layer only.
src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/HashComputationTask.kt (2)
16-23: Prefer repeat for clarity and ensure progress math is stable.Functionally equivalent, a bit clearer, and keeps progress logic identical.
- for (i in 0 until iterations) { - if (i % 10_000 == 0) { - yield() - updateProgress(i.toFloat() / iterations) - } - - hash = computeHash(hash, i) - } + repeat(iterations) { i -> + if (i % 10_000 == 0) { + yield() + updateProgress(i.toFloat() / iterations) + } + hash = computeHash(hash, i) + }
25-26: Force progress to 100% at completion.If iterations is not a multiple of 10_000, progress may finish below 1.0; set it explicitly at the end.
- return hash.toString(16).uppercase() + updateProgress(1f) + return hash.toString(16).uppercase()src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/PerformanceTest.kt (1)
31-61: Memory test is inherently flaky; avoid using it in unit tests or gate it.GC timing and heap pressure vary per environment; even with runBlocking, results will be noisy. Consider disabling in CI or converting into a benchmark.
Possible mitigations:
- Annotate with @disabled("Non-deterministic memory test; run locally only")
- Or gate with an env var:
+import org.junit.jupiter.api.Assumptions.assumeTrue @@ - fun `should handle memory efficiently`() = runTest { + fun `should handle memory efficiently`() = runBlocking { + assumeTrue(System.getenv("RUN_PERF_TESTS") == "1") { + "Skipping non-deterministic memory test on CI" + } val runtime = Runtime.getRuntime() val beforeMemory = runtime.totalMemory() - runtime.freeMemory() @@ System.gc() delay(100) @@ assertTrue( memoryIncrease < 100 * 1024 * 1024, "Memory increased by ${memoryIncrease / 1024 / 1024}MB" ) }If you want, I can extract these into JMH benchmarks instead.
src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/MergeSortTask.kt (3)
28-30: Finalize progress at completion.Ensures UI observers see 100% even if the last yield/progress checkpoint didn’t align exactly.
mergeSort(array, 0, arraySize - 1) - return comparisons + updateProgress(1f) + return comparisons
41-44: Avoid division by zero if totalOperations is 0 (tiny arrays).For arraySize <= 1, totalOperations can be 0. While this block won’t execute for such sizes, guarding the denominator is cheap and robust.
- if (currentOperation % 1000 == 0) { + if (currentOperation % 1000 == 0) { yield() - updateProgress(currentOperation.toFloat() / totalOperations) + updateProgress(currentOperation.toFloat() / maxOf(totalOperations, 1)) }
48-51: Reduce per-merge allocations (optional).sliceArray creates ranges and copies. copyOfRange is leaner; for best perf, consider a single reusable temp buffer for merges, but here’s a minimal improvement.
- val leftArray = arr.sliceArray(left..mid) - val rightArray = arr.sliceArray(mid + 1..right) + val leftArray = arr.copyOfRange(left, mid + 1) + val rightArray = arr.copyOfRange(mid + 1, right + 1)If you want the reusable-buffer variant, I can provide a refactor that allocates one temp array and uses it across merges.
src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/Task.kt (1)
5-12: Make Task covariant to improve API flexibility.Covariance lets Task be used where Task is expected (safe since R is only returned). This helps when aggregating heterogeneous tasks.
-interface Task<R> { +interface Task<out R> { val progress: StateFlow<Float> val status: StateFlow<TaskStatus> val name: String val description: String suspend fun run(): R fun cancel() }src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/BaseTaskTest.kt (3)
121-134: Join the cancelled job to avoid races before asserting CANCELLEDCanceling the launched job and immediately asserting CANCELLED can race on slower machines. Join the job to ensure state propagation before asserting.
Apply this diff:
val job = launch { task.run() } Assertions.assertEquals(TaskStatus.RUNNING, awaitItem()) job.cancel() + job.join() Assertions.assertEquals(TaskStatus.CANCELLED, awaitItem())
217-227: Order the assertions to avoid race in failing-path testCall task.run() in a launched child so the flow can emit RUNNING/ERROR while you’re awaiting them; then join. This prevents timing races inside the turbine block.
Apply this diff:
task.status.test { assertEquals(TaskStatus.IDLE, awaitItem()) - assertFailsWith<RuntimeException> { task.run() } + val job = launch { assertFailsWith<RuntimeException> { task.run() } } assertEquals(TaskStatus.RUNNING, awaitItem()) assertEquals(TaskStatus.ERROR, awaitItem()) + job.join() cancelAndIgnoreRemainingEvents() }
136-152: Float progress assertions are good; keep using tolerancesUsing deltas for float comparisons is correct. Optionally, you can also assert that the final progress reaches exactly 1f if the implementation guarantees it.
src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/StringMatchingTaskTest.kt (2)
21-25: Comment does not match the scenarioThe comment says “Empty pattern case”, but patternSize = 1 is not empty. Either adjust the comment or add an explicit empty-pattern case if supported by StringMatchingTask.
Apply this diff:
- // Empty pattern case + // Very small pattern caseIf empty patterns are supported, consider also testing patternSize = 0.
10-17: Reduce flakiness by making randomness deterministic or asserting invariantsThe “matches < 100” check is a magic threshold and may flake depending on alphabet/seed. Prefer deterministic inputs or a seed to the generator, or assert non-flaky invariants (e.g., 0 <= matches <= textSize - patternSize + 1).
Apply this diff as a safer invariant:
- // Statistical expectation for random text - assertTrue(matches < 100) // Should not match too frequently + // Invariant for any inputs + assertTrue(matches <= 1000 - 3 + 1)If StringMatchingTask accepts a seed or Random instance, pass a fixed one in tests for repeatability.
Would you like me to draft a small change to StringMatchingTask to accept a Random (with default), so tests can inject a fixed seed?
src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/QuickSortTask.kt (2)
34-38: Progress update condition rarely triggers; report at fixed thresholds and finalize to 1.0processedElements % 1000 == 0 will almost never be true since processedElements grows by high - low. Use a moving threshold and ensure a final updateProgress(1f) after sorting.
Apply this diff:
private var swaps = 0L private var processedElements = 0 + private var nextReport = 1_000 override suspend fun execute(): Long { val array = IntArray(arraySize) { Random.nextInt() } swaps = 0L processedElements = 0 + nextReport = 1_000 quickSort(array, 0, arraySize - 1) + // Ensure progress completes + updateProgress(1f) return swaps } private suspend fun quickSort(arr: IntArray, low: Int, high: Int) { if (low < high) { val pi = partition(arr, low, high) processedElements += (high - low) - if (processedElements % 1000 == 0) { + if (processedElements >= nextReport) { yield() updateProgress(processedElements.toFloat() / (arraySize * 2)) + nextReport += 1_000 }
30-43: Consider an iterative quick sort to avoid deep recursion in worst-case inputsUsing the last element as pivot can lead to O(n) recursion depth on nearly-sorted arrays and possible StackOverflowError. An iterative variant (explicit stack) or a better pivot strategy (median-of-three) would harden this.
I can provide an iterative quick sort variant with the same swap/progress accounting if you’d like.
src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/SortingTask.kt (4)
16-21: Progress may never reach 1.0; finalize progress at completionWith updateProgress(i / arraySize), the last update will be close to 1 but not exactly 1. Emit a final updateProgress(1f) before returning.
Apply this diff:
for (i in 0 until arraySize - 1) { if (i % 100 == 0) { yield() updateProgress(i.toFloat() / arraySize) } for (j in 0 until arraySize - i - 1) {And add at the end of execute():
- return swaps + updateProgress(1f) + return swaps
16-31: Add early-exit optimization for bubble sort to avoid worst-case O(n^2) on nearly sorted arraysA standard bubble sort optimization breaks if no swaps occur in an outer pass.
Apply this diff:
- for (i in 0 until arraySize - 1) { + for (i in 0 until arraySize - 1) { + var swapped = false if (i % 100 == 0) { yield() updateProgress(i.toFloat() / arraySize) } for (j in 0 until arraySize - i - 1) { if (array[j] > array[j + 1]) { val temp = array[j] array[j] = array[j + 1] array[j + 1] = temp swaps++ + swapped = true } } + if (!swapped) break }
6-11: Consider accepting a dispatcher like other tasks for consistency and testabilityQuickSortTask accepts a dispatcher; this task doesn’t. Accepting a CoroutineDispatcher (defaulting to Dispatchers.Default) improves consistency and allows injecting a TestDispatcher in tests.
Proposed change (needs imports for CoroutineDispatcher and Dispatchers):
-import kotlinx.coroutines.yield +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.yield @@ -class SortingTask( - private val arraySize: Int = 50_000, -) : BaseTask<Int>( - name = "Array Sorting", - description = "Sorting $arraySize elements", -) { +class SortingTask( + private val arraySize: Int = 50_000, + dispatcher: CoroutineDispatcher = Dispatchers.Default, +) : BaseTask<Int>( + name = "Array Sorting", + description = "Sorting $arraySize elements", + dispatcher = dispatcher, +) {If you want, I can apply the same pattern to other tasks for uniformity.
6-13: Default size 50,000 is extremely heavy for bubble sortAt 50k elements, bubble sort performs ~1.25e9 comparisons and swaps in the worst case, which may time out CI. Consider lowering the default or switching to a more efficient algorithm if this is intended for demos.
src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/BinaryTreeTask.kt (2)
32-37: Emit a final progress update at completionYou scale build progress by nodeCount * 2 implying a second phase, but there’s no progress update during/after calculateHeight/sumNodes. Emit updateProgress(1f) at the end to reflect completion.
Apply this diff:
- return "Height: $height, Sum: $nodeSum" + updateProgress(1f) + return "Height: $height, Sum: $nodeSum"
50-53: Suspending helpers don’t suspend or yieldcalculateHeight/sumNodes are marked suspend but never suspend/yield. Either remove suspend or add occasional yields for cooperation on large trees.
I can add a simple node-visit counter to yield/update progress during traversal if you’d like.
src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/CompressionTask.kt (2)
29-41: Naive longest-match search is O(window × matchLen) per position (may be slow for 100k input).For dataSize=100k with a 4096 search window and max match length 255, worst-case work can reach hundreds of millions of comparisons. Consider a more efficient search:
- Rolling hash (Rabin–Karp) over the window to find candidate matches.
- Suffix array/tree or suffix automaton for near-linear matching.
- At minimum, early-exit when the best possible improvement cannot beat
maxLength.If you’d like, I can sketch a rolling-hash based refactor that keeps the code simple while cutting the average cost.
49-50: Compression ratio math likely inaccurate (constant 3.0 assumes token size).
compressed.size * 3.0 / dataSizeassumes 3 encoded bytes per token. Given(offset, length, nextChar), many LZ77 variants use at least 4 bytes (2 for offset, 1 for length, 1 for literal). If you intend to estimate encoded size, make the assumption explicit and easy to tweak.Optionally, make the token size configurable:
- val compressionRatio = compressed.size * 3.0 / dataSize + val bytesPerToken = 4.0 // adjust if your encoding differs + val compressionRatio = (compressed.size * bytesPerToken) / dataSizeIf you plan to mix literals and references, we can compute a more precise estimate per token.
src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/ImageProcessingTask.kt (1)
44-46: Avoid heavy allocations when computing average; also make formatting locale-stable.
processed.map { it.toList() }.flatten().average()allocates intermediate lists for the whole image. Compute the sum directly and divide by total pixels, and prefer a fixedLocalefor tests/logs.Apply this diff:
- val avgBrightness = processed.map { it.toList() }.flatten().average() - return "Avg brightness: %.2f".format(avgBrightness) + val total: Long = processed.sumOf { row -> row.sum().toLong() } + val avgBrightness = total.toDouble() / (width.toLong() * height.toLong()) + return String.format(java.util.Locale.US, "Avg brightness: %.2f", avgBrightness)Note: The explicit
Locale.USavoids decimal comma issues on some systems.src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/CryptographicTaskTest.kt (1)
17-26: Prefer parameterized test over a loop for better reporting.Using a loop collapses multiple cases into one test and obscures which count failed. Parameterized tests give per-input diagnostics.
Apply this diff:
- @Test - fun `should handle different message counts`() = runTest { - val counts = listOf(10, 50, 100) - - for (count in counts) { - val task = CryptographicTask(count) - val result = task.run() - assertEquals(count, result) - } - } + @org.junit.jupiter.params.ParameterizedTest + @org.junit.jupiter.params.provider.ValueSource(ints = [10, 50, 100]) + fun `should handle different message counts`(count: Int) = runTest { + val task = CryptographicTask(count) + val result = task.run() + assertEquals(count, result) + }You’ll need the JUnit params dependency and imports:
- org.junit.jupiter.params.ParameterizedTest
- org.junit.jupiter.params.provider.ValueSource
src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/GraphTreeTasksTest.kt (2)
20-24: Reduce flakiness in height assertion.
height < 100can fail if the tree degenerates (worst-case equals nodeCount). Unless the builder enforces balancing, prefer<= 100or assert a range.Apply this diff:
- assertTrue(height < 100) // Height should be logarithmic + assertTrue(height <= 100) // Bound to nodeCount to avoid rare degeneracy flakes
35-45: Be aware:runTestvirtual time won’t control tasks running onDispatchers.Default.Your tasks execute via
BaseTaskonDispatchers.Default, sorunTest(timeout = 10.seconds)won’t leverage virtual time and may rely on wall-clock performance. If you want deterministic, fast tests, allow injecting a dispatcher into the concrete tasks and pass aStandardTestDispatcherfrom tests.I can help thread a dispatcher parameter through the task constructors (defaulting to
Dispatchers.Default) and update tests accordingly.src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/BaseTask.kt (2)
34-37: Cancellation race:cancel()beforerun()setsjobcan be ignored.If
cancel()is called beforejobis captured inrun(), the task can still start and run to completion, despite status being set to CANCELLED. Track a pre-start cancellation request and respect it inrun().Apply this diff:
private var job: Job? = null + @Volatile private var cancelRequested: Boolean = false @@ override fun cancel() { - job?.cancel() - updateStatus(TaskStatus.CANCELLED) + cancelRequested = true + job?.cancel() + updateStatus(TaskStatus.CANCELLED) }
39-49: Honor pre-start cancellation inrun()to avoid starting cancelled tasks.Check
cancelRequestedbefore transitioning to RUNNING or executing.Apply this diff:
override suspend fun run(): R { return withContext(dispatcher) { - job = coroutineContext[Job] - updateStatus(TaskStatus.RUNNING) - updateProgress(0f) + job = coroutineContext[Job] + if (cancelRequested) { + // Reset the flag for potential future runs + cancelRequested = false + updateStatus(TaskStatus.CANCELLED) + throw CancellationException("Cancelled before start") + } + updateStatus(TaskStatus.RUNNING) + updateProgress(0f)src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/IntegrationTest.kt (3)
59-76: Use virtual-time helpers instead of delay for determinismReplace hardcoded delays with advanceTimeBy/advanceUntilIdle to remove timing flakiness and make the test deterministic under StandardTestDispatcher.
- // Start all tasks - viewModel.runAllTasks() - delay(100) + // Start all tasks + viewModel.runAllTasks() + advanceTimeBy(100) // Cancel all tasks viewModel.cancelAllTasks() - delay(100) + advanceUntilIdle()You’ll need imports (see separate comment).
3-12: Add missing test utilities imports for virtual-time controlIf you adopt advanceTimeBy/advanceUntilIdle as suggested, add these:
import kotlinx.coroutines.test.StandardTestDispatcher import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.resetMain import kotlinx.coroutines.test.runTest import kotlinx.coroutines.test.setMain
83-84: Duplicate lifecycle call: onCleared() is already called in @AfterEachtearDown() calls viewModel.onCleared(). Calling it again here is redundant and risks double-cleanup side effects.
- viewModel.onCleared()src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/StringMatchingTask.kt (2)
37-41: Ensure progress reaches 1.0 and guard progress division
- Progress never reaches 1f currently. Emit a final update at the end.
- For safety, handle empty text (defensive, even if preconditions added).
while (i < text.length) { - if (i % 10000 == 0) { + if (i % 10000 == 0) { yield() - updateProgress(i.toFloat() / text.length) + if (text.isNotEmpty()) { + updateProgress(i.toFloat() / text.length) + } } } - return matches + updateProgress(1f) + return matchesAlso applies to: 60-61
15-25: Micro-optimization: pre-size builders and reuse RandomLarge string builds benefit from pre-sizing and a local Random to avoid repeated global RNG overhead.
- val text = buildString { - repeat(textSize) { - append(('a'..'z').random()) - } - } + val rnd = kotlin.random.Random.Default + val text = StringBuilder(textSize).apply { + repeat(textSize) { + append(('a'..'z').random(rnd)) + } + }.toString() - val pattern = buildString { - repeat(patternSize) { - append(('a'..'z').random()) - } - } + val pattern = StringBuilder(patternSize).apply { + repeat(patternSize) { + append(('a'..'z').random(rnd)) + } + }.toString()src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/GraphAlgorithmsTask.kt (2)
42-47: Progress never reaches 1.0; finalize it after Dijkstra completesEmit a final progress update to 1f to signal completion to observers.
for (count in 0 until vertices - 1) { if (count % 50 == 0) { yield() updateProgress(count.toFloat() / vertices) } } - return dist + updateProgress(1f) + return distAlso applies to: 61-62
13-26: Algorithmic efficiency and representation mismatch (sparse graph in dense matrix)You build a sparse graph (E ≈ 10V) but use an adjacency matrix and O(V^2) Dijkstra with a linear min scan, which is suboptimal and memory-heavy beyond ~2–3k vertices. Consider an adjacency list + binary heap priority queue to achieve O(E log V) and reduce memory.
If you’d like, I can provide a version with:
- List<MutableList<Pair<Int, Int>>> adjacency
- PriorityQueue for Dijkstra
- Same yield/progress behavior mapped to processed vertices
Also applies to: 37-59
src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/SortingTasksTest.kt (2)
58-61: Remove unused variable or assert progress for QuickSort as wellquickSortTask is created but unused.
Option A: remove it.
- val mergeSortTask = MergeSortTask(arraySize = 5000, dispatcher = testDispatcher) - val quickSortTask = QuickSortTask(arraySize = 5000, dispatcher = testDispatcher) + val mergeSortTask = MergeSortTask(arraySize = 5000, dispatcher = testDispatcher)Option B: add a similar progress assertion for QuickSort to increase coverage.
34-45: MergeSort assertion may be brittle; consider looser bounds or verifying sortednessComparisons > 1000*8 might fail under some array distributions or implementation details. As an alternative, add a correctness check (e.g., verify array is sorted) or relax the comparison bound.
No change required if it’s stable in your environment; flagging as a potential flake.
src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/ParameterizedTaskTests.kt (4)
32-51: Harden cancellation test to avoid flakiness/hangs and improve readability
- Cooperative cancellation isn’t guaranteed if a task performs CPU-bound work without suspension points. Wrapping cancel+join in a timeout prevents the test from hanging indefinitely.
- Minor readability: simplify the final assertion with a set membership check.
Apply these diffs:
@@ - delay(50) - job.cancel() - job.join() + delay(50) + // Guard against uncooperative cancellation to avoid hanging tests + kotlinx.coroutines.withTimeout(2_000) { + job.cancel() + job.join() + } @@ - assertTrue( - task.status.value == TaskStatus.CANCELLED || - task.status.value == TaskStatus.COMPLETED, - ) + assertTrue( + task.status.value in setOf(TaskStatus.CANCELLED, TaskStatus.COMPLETED), + )
53-67: Make progress test deterministic and ensure cleanup of the runner job
- Keep a handle to the
task.run()job and cancel/join it to avoid leaks.- Assert that at least one emission was captured before accessing
first()to avoid accidental failures if collection doesn’t start in time (even though StateFlow should emit immediately).@@ - val job = launch { + val job = launch { task.progress.collect { progressUpdates.add(it) } } - launch { task.run() } + val runJob = launch { task.run() } delay(500) job.cancel() - assertTrue(progressUpdates.first() == 0f) + runJob.cancel() + runJob.join() + assertTrue(progressUpdates.isNotEmpty(), "No progress emissions captured") + assertEquals(0f, progressUpdates.first())
1-1: Narrow the suppression scope (nit)Prefer limiting suppressions to the smallest scope that needs them. File-level “SwallowedException” can be localized to the specific test.
-@file:Suppress("MagicNumber", "SwallowedException") +@file:Suppress("MagicNumber")And annotate the cancellable test locally:
@@ - @ParameterizedTest + @ParameterizedTest + @Suppress("SwallowedException") @MethodSource("taskProvider") fun `all tasks should be cancellable`(task: Task<*>) = runTest {
69-85: Minor Kotlin-idiomatic nit: prefer List over Java Stream in providerJUnit 5 supports
List<Arguments>sources. Returning a KotlinListavoids the JavaStreamimport and reads more idiomatically. Optional.- @JvmStatic - fun taskProvider(): Stream<Arguments> = Stream.of( + @JvmStatic + fun taskProvider(): List<Arguments> = listOf( Arguments.of(PrimeCalculationTask(100)), Arguments.of(MergeSortTask(100)), Arguments.of(QuickSortTask(100)), Arguments.of(MatrixMultiplicationTask(10)), Arguments.of(MandelbrotTask(10, 10)), Arguments.of(HashComputationTask(100)), Arguments.of(BinaryTreeTask(100)), Arguments.of(GraphAlgorithmsTask(10)), Arguments.of(StringMatchingTask(100, 5)), Arguments.of(ImageProcessingTask(10, 10)), Arguments.of(NeuralNetworkTask(10, 5, 3, 1)), Arguments.of(CompressionTask(100)), Arguments.of(CryptographicTask(10)), - ) + )If you prefer to keep returning a
Stream<Arguments>, consider:
- Building a Kotlin list and calling
.stream()at the end, or- Keeping as-is (the current use is correct).
api/Kotlin-Lab.api (6)
20114-20121: Hide implementation detail: make TreeNode non-public
BinaryTreeTask$TreeNodeis publicly exposed. Unless users are expected to manipulate the tree externally, mark itinternalorprivateto avoid locking this structure into your public ABI.Suggested change (BinaryTreeTask.kt):
-public class class BinaryTreeTask { - public class TreeNode(val value: Int) { ... } +class BinaryTreeTask { + private class TreeNode(val value: Int) { ... } }
20108-20112: Constructor consistency: align parameter ordering and defaults across tasksSome tasks accept only size parameters; others also accept
CoroutineDispatcher. Keep a consistent convention (e.g., domain params first, optionaldispatcher: CoroutineDispatcher = Dispatchers.Defaultlast) across all tasks to reduce friction and surprises.Also applies to: 20123-20127, 20129-20133, 20135-20139, 20141-20145, 20147-20151, 20158-20162, 20164-20168, 20170-20174, 20176-20180, 20182-20186, 20188-20192, 20194-20198, 20200-20204
20153-20156: Consider moving MainKt usage demo to a samples moduleKeeping runnable demos in a dedicated samples/examples module helps keep the core library lean and avoids accidental coupling to demonstration code.
20215-20230: TaskResult is already genericThe
TaskResult<T>declaration has been updated to use a type parameter forresult:• File: src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/TaskResult.kt
data class TaskResult<T>( val taskName: String, val result: T?, val executionTime: Long, val status: TaskStatus, )No further changes are needed to make
TaskResultgeneric.Optional refactoring:
- Consider replacing
executionTime: LongwithexecutionTime: Durationfromkotlin.timeto make the time units explicit.
20094-20106: Ensure structured concurrency with a dedicatedCoroutineScopeandSupervisorJob(optional refactor)I like how you’ve narrowed
progressandstatusdown to typedStateFlowviaasStateFlow()—that part is solid. A couple of optional hardening suggestions to avoid lifecycle leaks or unexpected cancelations:• Currently,
run()useswithContext(dispatcher)and pulls theJobfrom the caller’s context. That works, but it means each task doesn’t own its own scope or supervisor job. Consider:
- Defining
private val job = SupervisorJob() protected val scope = CoroutineScope(dispatcher + job)- Running your work inside that scope (e.g.
scope.async { … }orwithContext(scope.coroutineContext) { … }) so child coroutines don’t bubble up cancellation to unintended parents.• The status transitions (IDLE → RUNNING → COMPLETED / CANCELLED / ERROR) and progress clamping look correct on success, but note that on cancellation or error you only update the status. If you want a guaranteed final progress value (e.g.
1f), you could also callupdateProgress(1f)in those branches.Example refactor outline:
abstract class BaseTask<R>( override val name: String, override val description: String, private val dispatcher: CoroutineDispatcher = Dispatchers.Default, ) : Task<R> { - private var job: Job? = null + private val job = SupervisorJob() + protected val scope = CoroutineScope(dispatcher + job) private val _progress = MutableStateFlow(0f) @@ - override fun cancel() { - job?.cancel() - updateStatus(TaskStatus.CANCELLED) - } + override fun cancel() { + job.cancel() + updateStatus(TaskStatus.CANCELLED) + updateProgress(1f) // if you want a final progress bump + } override suspend fun run(): R { - return withContext(dispatcher) { + return withContext(scope.coroutineContext) { updateStatus(TaskStatus.RUNNING) updateProgress(0f) try { val result = execute() - updateStatus(TaskStatus.COMPLETED) - updateProgress(1f) + updateStatus(TaskStatus.COMPLETED) + updateProgress(1f) result } catch (e: CancellationException) { - updateStatus(TaskStatus.CANCELLED) + updateStatus(TaskStatus.CANCELLED) + updateProgress(1f) throw e } catch (e: Exception) { - updateStatus(TaskStatus.ERROR) + updateStatus(TaskStatus.ERROR) + updateProgress(1f) throw e } } }These changes are optional but will give each task a self-contained lifecycle and ensure consistent final progress updates.
20243-20256: TaskViewModel: introduce per-task Job tracking and cancellationThe
TaskViewModelalready:
- Uses a
SupervisorJob-backedCoroutineScope(private val scope = CoroutineScope(dispatcher + SupervisorJob()))- Cancels the scope in
onCleared()- Exposes results as
StateFlow<List<TaskResult<*>>>To strengthen per-task cancellation and simplify
cancelAllTasks(), consider:• At the top of the class, add:
private val taskJobs = ConcurrentHashMap<String, Job>()• In
runTask(task: Task<*>), capture and track each launched job:fun runTask(task: Task<*>) { - scope.launch { + // cancel any previous run of the same task + taskJobs.remove(task.name)?.cancel() + val job = scope.launch { val startTime = System.currentTimeMillis() try { … } catch (…) { … } } + taskJobs[task.name] = job }• Add individual cancellation:
fun cancelTask(name: String) = taskJobs.remove(name)?.cancel()• Replace
cancelAllTasks()with:fun cancelAllTasks() { - _tasks.value.forEach { it.cancel() } + taskJobs.values.forEach { it.cancel() } + taskJobs.clear() }Optional:
- Switch to a
MutableSharedFlow<TaskResult<*>>(or sealed events) to emit each result as it arrives, rather than accumulating lists in state.src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/TaskViewModel.kt (7)
5-15: Add imports required by proposed refactors.If you adopt the atomic updates and job-tracking/named coroutines suggestions, you’ll need these:
import kotlinx.coroutines.flow.update import kotlinx.coroutines.CoroutineName import kotlinx.coroutines.Job import java.util.concurrent.ConcurrentHashMap
31-37: Use a monotonic clock for timing and name the coroutine for easier debugging.System.currentTimeMillis can jump (NTP/timezone changes). Prefer a monotonic source. Also, naming jobs makes debugging and logs clearer.
Apply:
- scope.launch { - val startTime = System.currentTimeMillis() + scope.launch(CoroutineName("Task-${task.name}")) { + val startNs = System.nanoTime() @@ - val executionTime = System.currentTimeMillis() - startTime + val executionTime = (System.nanoTime() - startNs) / 1_000_000 @@ - val executionTime = System.currentTimeMillis() - startTime + val executionTime = (System.nanoTime() - startNs) / 1_000_000 @@ - val executionTime = System.currentTimeMillis() - startTime + val executionTime = (System.nanoTime() - startNs) / 1_000_000Also applies to: 45-46, 53-54
70-72: Ensure coroutine jobs are cancelled, not just theTaskabstraction.If you adopt job-tracking, cancel the jobs here to guarantee termination; keep calling
Task.cancel()if it does any additional cleanup.fun cancelAllTasks() { - _tasks.value.forEach { it.cancel() } + jobs.values.forEach { it.cancel() } + _tasks.value.forEach { it.cancel() } }
84-86: Consider invokingcancelAllTasks()before cancelling the scope.This ensures any
Task-level cleanup runs even if the scope is already cancelled.fun onCleared() { - scope.cancel() + cancelAllTasks() + scope.cancel() }
38-43: Confirm success status source to avoid stale values in results.On success you snapshot
status = task.status.value. IfTask.run()updates status after producing the result, this could capture RUNNING instead of COMPLETED/SUCCESS. Consider setting an explicit success status here, or ensureTask.run()updates status before returning.
64-68: Optionally return or await jobs fromrunAllTasks().Returning
List<Job>(or awaiting) can make tests and callers explicitly know when all tasks have been scheduled or completed.Example options:
- Return jobs:
fun runAllTasks(): List<Job> = _tasks.value.map { runTask(it) }- Or, if keeping Unit, collect jobs in the map and expose a helper
suspend fun joinAll()that awaits them.
44-52: Optional: don’t swallowCancellationExceptionsilently.After recording a CANCELLED result, consider rethrowing
CancellationExceptionso the Job completes in a cancelled state (useful for structured logging and diagnostics). With aSupervisorJob, sibling tasks won’t be affected.} catch (e: CancellationException) { val executionTime = System.currentTimeMillis() - startTime _results.value += TaskResult( taskName = task.name, result = null, executionTime = executionTime, status = TaskStatus.CANCELLED, ) + throw e
| <a href="https://app.fossa.com/projects/git%2Bgithub.com%2Fashtanko%2Falgorithms-in-depth?ref=badge_shield&issueType=license"><img alt="FOSSA Status" src="https://app.fossa.com/api/projects/git%2Bgithub.com%2Fashtanko%2Falgorithms-in-depth.svg?type=shield&issueType=license"/></a> | ||
| <a href="https://ktlint.github.io/"><img alt="CodeStyle" src="https://img.shields.io/badge/code%20style-%E2%9D%A4-FF4081.svg"/></a> | ||
| <a href="https://kotlinlang.org/"><img alt="Kotlin Version" src="https://img.shields.io/badge/kotlin-2.2.0-blue.svg"/></a> | ||
| <a href="https://kotlinlang.org/"><img alt="Kotlin Version" src="https://img.shields.io/badge/kotlin-2.2.10-blue.svg"/></a> |
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Avoid committing backup files; also verify Kotlin badge matches build toolchain
.bak files are typically transient and shouldn’t be versioned. Consider removing this file or moving it under docs with a proper name. While you’re here, confirm the 2.2.10 badge aligns with the Kotlin toolchain used by Gradle and libs.versions.toml.
Run this script to surface Kotlin versions used in the repo and the badge versions referenced in docs:
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Kotlin versions in build files (plugins/deps):"
rg -n -C1 -g '!**/build/**' -P '(org\.jetbrains\.kotlin|kotlin-stdlib|kotlinx\-coroutines|kotlin\("jvm"\)|kotlin\("android"\)|kotlin\("multiplatform"\))\W*[:= ]*\W*["'\'']?([0-9]+\.[0-9]+\.[0-9]+)?' || true
echo
echo "Kotlin version badges in docs (README, config/*):"
rg -n -C0 -P 'img\.shields\.io/badge/kotlin-([0-9]+\.[0-9]+\.[0-9]+)' README.md config/ || trueLength of output: 12494
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Locating [versions] section in gradle/libs.versions.toml:"
rg -n '^\[versions\]' -n gradle/libs.versions.toml -A5 || true
echo
echo "Extracting 'kotlin' version definition:"
rg -nP '^\s*kotlin\s*=\s*".*"' -n gradle/libs.versions.toml || trueLength of output: 705
Remove the backup file config/main.md.bak
- Delete config/main.md.bak – backup files shouldn’t be committed.
- No action needed on the Kotlin badge: it’s 2.2.10, which matches the
kotlin = "2.2.10"entry in gradle/libs.versions.toml.
🤖 Prompt for AI Agents
In config/main.md.bak around line 13, this is a backup file that should not be
committed; remove config/main.md.bak from the repository and the commit (delete
the file), and if already added to git, run git rm --cached or git rm to remove
it and update .gitignore to prevent future .bak files from being committed.
| private suspend fun sumNodes(node: TreeNode?): Long { | ||
| if (node == null) return 0 | ||
| return node.value + sumNodes(node.left) + sumNodes(node.right) | ||
| } |
There was a problem hiding this comment.
Compilation error: returning Int where Long is required
sumNodes is declared to return Long but returns 0 (Int) for the null case. Use 0L to match the return type.
Apply this diff:
- private suspend fun sumNodes(node: TreeNode?): Long {
- if (node == null) return 0
- return node.value + sumNodes(node.left) + sumNodes(node.right)
- }
+ private suspend fun sumNodes(node: TreeNode?): Long {
+ if (node == null) return 0L
+ return node.value + sumNodes(node.left) + sumNodes(node.right)
+ }🤖 Prompt for AI Agents
In src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/BinaryTreeTask.kt
around lines 55 to 58, the function sumNodes is declared to return Long but
returns the Int literal 0 for the null case; change the null branch to return 0L
so the literal is Long and all arithmetic uses Long (i.e., replace "return 0"
with "return 0L").
| val nextChar = if (i + maxLength < data.size) data[i + maxLength] else 0 | ||
| compressed.add(Triple(maxOffset, maxLength, nextChar)) | ||
| i += maxLength + 1 |
There was a problem hiding this comment.
Fix type mismatch for nextChar (compile-time error).
The else branch uses an Int literal (0) where a Byte is required, leading to a type mismatch when constructing Triple<Int, Int, Byte>.
Apply this diff:
- val nextChar = if (i + maxLength < data.size) data[i + maxLength] else 0
+ val nextChar = if (i + maxLength < data.size) data[i + maxLength] else 0.toByte()🤖 Prompt for AI Agents
In src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/CompressionTask.kt
around lines 44 to 46, the else branch assigns an Int literal (0) to nextChar
while data elements are Byte, causing a type mismatch when building Triple<Int,
Int, Byte>; change the literal to a Byte (e.g., 0.toByte()) so nextChar is a
Byte and the Triple type matches.
| override suspend fun execute(): Int { | ||
| var totalIterations = 0 | ||
| val totalPixels = width * height | ||
| var processedPixels = 0 | ||
|
|
||
| for (y in 0 until height) { | ||
| if (y % 10 == 0) yield() | ||
|
|
||
| for (x in 0 until width) { | ||
| val cx = (x - width / 2.0) * 4.0 / width | ||
| val cy = (y - height / 2.0) * 4.0 / height | ||
|
|
||
| totalIterations += calculateMandelbrot(cx, cy) | ||
| processedPixels++ | ||
|
|
||
| if (processedPixels % 1000 == 0) { | ||
| updateProgress(processedPixels.toFloat() / totalPixels) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return totalIterations | ||
| } |
There was a problem hiding this comment.
Integer overflow risk for large dimensions; also ensure progress reaches 1.0
With dimensions like 10_000×10_000 (as used in Main), summing per-pixel iterations up to 256 will overflow Int. Accumulate in Long and clamp on return (or change the task’s generic to Long if feasible). Also, emit a final progress update to 1.0.
- override suspend fun execute(): Int {
- var totalIterations = 0
+ override suspend fun execute(): Int {
+ var totalIterations = 0L
val totalPixels = width * height
var processedPixels = 0
for (y in 0 until height) {
if (y % 10 == 0) yield()
for (x in 0 until width) {
val cx = (x - width / 2.0) * 4.0 / width
val cy = (y - height / 2.0) * 4.0 / height
- totalIterations += calculateMandelbrot(cx, cy)
+ totalIterations += calculateMandelbrot(cx, cy).toLong()
processedPixels++
if (processedPixels % 1000 == 0) {
updateProgress(processedPixels.toFloat() / totalPixels)
}
}
}
- return totalIterations
+ // Ensure 100% progress at completion
+ updateProgress(1f)
+ // Clamp to Int to preserve signature; consider changing BaseTask<Int> -> BaseTask<Long> instead.
+ return totalIterations.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()
}If you’d like, I can follow up with a patch converting MandelbrotTask to BaseTask and wiring it through TaskResult/TaskViewModel.
🤖 Prompt for AI Agents
In src/main/kotlin/dev/shtanko/concurrency/coroutines/tasks/MandelbrotTask.kt
around lines 15–37, the per-pixel iteration sum can overflow Int for large
dimensions and the progress may never reach 1.0; change totalIterations to a
Long and accumulate into it, then at the end clamp to Int range with
totalIterations.coerceAtMost(Int.MAX_VALUE.toLong()).toInt() for the function
return (or switch the task generic to Long if you prefer), and emit a final
updateProgress(1.0f) after the loops so progress always reaches completion.
|
|
||
| import kotlinx.coroutines.async | ||
| import kotlinx.coroutines.delay | ||
| import kotlinx.coroutines.test.runTest |
There was a problem hiding this comment.
Don’t use kotlinx.coroutines.test.runTest for wall-clock perf/memory checks.
runTest uses a virtual time scheduler and is not reliable for real-time measurements or GC behavior. Use runBlocking (or plain JUnit) for these integration-style checks.
-import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.runBlocking
+import kotlin.system.measureTimeMillis🤖 Prompt for AI Agents
In src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/PerformanceTest.kt
around line 5, the test imports and uses kotlinx.coroutines.test.runTest which
uses virtual time and skews wall-clock and GC measurements; replace the import
and any runTest usage with kotlinx.coroutines.runBlocking (or a plain JUnit
test) so the test executes on real wall-clock time and observes actual GC
behavior, update the import statement accordingly and ensure any test coroutine
code runs inside runBlocking (or a standard JUnit test function) instead of
runTest.
| fun `tasks should complete within expected time`() = runTest { | ||
| val testCases = listOf( | ||
| PrimeCalculationTask(1000) to 1000L, | ||
| MergeSortTask(1000) to 500L, | ||
| HashComputationTask(1000) to 100L | ||
| ) | ||
|
|
||
| for ((task, maxTimeMs) in testCases) { | ||
| val startTime = System.currentTimeMillis() | ||
| task.run() | ||
| val duration = System.currentTimeMillis() - startTime | ||
|
|
||
| assertTrue( | ||
| duration < maxTimeMs, | ||
| "${task.name} took ${duration}ms, expected less than ${maxTimeMs}ms" | ||
| ) | ||
| } | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Make timing deterministic: use runBlocking and measureTimeMillis.
This avoids virtual time interference and is more idiomatic for integration timing.
- fun `tasks should complete within expected time`() = runTest {
+ fun `tasks should complete within expected time`() = runBlocking {
val testCases = listOf(
PrimeCalculationTask(1000) to 1000L,
MergeSortTask(1000) to 500L,
HashComputationTask(1000) to 100L
)
- for ((task, maxTimeMs) in testCases) {
- val startTime = System.currentTimeMillis()
- task.run()
- val duration = System.currentTimeMillis() - startTime
+ for ((task, maxTimeMs) in testCases) {
+ val duration = measureTimeMillis {
+ task.run()
+ }
assertTrue(
duration < maxTimeMs,
"${task.name} took ${duration}ms, expected less than ${maxTimeMs}ms"
)
}
}Optionally, wrap each run in withTimeout to avoid hangs and use more generous thresholds on CI runners.
| @Test | ||
| fun `cancelAllTasks cancels running tasks`() = testScope.runTest { | ||
| val task = CancellableTask() | ||
| viewModel.addTask(task) | ||
|
|
||
| viewModel.results.test { | ||
| assertEquals(emptyList(), awaitItem()) // initial | ||
|
|
||
| viewModel.runTask(task) | ||
| advanceTimeBy(150) // let it start a bit | ||
|
|
||
| viewModel.cancelAllTasks() | ||
| advanceUntilIdle() | ||
|
|
||
| val results = awaitItem() | ||
| assertEquals(1, results.size) | ||
| assertEquals("CancelTask", results[0].taskName) | ||
| assertEquals(TaskStatus.COMPLETED, results[0].status) | ||
|
|
||
| cancelAndIgnoreRemainingEvents() | ||
| } | ||
| } |
There was a problem hiding this comment.
Incorrect expectation: cancelled task should not be marked COMPLETED in results
This test cancels a running task but asserts COMPLETED. Results should reflect cancellation.
- assertEquals(TaskStatus.COMPLETED, results[0].status)
+ assertEquals(TaskStatus.CANCELLED, results[0].status)🤖 Prompt for AI Agents
In src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/TaskViewModelTest.kt
around lines 180 to 201, the test currently expects a cancelled task to be
marked COMPLETED; update the assertion to expect the cancelled status instead.
Replace the assertion asserting TaskStatus.COMPLETED with one asserting
TaskStatus.CANCELLED (or the enum value your code uses for cancelled tasks),
keeping the rest of the checks (size and taskName) the same so the test verifies
the task was cancelled rather than completed.
| @Test | ||
| fun `runAllTasks runs all tasks`() = testScope.runTest { | ||
| val t1 = SuccessfulTask() | ||
| val t2 = SuccessfulTask() | ||
| viewModel.addTask(t1) | ||
| viewModel.addTask(t2) | ||
|
|
||
| viewModel.results.test { | ||
| assertEquals(emptyList(), awaitItem()) // initial | ||
|
|
||
| viewModel.runAllTasks() | ||
| advanceUntilIdle() | ||
|
|
||
| val results = awaitItem() | ||
| assertEquals(1, results.size) | ||
| assertEquals(setOf("SuccessTask"), results.map { it.taskName }.toSet()) | ||
|
|
||
| cancelAndIgnoreRemainingEvents() | ||
| } | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
runAllTasks with two tasks should produce two results
Asserting a single result contradicts adding two tasks. Also using a Set hides the duplicate names. Assert the count and contents explicitly.
- val results = awaitItem()
- assertEquals(1, results.size)
- assertEquals(setOf("SuccessTask"), results.map { it.taskName }.toSet())
+ val results = awaitItem()
+ assertEquals(2, results.size)
+ assertEquals(listOf("SuccessTask", "SuccessTask"), results.map { it.taskName })🤖 Prompt for AI Agents
In src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/TaskViewModelTest.kt
around lines 235 to 254, the test adds two tasks but asserts only one result and
uses a Set which hides duplicates; change the assertions to expect two results
(assertEquals(2, results.size)) and verify the actual list contents (e.g.,
assert that results.map { it.taskName } equals the expected list of two names or
containsBoth entries) so the test validates both task results explicitly; keep
cancelAndIgnoreRemainingEvents() and maintain runAllTasks/advanceUntilIdle
sequencing.
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/IntegrationTest.kt (1)
51-56: Flaky assertion: asserting IDLE after runAllTasks doesn’t validate execution; wait for COMPLETED and close TurbineAsserting all are IDLE right after starting proves nothing. Wait until all four report COMPLETED and remember to cancel the Turbine collection.
- viewModel.tasks.test { - val tasks = awaitItem() - assertTrue(tasks.size >= 4) - val idleCount = tasks.count { it.status.value == TaskStatus.IDLE } - assertEquals(4, idleCount) - } + viewModel.tasks.test { + var completedCount = 0 + while (true) { + val tasks = awaitItem() + if (tasks.size >= 4) { + completedCount = tasks.count { it.status.value == TaskStatus.COMPLETED } + if (completedCount == 4) break + } + } + assertEquals(4, completedCount) + cancelAndIgnoreRemainingEvents() + }
🧹 Nitpick comments (4)
src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/IntegrationTest.kt (4)
34-38: Call onCleared before resetting Main to avoid lifecycle cleanup on a reset dispatcherIf onCleared touches Main-scoped resources, resetting Main first can cause brittle cleanup. Swap the order.
- Dispatchers.resetMain() - viewModel.onCleared() + viewModel.onCleared() + Dispatchers.resetMain()
40-42: Drop redundant @OptIn and replace @RepeatedTest(10) with @test (determinism > repetition)With a TestDispatcher and explicit scheduler advancement, repetition isn’t needed to fight flakiness; also the function-level @OptIn is redundant because the class is already opted-in.
- @OptIn(ExperimentalCoroutinesApi::class) - @RepeatedTest(10) + @Test fun `should run multiple different tasks concurrently`() = testScope.runTest(timeout = 30.seconds) {
8-13: Import missing test utilities used in the suggestionsYou’ll need advanceUntilIdle for deterministic progression.
import kotlinx.coroutines.test.StandardTestDispatcher import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.test.resetMain import kotlinx.coroutines.test.runTest import kotlinx.coroutines.test.setMain +import kotlinx.coroutines.test.advanceUntilIdle
83-84: Remove redundant onCleared; tearDown already clearsAvoid double-cleanup in tests.
- viewModel.onCleared()
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (1)
src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/IntegrationTest.kt(1 hunks)
🔇 Additional comments (1)
src/test/kotlin/dev/shtanko/concurrency/coroutines/tasks/IntegrationTest.kt (1)
28-33: Good test harness setup with injected TestDispatcherInjecting the StandardTestDispatcher and setting Dispatchers.Main in setup is solid and makes tests deterministic.
# Conflicts: # README.md # config/main.md.bak
…rationTest.kt Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
|
Note Unit test generation is an Early Access feature. Expect some limitations and changes as we gather feedback and continue to improve it. Generating unit tests... This may take up to 20 minutes. |
…gMatchingTask.kt Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
…rationTest.kt Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
|
Caution The CodeRabbit agent's plans did not produce any file changes. |
1 participant
Summary by CodeRabbit
New Features
Tests
✏️ Tip: You can customize this high-level summary in your review settings.