From 656d9fce9b6485bdacdf2944a617d6d5cc8aa97d Mon Sep 17 00:00:00 2001 From: Joshua Tam <297250+joshuatam@users.noreply.github.com> Date: Thu, 16 Apr 2026 12:57:48 +0800 Subject: [PATCH 1/3] refactor: Improve GOGDownloadManager download efficiency 1. Adopt Flow queuing concept from JavaSteam 2. add DownloadSpeedConfig to be used later for other store enhancement 3. update kotlinx-coroutines-core version to match JavaSteam version using 4. Update NetworkUtils httpForParallelDownloads for timeout and http protocol config --- .../app/gamenative/service/SteamService.kt | 33 +- .../service/gog/GOGDownloadManager.kt | 385 ++++++++++-------- .../gamenative/utils/DownloadSpeedConfig.kt | 56 +++ .../java/app/gamenative/utils/NetworkUtils.kt | 3 + gradle/libs.versions.toml | 2 +- 5 files changed, 278 insertions(+), 201 deletions(-) create mode 100644 app/src/main/java/app/gamenative/utils/DownloadSpeedConfig.kt diff --git a/app/src/main/java/app/gamenative/service/SteamService.kt b/app/src/main/java/app/gamenative/service/SteamService.kt index e0cd1ec9b6..9a9cc4c9cb 100644 --- a/app/src/main/java/app/gamenative/service/SteamService.kt +++ b/app/src/main/java/app/gamenative/service/SteamService.kt @@ -173,6 +173,7 @@ import app.gamenative.statsgen.Achievement import app.gamenative.statsgen.StatType import app.gamenative.statsgen.StatsAchievementsGenerator import app.gamenative.statsgen.VdfParser +import app.gamenative.utils.DownloadSpeedConfig import java.nio.ByteBuffer import java.nio.ByteOrder @@ -1657,34 +1658,10 @@ class SteamService : Service(), IChallengeUrlChanged { return@launch } - // Some notes here: - // Write should always be 1 in mobile device, as normally it does not use a SSD for storage - // And to have maximum throughput, set downloadRatio = decompressRatio = 1.0 x CPU Cores - var downloadRatio = 0.0 - var decompressRatio = 0.0 - - when (PrefManager.downloadSpeed) { - 8 -> { - downloadRatio = 0.6 - decompressRatio = 0.2 - } - 16 -> { - downloadRatio = 1.2 - decompressRatio = 0.4 - } - 24 -> { - downloadRatio = 1.5 - decompressRatio = 0.5 - } - 32 -> { - downloadRatio = 2.4 - decompressRatio = 0.8 - } - } - - val cpuCores = Runtime.getRuntime().availableProcessors() - val maxDownloads = (cpuCores * downloadRatio).toInt().coerceAtLeast(1) - val maxDecompress = (cpuCores * decompressRatio).toInt().coerceAtLeast(1) + // Moved to DownloadSpeedConfig + val cpuCores = DownloadSpeedConfig.cpuCores + val maxDownloads = DownloadSpeedConfig.maxDownloads + val maxDecompress = DownloadSpeedConfig.maxDecompress Timber.i("CPU Cores: $cpuCores") Timber.i("maxDownloads: $maxDownloads") diff --git a/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt b/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt index cf3bcf0fa3..216926b268 100644 --- a/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt +++ b/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt @@ -1,9 +1,7 @@ package app.gamenative.service.gog import android.content.Context -import app.gamenative.PrefManager import app.gamenative.data.DownloadInfo -import app.gamenative.service.StreamingAssembly import app.gamenative.service.gog.api.DepotFile import app.gamenative.service.gog.api.FileChunk import app.gamenative.service.gog.api.GOGApiClient @@ -12,11 +10,13 @@ import app.gamenative.service.gog.api.GOGManifestParser import app.gamenative.service.gog.api.V1DepotFile import app.gamenative.enums.Marker import app.gamenative.utils.CdnRankingUtils +import app.gamenative.utils.DownloadSpeedConfig import app.gamenative.utils.MarkerUtils import app.gamenative.utils.Net import org.json.JSONArray import org.json.JSONObject import dagger.hilt.android.qualifiers.ApplicationContext +import kotlinx.coroutines.CoroutineScope import java.io.ByteArrayOutputStream import java.io.BufferedOutputStream import java.io.File @@ -29,15 +29,24 @@ import java.util.concurrent.CopyOnWriteArrayList import javax.inject.Inject import javax.inject.Singleton import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.async -import kotlinx.coroutines.awaitAll -import kotlinx.coroutines.channels.Channel -import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.Job +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.collect +import kotlinx.coroutines.flow.flatMapMerge +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.flowOn import kotlinx.coroutines.launch import kotlinx.coroutines.withContext import okhttp3.OkHttpClient import okhttp3.Request import timber.log.Timber +import java.io.IOException +import java.io.RandomAccessFile +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.ConcurrentHashMap.newKeySet +import java.util.concurrent.atomic.AtomicInteger /** * Custom exception for HTTP status errors with typed status code @@ -762,9 +771,8 @@ class GOGDownloadManager @Inject constructor( return "$normalizedPathBase/main.bin$querySuffix" } - - // assembles files as chunks arrive, deletes chunks once their last consumer is assembled + @OptIn(ExperimentalCoroutinesApi::class) private suspend fun downloadAndAssembleChunks( chunkUrlCandidates: Map>, chunkCacheDir: File, @@ -776,182 +784,209 @@ class GOGDownloadManager @Inject constructor( chunkToProductMap: Map, ): Result = withContext(Dispatchers.IO) { try { - val fileChunkIds = files.map { f -> f.chunks.map { it.compressedMd5 } } - val chunkLastFile = StreamingAssembly.buildChunkLastFileMap(fileChunkIds) - val parallelDownloads = PrefManager.downloadSpeed.coerceAtLeast(1) + val scope = CoroutineScope(Dispatchers.IO) + val parallelDownloads = DownloadSpeedConfig.maxDownloads + val parallelAssemble = DownloadSpeedConfig.maxDecompress val downloadHttpClient = Net.httpForParallelDownloads(parallelDownloads) - var currentChunkUrlCandidates = chunkUrlCandidates + val currentChunkUrlCandidates = ConcurrentHashMap(chunkUrlCandidates) val totalChunks = chunkHashes.size val totalFiles = files.size - val downloadedChunkIds = mutableSetOf() - var nextFileToAssemble = 0 + val chunkUsageCounts = ConcurrentHashMap() + val downloadedChunkIds = newKeySet() + val pendingChunks = AtomicInteger(chunkHashes.size) + + chunkHashes.forEach { chunkMd5 -> + chunkUsageCounts[chunkMd5] = AtomicInteger( + files.sumOf { file -> file.chunks.count { chunk -> chunk.compressedMd5 == chunkMd5 } } + ) + } + + val networkChunkFlow = MutableSharedFlow(extraBufferCapacity = Int.MAX_VALUE) + val assembleFlow = MutableSharedFlow>>(extraBufferCapacity = Int.MAX_VALUE) + + var assemblyFailure: Throwable? = null // assemble every file whose chunks have all arrived (or that has zero chunks) - suspend fun assembleReady(): Result { - while (nextFileToAssemble < totalFiles) { - val file = files[nextFileToAssemble] - if (!file.chunks.all { it.compressedMd5 in downloadedChunkIds }) break - if (!downloadInfo.isActive()) return Result.failure(Exception("Download cancelled")) - - val r = assembleFile(file, chunkCacheDir, installDir) - if (r.isFailure) return Result.failure( - r.exceptionOrNull() ?: Exception("Failed to assemble ${file.path}"), - ) + suspend fun assembleReady(chunkMd5: String): Result { + if (!downloadInfo.isActive()) { + return Result.failure(Exception("Download cancelled")) + } - for (chunk in file.chunks) { - if (chunkLastFile[chunk.compressedMd5] == nextFileToAssemble) { - File(chunkCacheDir, "${chunk.compressedMd5}.chunk").delete() - } - } - nextFileToAssemble++ + // 1. Find all files that contain this chunk + val matchedFiles = files.filter { file -> + file.chunks.any { chunk -> chunk.compressedMd5 == chunkMd5 } } - return Result.success(Unit) - } - Timber.tag("GOG").d("Streaming download+assembly: $totalChunks chunks, $totalFiles files") + // 2. For each file found, try to assemble if all chunks are ready + var assemblySuccessCount = 0 - downloadInfo.setProgress(0.0f) - downloadInfo.setActive(true) + matchedFiles.forEach { file -> + // Find the specific chunk that matches the current chunkMd5 + val chunk = file.chunks.first { chunk -> chunk.compressedMd5 == chunkMd5 } + val result = assembleFile(file, chunk, chunkCacheDir, installDir) + if (result.isSuccess) { + // 3. If assembly is successful and all chunks in downloadedChunkIds, increment file counter + assemblySuccessCount++ + } else { + Timber.tag("GOG").d(result.exceptionOrNull()?.message ?: "Failed to assemble ${file.path}") + } + } - chunkHashes.chunked(parallelDownloads).forEach { chunkBatch -> - if (!downloadInfo.isActive()) { - Timber.tag("GOG").w("Download cancelled by user") - return@withContext Result.failure(Exception("Download cancelled")) + // 4. Decrement usage count only when assembly is successful + if (assemblySuccessCount > 0) { + val usageCount = chunkUsageCounts[chunkMd5]?.addAndGet(-assemblySuccessCount) + if (usageCount != null && usageCount <= 0) { + val cacheFile = File(chunkCacheDir, "${chunkMd5}.chunk") + cacheFile.delete() + } } - // Download batch in parallel and process chunk completions as they arrive. - val completionChannel = Channel>>(chunkBatch.size) - val resultsByChunk = mutableMapOf>() - var assemblyFailure: Throwable? = null + return Result.success(Unit) + } - coroutineScope { - chunkBatch.forEach { chunkMd5 -> - launch { + val networkChunkJob: Job = scope.launch { + networkChunkFlow + .flatMapMerge(concurrency = parallelDownloads) { chunkMd5 -> + flow { val result = run { val urls = currentChunkUrlCandidates[chunkMd5] ?: return@run Result.failure( Exception("No URL candidates found for chunk $chunkMd5"), ) downloadChunkWithRetry(chunkMd5, urls, chunkCacheDir, downloadInfo, downloadHttpClient) } - completionChannel.send(chunkMd5 to result) + + // Always emit result to assembleFlow for processing (success or failure) + assembleFlow.tryEmit(chunkMd5 to result) + emit(Unit) } } + .flowOn(Dispatchers.IO) + .collect() + } + + val assembleJob: Job = scope.launch { + assembleFlow + .flatMapMerge>, Unit>(concurrency = parallelAssemble) { (chunkMd5, result) -> + flow { + if (result.isSuccess && assemblyFailure == null) { + // Successful download - add to completed set and try assembly + downloadedChunkIds.add(chunkMd5) + + val assembleResult = assembleReady(chunkMd5) + if (assembleResult.isFailure) { + assemblyFailure = assembleResult.exceptionOrNull() + ?: Exception("Failed to assemble ready files") + Timber.tag("GOG").d("Chunk $chunkMd5 assembleReady Failed: ${assemblyFailure.message}") + + // Requeue the chunk for retry + downloadedChunkIds.remove(chunkMd5) + networkChunkFlow.tryEmit(chunkMd5) + return@flow + } + + val progress = downloadedChunkIds.size.toFloat() / totalChunks + downloadInfo.setProgress(progress) + downloadInfo.updateStatusMessage( + "Downloading (${downloadedChunkIds.size}/$totalChunks chunks)", + ) + + // Decrement pending chunks counter + pendingChunks.decrementAndGet() + } else if (result.isFailure) { + // Failed download - handle retry logic + val exception = result.exceptionOrNull() + Timber.tag("GOG").d("Chunk $chunkMd5 download failed: ${exception?.message}") + + if (exception is HttpStatusException) { + Timber.tag("GOG").d("Chunk $chunkMd5 download failed: HttpError ${exception.statusCode}, ${exception.message}") + if (exception.statusCode in listOf(401, 403, 404, 500)) { + if (secureLinkContext != null) { + Timber.tag("GOG").w("Chunk $chunkMd5 urls expired, refreshing") + + val refreshResult = refreshSecureLinks(secureLinkContext, listOf(chunkMd5)) + if (refreshResult.isSuccess) { + currentChunkUrlCandidates[chunkMd5] = refreshResult.getOrThrow().getValue(chunkMd5) + networkChunkFlow.tryEmit(chunkMd5) + return@flow + } + } + } + } - repeat(chunkBatch.size) { - val (chunkMd5, result) = completionChannel.receive() - resultsByChunk[chunkMd5] = result - - // Optimistic streaming behavior inside the batch. - // If this batch later needs secure-link refresh, we still keep existing - // refresh semantics (refresh+retry whole batch), and set semantics - // prevent double-counting already completed chunks. - if (result.isSuccess && assemblyFailure == null) { - downloadedChunkIds.add(chunkMd5) - val assembleResult = assembleReady() - if (assembleResult.isFailure) { - assemblyFailure = assembleResult.exceptionOrNull() - ?: Exception("Failed to assemble ready files") - return@repeat + // For other failures, could add additional retry logic here + Timber.tag("GOG").e("Chunk $chunkMd5 failed permanently: ${exception?.message}") } - val progress = downloadedChunkIds.size.toFloat() / totalChunks - downloadInfo.setProgress(progress) - downloadInfo.updateStatusMessage( - "Downloading (${downloadedChunkIds.size}/$totalChunks chunks, $nextFileToAssemble/$totalFiles files)", - ) + emit(Unit) } } - } - completionChannel.close() - if (assemblyFailure != null) { - return@withContext Result.failure(assemblyFailure!!) - } - val results = chunkBatch.map { chunkMd5 -> - resultsByChunk[chunkMd5] ?: Result.failure( - Exception("Missing batch result for chunk $chunkMd5"), - ) - } + .flowOn(Dispatchers.IO) + .collect() + } - // Handle expired secure links (401/403/404) - val expiredLinkFailures = results.zip(chunkBatch).filter { (result, _) -> - val exception = result.exceptionOrNull() - exception is HttpStatusException && exception.statusCode in listOf(401, 403, 404) - } - val expiredChunkIds = expiredLinkFailures.map { (_, chunkMd5) -> chunkMd5 }.toSet() - val nonExpiredFailures = results.zip(chunkBatch).filter { (result, chunkMd5) -> - result.isFailure && chunkMd5 !in expiredChunkIds - } + Timber.tag("GOG").d("Streaming download+assembly: $totalChunks chunks, $totalFiles files") - if (expiredLinkFailures.isNotEmpty() && secureLinkContext != null) { - nonExpiredFailures.firstOrNull()?.first?.let { failedResult -> - return@withContext Result.failure( - failedResult.exceptionOrNull() - ?: Exception("Failed to download chunk with non-expired error"), - ) - } + downloadInfo.setProgress(0.0f) + downloadInfo.setActive(true) - Timber.tag("GOG").w("Detected ${expiredLinkFailures.size} expired secure link(s), refreshing...") + // Start downloads by launching a separate coroutine to emit chunks + scope.launch { + if (!downloadInfo.isActive()) { + Timber.tag("GOG").w("Download cancelled by user") + return@launch + } - expiredLinkFailures.forEach { (result, chunkMd5) -> - val productId = chunkToProductMap[chunkMd5] - Timber.tag("GOG").w("Chunk $chunkMd5 belongs to product $productId: ${result.exceptionOrNull()?.message}") - } + val chunksAdded = mutableListOf() - val refreshResult = refreshSecureLinks(secureLinkContext, chunkHashes) - if (refreshResult.isSuccess) { - currentChunkUrlCandidates = refreshResult.getOrThrow() - val failedChunkIds = expiredChunkIds.toList() - Timber.tag("GOG").i("Secure links refreshed, retrying ${failedChunkIds.size} failed chunk(s)") + files.forEach { file -> + Timber.tag("GOG").v("Pre-allocating ${file.path}") - val retryResults = failedChunkIds.map { chunkMd5 -> - async { - val urls = currentChunkUrlCandidates[chunkMd5] ?: return@async Result.failure( - Exception("No URL candidates found for chunk $chunkMd5 after refresh"), - ) - downloadChunkWithRetry(chunkMd5, urls, chunkCacheDir, downloadInfo, downloadHttpClient) - } - }.awaitAll() + // Allocating file before download + val outputFile = File(installDir, file.path) + outputFile.parentFile?.mkdirs() - retryResults.firstOrNull { it.isFailure }?.let { failedResult -> - return@withContext Result.failure( - failedResult.exceptionOrNull() ?: Exception("Failed to download chunk after link refresh"), - ) - } + val totalSize = file.chunks.sumOf { it.size } - // After a successful refresh+retry pass, ensure all chunks in this batch - // are reflected in assembly/progress state. - failedChunkIds.forEach { downloadedChunkIds.add(it) } - assembleReady().onFailure { return@withContext Result.failure(it) } + try { + // okio resize can OOM for large files on android. + RandomAccessFile(outputFile.path, "rw").use { + it.setLength(totalSize) + } - val progress = downloadedChunkIds.size.toFloat() / totalChunks - downloadInfo.setProgress(progress) - downloadInfo.updateStatusMessage( - "Downloading (${downloadedChunkIds.size}/$totalChunks chunks, $nextFileToAssemble/$totalFiles files)", - ) - } else { - return@withContext Result.failure( - refreshResult.exceptionOrNull() ?: Exception("Failed to refresh secure links"), - ) - } - } else { - results.firstOrNull { it.isFailure }?.let { failedResult -> - return@withContext Result.failure( - failedResult.exceptionOrNull() ?: Exception("Failed to download chunk"), - ) + file.chunks.forEach { chunk -> + if (!chunksAdded.contains(chunk.compressedMd5)) { + chunksAdded.add(chunk.compressedMd5) + networkChunkFlow.emit(chunk.compressedMd5) + Timber.tag("GOG").v("Emitted chunk ${chunk.compressedMd5} to download flow") + } + } + } catch (e: IOException) { + throw IOException("Failed to allocate file ${outputFile.path}: ${e.message}") } } + } + // Wait for all pending chunks to complete processing + while (pendingChunks.get() > 0) { + Timber.tag("GOG").d("Waiting for ${pendingChunks.get()} pending chunks to complete") + delay(1000) } - // assemble any remaining files whose chunks are all present (or have zero chunks) - assembleReady().onFailure { return@withContext Result.failure(it) } + // Cancel the download flow jobs since no more chunks will be added + networkChunkJob.cancel() + + // Cancel the assemble flow jobs since no more files will be added + assembleJob.cancel() + + // Remove the cache dir + chunkCacheDir.deleteRecursively() - if (nextFileToAssemble != totalFiles) { - throw Exception("Assembly incomplete: only $nextFileToAssemble of $totalFiles files assembled") + if (assemblyFailure != null) { + return@withContext Result.failure(assemblyFailure!!) } - Timber.tag("GOG").i("Streaming complete: $totalChunks chunks, $nextFileToAssemble files assembled") + Timber.tag("GOG").i("Streaming complete: $totalChunks chunks, ${files.size} files assembled") Result.success(Unit) } catch (e: Exception) { Timber.tag("GOG").e(e, "Failed to download and assemble") @@ -1368,6 +1403,7 @@ class GOGDownloadManager @Inject constructor( */ private suspend fun assembleFile( file: DepotFile, + chunk: FileChunk, chunkCacheDir: File, installDir: File, ): Result = withContext(Dispatchers.IO) { @@ -1375,54 +1411,59 @@ class GOGDownloadManager @Inject constructor( val outputFile = File(installDir, file.path) outputFile.parentFile?.mkdirs() - outputFile.outputStream().use { output -> - for (chunk in file.chunks) { - // Get compressed chunk file - val chunkFile = File(chunkCacheDir, "${chunk.compressedMd5}.chunk") + // Get compressed chunk file + val chunkFile = File(chunkCacheDir, "${chunk.compressedMd5}.chunk") - if (!chunkFile.exists()) { - return@withContext Result.failure( - Exception("Chunk file missing: ${chunk.compressedMd5}"), - ) - } + if (!chunkFile.exists()) { + return@withContext Result.failure( + Exception("Chunk file missing: ${chunk.compressedMd5}"), + ) + } - // Read compressed data - val compressedBytes = chunkFile.readBytes() + // Read compressed data + val compressedBytes = chunkFile.readBytes() - // Decompress chunk - val decompressedBytes = decompressChunk(compressedBytes, chunk) - if (decompressedBytes.isFailure) { - return@withContext Result.failure( - decompressedBytes.exceptionOrNull() - ?: Exception("Failed to decompress chunk ${chunk.compressedMd5}"), - ) - } + // Decompress chunk + val decompressedBytes = decompressChunk(compressedBytes, chunk) + if (decompressedBytes.isFailure) { + return@withContext Result.failure( + decompressedBytes.exceptionOrNull() + ?: Exception("Failed to decompress chunk ${chunk.compressedMd5}"), + ) + } - val data = decompressedBytes.getOrThrow() + val data = decompressedBytes.getOrThrow() - // Verify decompressed MD5 - val actualMd5 = calculateMd5(data) - if (actualMd5 != chunk.md5) { - return@withContext Result.failure( - Exception("Decompressed MD5 mismatch for chunk: expected ${chunk.md5}, got $actualMd5"), - ) - } + // Verify decompressed MD5 + val actualMd5 = calculateMd5(data) + if (actualMd5 != chunk.md5) { + return@withContext Result.failure( + Exception("Decompressed MD5 mismatch for chunk: expected ${chunk.md5}, got $actualMd5"), + ) + } - // Write to output file - output.write(data) - } + val chunkIndex = file.chunks.indexOfFirst { it.compressedMd5 == chunk.compressedMd5 } + val writeOffset = file.chunks.take(chunkIndex).sumOf { it.size } + + // Write decompressed chunk at specific file offset using RandomAccessFile + RandomAccessFile(outputFile.path, "rw").use { randomAccessFile -> + randomAccessFile.seek(writeOffset) + randomAccessFile.write(data) } // Verify final file hash if provided if (file.md5 != null) { val fileMd5 = calculateMd5File(outputFile) if (fileMd5 != file.md5) { - Timber.tag("GOG").w("File MD5 mismatch: ${file.path}, expected ${file.md5}, got $fileMd5") + // Timber.tag("GOG").w("File MD5 mismatch: ${file.path}, expected ${file.md5}, got $fileMd5") // Don't fail - some games have incorrect MD5 in manifest + // And as it is changed to use RandomAccessFile, it happens when not all chunks are completed download + } else { + // Move the log here for files finally assembled + Timber.tag("GOG").v("Assembled: ${file.path} (${outputFile.length()} bytes)") } } - Timber.tag("GOG").v("Assembled: ${file.path} (${outputFile.length()} bytes)") Result.success(outputFile) } catch (e: Exception) { Timber.tag("GOG").e(e, "Failed to assemble file ${file.path}") diff --git a/app/src/main/java/app/gamenative/utils/DownloadSpeedConfig.kt b/app/src/main/java/app/gamenative/utils/DownloadSpeedConfig.kt new file mode 100644 index 0000000000..cb94d832d0 --- /dev/null +++ b/app/src/main/java/app/gamenative/utils/DownloadSpeedConfig.kt @@ -0,0 +1,56 @@ +package app.gamenative.utils + +import app.gamenative.PrefManager + +class DownloadSpeedConfig { + + companion object { + val downloadRatio = when (PrefManager.downloadSpeed) { + 8 -> { + 0.6 + } + + 16 -> { + 1.2 + } + + 24 -> { + 1.5 + } + + 32 -> { + 2.4 + } + + else -> { + 0.6 + } + } + + val decompressRatio = when (PrefManager.downloadSpeed) { + 8 -> { + 0.2 + } + + 16 -> { + 0.4 + } + + 24 -> { + 0.5 + } + + 32 -> { + 0.8 + } + + else -> { + 0.2 + } + } + + val cpuCores = Runtime.getRuntime().availableProcessors() + val maxDownloads = (cpuCores * downloadRatio).toInt().coerceAtLeast(1) + val maxDecompress = (cpuCores * decompressRatio).toInt().coerceAtLeast(1) + } +} diff --git a/app/src/main/java/app/gamenative/utils/NetworkUtils.kt b/app/src/main/java/app/gamenative/utils/NetworkUtils.kt index 4e91e8f76d..856827f2b9 100644 --- a/app/src/main/java/app/gamenative/utils/NetworkUtils.kt +++ b/app/src/main/java/app/gamenative/utils/NetworkUtils.kt @@ -56,6 +56,9 @@ object Net { } return http.newBuilder() .dispatcher(dispatcher) + .readTimeout(5, TimeUnit.MINUTES) + .callTimeout(0, TimeUnit.MILLISECONDS) + .protocols(listOf(Protocol.HTTP_1_1)) .build() } } diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 36b15bfa84..dfcd0c3a43 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -5,7 +5,7 @@ apache-compress = "1.27.1" # https://mvnrepository.com/artifact/org.apache.commo apng = "3.0.2" # https://mvnrepository.com/artifact/com.github.penfeizhou.android.animation/apng composeBom = "2025.01.01" # https://mvnrepository.com/artifact/androidx.compose/compose-bom coreKtx = "1.15.0" # https://mvnrepository.com/artifact/androidx.core/core-ktx -coroutines = "1.10.1" # https://mvnrepository.com/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core +coroutines = "1.10.2" # https://mvnrepository.com/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core dagger-hilt = "2.55" # https://mvnrepository.com/artifact/com.google.dagger/hilt-android dataStore = "1.1.2" # https://mvnrepository.com/artifact/androidx.datastore/datastore-preferences espressoCore = "3.6.1" # https://mvnrepository.com/artifact/androidx.test.espresso/espresso-core From 7f14056cae31df73dda5fcd37c22ea2acdd6bc39 Mon Sep 17 00:00:00 2001 From: JT <297250+joshuatam@users.noreply.github.com> Date: Thu, 23 Apr 2026 20:19:02 +0800 Subject: [PATCH 2/3] ai comments --- .../app/gamenative/service/SteamService.kt | 7 +-- .../service/gog/GOGDownloadManager.kt | 27 ++++++----- .../gamenative/utils/DownloadSpeedConfig.kt | 45 ++++++------------- 3 files changed, 33 insertions(+), 46 deletions(-) diff --git a/app/src/main/java/app/gamenative/service/SteamService.kt b/app/src/main/java/app/gamenative/service/SteamService.kt index 9a9cc4c9cb..23dec5c356 100644 --- a/app/src/main/java/app/gamenative/service/SteamService.kt +++ b/app/src/main/java/app/gamenative/service/SteamService.kt @@ -1659,9 +1659,10 @@ class SteamService : Service(), IChallengeUrlChanged { } // Moved to DownloadSpeedConfig - val cpuCores = DownloadSpeedConfig.cpuCores - val maxDownloads = DownloadSpeedConfig.maxDownloads - val maxDecompress = DownloadSpeedConfig.maxDecompress + val speedConfig = DownloadSpeedConfig() + val cpuCores = speedConfig.cpuCores + val maxDownloads = speedConfig.maxDownloads + val maxDecompress = speedConfig.maxDecompress Timber.i("CPU Cores: $cpuCores") Timber.i("maxDownloads: $maxDownloads") diff --git a/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt b/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt index 216926b268..100586f699 100644 --- a/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt +++ b/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt @@ -785,8 +785,9 @@ class GOGDownloadManager @Inject constructor( ): Result = withContext(Dispatchers.IO) { try { val scope = CoroutineScope(Dispatchers.IO) - val parallelDownloads = DownloadSpeedConfig.maxDownloads - val parallelAssemble = DownloadSpeedConfig.maxDecompress + val speedConfig = DownloadSpeedConfig() + val parallelDownloads = speedConfig.maxDownloads + val parallelAssemble = speedConfig.maxDecompress val downloadHttpClient = Net.httpForParallelDownloads(parallelDownloads) val currentChunkUrlCandidates = ConcurrentHashMap(chunkUrlCandidates) @@ -822,15 +823,17 @@ class GOGDownloadManager @Inject constructor( var assemblySuccessCount = 0 matchedFiles.forEach { file -> - // Find the specific chunk that matches the current chunkMd5 - val chunk = file.chunks.first { chunk -> chunk.compressedMd5 == chunkMd5 } - val result = assembleFile(file, chunk, chunkCacheDir, installDir) - if (result.isSuccess) { - // 3. If assembly is successful and all chunks in downloadedChunkIds, increment file counter - assemblySuccessCount++ - } else { - Timber.tag("GOG").d(result.exceptionOrNull()?.message ?: "Failed to assemble ${file.path}") - } + file.chunks.withIndex() + .filter { (_, chunk) -> chunk.compressedMd5 == chunkMd5 } + .forEach { (chunkIndex, chunk) -> + val result = assembleFile(file, chunk, chunkIndex, chunkCacheDir, installDir) + if (result.isSuccess) { + // 3. If assembly is successful and all chunks in downloadedChunkIds, increment file counter + assemblySuccessCount++ + } else { + Timber.tag("GOG").d(result.exceptionOrNull()?.message ?: "Failed to assemble ${file.path}") + } + } } // 4. Decrement usage count only when assembly is successful @@ -1404,6 +1407,7 @@ class GOGDownloadManager @Inject constructor( private suspend fun assembleFile( file: DepotFile, chunk: FileChunk, + chunkIndex: Int, chunkCacheDir: File, installDir: File, ): Result = withContext(Dispatchers.IO) { @@ -1442,7 +1446,6 @@ class GOGDownloadManager @Inject constructor( ) } - val chunkIndex = file.chunks.indexOfFirst { it.compressedMd5 == chunk.compressedMd5 } val writeOffset = file.chunks.take(chunkIndex).sumOf { it.size } // Write decompressed chunk at specific file offset using RandomAccessFile diff --git a/app/src/main/java/app/gamenative/utils/DownloadSpeedConfig.kt b/app/src/main/java/app/gamenative/utils/DownloadSpeedConfig.kt index cb94d832d0..9ced7fa76f 100644 --- a/app/src/main/java/app/gamenative/utils/DownloadSpeedConfig.kt +++ b/app/src/main/java/app/gamenative/utils/DownloadSpeedConfig.kt @@ -3,54 +3,37 @@ package app.gamenative.utils import app.gamenative.PrefManager class DownloadSpeedConfig { + private data class Ratios(val download: Double, val decompress: Double) - companion object { - val downloadRatio = when (PrefManager.downloadSpeed) { + private val ratios: Ratios + get() = when (PrefManager.downloadSpeed) { 8 -> { - 0.6 + Ratios(download = 0.6, decompress = 0.2) } 16 -> { - 1.2 + Ratios(download = 1.2, decompress = 0.4) } 24 -> { - 1.5 + Ratios(download = 1.5, decompress = 0.5) } 32 -> { - 2.4 + Ratios(download = 2.4, decompress = 0.8) } else -> { - 0.6 + Ratios(download = 0.6, decompress = 0.2) } } - val decompressRatio = when (PrefManager.downloadSpeed) { - 8 -> { - 0.2 - } - - 16 -> { - 0.4 - } - - 24 -> { - 0.5 - } - - 32 -> { - 0.8 - } + val cpuCores: Int + get() = Runtime.getRuntime().availableProcessors() - else -> { - 0.2 - } - } + val maxDownloads: Int + get() = (cpuCores * ratios.download).toInt().coerceAtLeast(1) - val cpuCores = Runtime.getRuntime().availableProcessors() - val maxDownloads = (cpuCores * downloadRatio).toInt().coerceAtLeast(1) - val maxDecompress = (cpuCores * decompressRatio).toInt().coerceAtLeast(1) - } + val maxDecompress: Int + get() = (cpuCores * ratios.decompress).toInt().coerceAtLeast(1) } From 5f805047290a502943d25133389c905970fc9a1c Mon Sep 17 00:00:00 2001 From: JT <297250+joshuatam@users.noreply.github.com> Date: Thu, 23 Apr 2026 21:16:30 +0800 Subject: [PATCH 3/3] add retry logic when same pendingChunks appear 10 times in a row --- .../service/gog/GOGDownloadManager.kt | 31 +++++++++++++++++-- 1 file changed, 29 insertions(+), 2 deletions(-) diff --git a/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt b/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt index 100586f699..05c8f108fd 100644 --- a/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt +++ b/app/src/main/java/app/gamenative/service/gog/GOGDownloadManager.kt @@ -971,9 +971,36 @@ class GOGDownloadManager @Inject constructor( } // Wait for all pending chunks to complete processing - while (pendingChunks.get() > 0) { - Timber.tag("GOG").d("Waiting for ${pendingChunks.get()} pending chunks to complete") + var lastPendingChunks = pendingChunks.get() + var currentPendingChunks = lastPendingChunks + var samePendingChunksAttempts = 0 + while (currentPendingChunks > 0) { + Timber.tag("GOG").d("Waiting for $currentPendingChunks pending chunks to complete") + + if (currentPendingChunks == lastPendingChunks) { + samePendingChunksAttempts++ + } else { + lastPendingChunks = currentPendingChunks + samePendingChunksAttempts = 0 + } + + if (samePendingChunksAttempts >= 10) { + val missingChunks = chunkHashes.filterNot { downloadedChunkIds.contains(it) } + if (missingChunks.isNotEmpty()) { + Timber.tag("GOG").w( + "Pending chunks stuck at $currentPendingChunks for $samePendingChunksAttempts checks; " + + "re-emitting ${missingChunks.size} missing chunk(s) for retry", + ) + missingChunks.forEach { networkChunkFlow.tryEmit(it) } + } + + samePendingChunksAttempts = 0 + } + + // Wait for 1 second to recheck delay(1000) + + currentPendingChunks = pendingChunks.get() } // Cancel the download flow jobs since no more chunks will be added