Revert "Load test results from cache concurrently"

This reverts commit 4001cd2.
badoo · Dec 6, 2024 · 555017d · 555017d
1 parent a384375
commit 555017d
Show file tree

Hide file tree

Showing 7 changed files with 29 additions and 47 deletions.
diff --git a/core/src/main/kotlin/com/malinskiy/marathon/cache/test/CacheResult.kt b/core/src/main/kotlin/com/malinskiy/marathon/cache/test/CacheResult.kt
@@ -6,12 +6,12 @@ import com.malinskiy.marathon.execution.TestShard
 
 sealed class CacheResult {
 
-    data class Hit(
+    class Hit(
         val pool: DevicePoolId,
         val testResult: TestResult
     ) : CacheResult()
 
-    data class Miss(
+    class Miss(
         val pool: DevicePoolId,
         val testShard: TestShard
     ) : CacheResult()

diff --git a/core/src/main/kotlin/com/malinskiy/marathon/cache/test/TestCacheLoader.kt b/core/src/main/kotlin/com/malinskiy/marathon/cache/test/TestCacheLoader.kt
@@ -13,16 +13,10 @@ import com.malinskiy.marathon.test.Test
 import com.malinskiy.marathon.test.toSimpleSafeTestName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Deferred
-import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.async
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ReceiveChannel
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.collect
-import kotlinx.coroutines.flow.flatMapMerge
-import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.flow.receiveAsFlow
-import kotlin.time.measureTimedValue
+import kotlin.system.measureTimeMillis
 
 class TestCacheLoader(
     private val configuration: Configuration,
@@ -42,17 +36,27 @@ class TestCacheLoader(
 
     fun initialize(scope: CoroutineScope) = with(scope) {
         cacheCheckCompleted = async {
-            testsToCheck.receiveAsFlow()
-                .concurrentMap(configuration.cache.concurrency) { test ->
-                    val (result, duration) = measureTimedValue { loadFromCache(test) }
-                    _results.send(result)
+            // TODO: check concurrently
+            for (test in testsToCheck) {
+                var result: CacheResult? = null
+                val timeMillis = measureTimeMillis {
+                    val cacheKey = cacheKeyFactory.getCacheKey(test.poolId, test.test)
 
-                    val hitOrMiss = if (result is Hit) "hit" else "miss"
-                    logger.debug {
-                        "Cache $hitOrMiss for ${test.test.toSimpleSafeTestName()}, took ${duration.inWholeMilliseconds} milliseconds"
+                    result = cache.load(cacheKey, test.test)?.let {
+                        Hit(test.poolId, it)
+                    } ?: Miss(test.poolId, TestShard(listOf(test.test)))
+
+                    _results.send(result!!)
+                }
+
+                logger.debug {
+                    val hitOrMiss = when (result!!) {
+                        is Hit -> "hit"
+                        is Miss -> "miss"
                     }
+                    "Cache $hitOrMiss for ${test.test.toSimpleSafeTestName()}, took $timeMillis milliseconds"
                 }
-                .collect()
+            }
         }
     }
 
@@ -63,7 +67,7 @@ class TestCacheLoader(
                 if (configuration.strictRunConfiguration.filter.matches(test)) {
                     testCacheBlackList.add(test)
                 } else {
-                    testsToCheck.send(TestToCheck(poolId, test))
+                    testsToCheck.send(TestToCheck(poolId, test, isStrictRun = false))
                 }
             }
 
@@ -83,21 +87,5 @@ class TestCacheLoader(
         logger.debug { "Cache loader is terminated" }
     }
 
-    private suspend fun loadFromCache(test: TestToCheck): CacheResult {
-        val cacheKey = cacheKeyFactory.getCacheKey(test.poolId, test.test)
-        return cache.load(cacheKey, test.test)?.let {
-            Hit(test.poolId, it)
-        } ?: Miss(test.poolId, TestShard(listOf(test.test)))
-    }
-
-    @OptIn(ExperimentalCoroutinesApi::class)
-    private fun <T, R> Flow<T>.concurrentMap(concurrency: Int, transform: suspend (T) -> R): Flow<R> =
-        flatMapMerge(concurrency) { value ->
-            flow { emit(transform(value)) }
-        }
-
-    private data class TestToCheck(
-        val poolId: DevicePoolId,
-        val test: Test
-    )
+    private class TestToCheck(val poolId: DevicePoolId, val test: Test, val isStrictRun: Boolean)
 }
diff --git a/core/src/main/kotlin/com/malinskiy/marathon/execution/CacheConfiguration.kt b/core/src/main/kotlin/com/malinskiy/marathon/execution/CacheConfiguration.kt
@@ -6,8 +6,7 @@ import com.malinskiy.marathon.cache.config.RemoteCacheConfiguration
 // TODO: support optional cache push
 data class CacheConfiguration(
     val local: LocalCacheConfiguration = LocalCacheConfiguration.Disabled,
-    val remote: RemoteCacheConfiguration = RemoteCacheConfiguration.Disabled,
-    val concurrency: Int = 8
+    val remote: RemoteCacheConfiguration = RemoteCacheConfiguration.Disabled
 ) {
 
     val isEnabled: Boolean

diff --git a/marathon-gradle-plugin/src/main/kotlin/com/malinskiy/marathon/CachePluginConfiguration.kt b/marathon-gradle-plugin/src/main/kotlin/com/malinskiy/marathon/CachePluginConfiguration.kt
@@ -42,11 +42,10 @@ internal fun LocalCacheExtension.initDefaults() {
     removeUnusedEntriesAfterDays.convention(7)
 }
 
-internal fun CachePluginConfiguration.toCacheConfiguration(concurrency: Int): CacheConfiguration =
+internal fun CachePluginConfiguration.toCacheConfiguration(): CacheConfiguration =
     CacheConfiguration(
         local = local.toConfig(),
-        remote = remote.toConfig(),
-        concurrency = concurrency
+        remote = remote.toConfig()
     )
 
 private fun LocalCacheExtension.toConfig(): LocalCacheConfiguration =

diff --git a/marathon-gradle-plugin/src/main/kotlin/com/malinskiy/marathon/ConfigurationFactory.kt b/marathon-gradle-plugin/src/main/kotlin/com/malinskiy/marathon/ConfigurationFactory.kt
@@ -9,11 +9,10 @@ import java.io.File
 internal fun createCommonConfiguration(
     extensionConfig: MarathonExtension,
     adbPath: File,
-    outputDir: File,
-    maxWorkers: Int
+    outputDir: File
 ): Configuration = Configuration(
     outputDir = outputDir,
-    cache = extensionConfig.cache.toCacheConfiguration(maxWorkers),
+    cache = extensionConfig.cache.toCacheConfiguration(),
     poolingStrategy = extensionConfig.poolingStrategy.toStrategy(),
     shardingStrategy = extensionConfig.shardingStrategy.toStrategy(),
     sortingStrategy = extensionConfig.sortingStrategy.toStrategy(),

diff --git a/marathon-gradle-plugin/src/main/kotlin/com/malinskiy/marathon/MarathonBuildService.kt b/marathon-gradle-plugin/src/main/kotlin/com/malinskiy/marathon/MarathonBuildService.kt
@@ -13,8 +13,7 @@ abstract class MarathonBuildService : BuildService<MarathonBuildService.Params>,
         val configuration = createCommonConfiguration(
             parameters.marathonConfig.get(),
             parameters.adbPath.get().asFile,
-            parameters.outputDir.get().asFile,
-            parameters.maxWorkers.get()
+            parameters.outputDir.get().asFile
         )
         WorkerContext(configuration)
     }
@@ -39,7 +38,6 @@ abstract class MarathonBuildService : BuildService<MarathonBuildService.Params>,
         val adbPath: DirectoryProperty
         val outputDir: DirectoryProperty
         val marathonConfig: Property<MarathonExtension>
-        val maxWorkers: Property<Int>
     }
 
     companion object {

diff --git a/marathon-gradle-plugin/src/main/kotlin/com/malinskiy/marathon/MarathonPlugin.kt b/marathon-gradle-plugin/src/main/kotlin/com/malinskiy/marathon/MarathonPlugin.kt
@@ -40,7 +40,6 @@ class MarathonPlugin : Plugin<Project> {
             parameters.adbPath.set(findAdbPath(projectDir))
             parameters.outputDir.set(layout.buildDirectory.dir("reports/marathon"))
             parameters.marathonConfig.set(marathonConfig)
-            parameters.maxWorkers.set(gradle.startParameter.maxWorkerCount)
         }
 
         tasks.register<MarathonWorkerRunTask>(WORKER_TASK_NAME)