Implement parallel page loading and preloading in reader (#2986)

app/src/main/java/eu/kanade/presentation/more/settings/screen/SettingsReaderScreen.kt

@@ -59,6 +59,11 @@ object SettingsReaderScreen : SearchableSettings {
                 preference = readerPref.pageTransitions,
                 title = stringResource(MR.strings.pref_page_transitions),
             ),
+            Preference.PreferenceItem.SwitchPreference(
+                preference = readerPref.loadAheadOnlyOverWifi,
+                title = stringResource(MR.strings.pref_load_ahead_only_over_wifi),
+                subtitle = stringResource(MR.strings.pref_load_ahead_only_over_wifi_summary),
+            ),
             getDisplayGroup(readerPreferences = readerPref),
             getEInkGroup(readerPreferences = readerPref),
             getReadingGroup(readerPreferences = readerPref),

app/src/main/java/eu/kanade/tachiyomi/ui/reader/ReaderViewModel.kt

@@ -39,6 +39,7 @@ import eu.kanade.tachiyomi.source.model.Page
 import eu.kanade.tachiyomi.source.online.HttpSource
 import eu.kanade.tachiyomi.ui.reader.loader.ChapterLoader
 import eu.kanade.tachiyomi.ui.reader.loader.DownloadPageLoader
+import eu.kanade.tachiyomi.ui.reader.loader.HttpPageLoader
 import eu.kanade.tachiyomi.ui.reader.model.InsertPage
 import eu.kanade.tachiyomi.ui.reader.model.ReaderChapter
 import eu.kanade.tachiyomi.ui.reader.model.ReaderPage
@@ -53,6 +54,7 @@ import eu.kanade.tachiyomi.util.editCover
 import eu.kanade.tachiyomi.util.lang.byteSize
 import eu.kanade.tachiyomi.util.storage.DiskUtil
 import eu.kanade.tachiyomi.util.storage.cacheImageDir
+import eu.kanade.tachiyomi.util.system.activeNetworkState
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -63,6 +65,7 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
@@ -349,6 +352,7 @@ class ReaderViewModel(
                     downloadProvider,
                     chapterCache,
                     appScope,
+                    ::canLoadAhead,
                     manga,
                     source,
                 )
@@ -380,6 +384,7 @@ class ReaderViewModel(
             chapterList.getOrNull(chapterPos - 1),
             chapterList.getOrNull(chapterPos + 1),
         )
+        preloadNextChapterAfter(newChapters)
 
         withContext(Dispatchers.Main) {
             val toDownload = cancelQueuedDownloads(newChapters.currChapter)
@@ -445,6 +450,34 @@ class ReaderViewModel(
         }
     }
 
+    private var nextChapterPreloadJob: Job? = null
+
+    /**
+     * Preloads the next chapter of [chapters] once the current one has loaded, so it doesn't take bandwidth from the
+     * pages being read. Skipped when loading ahead isn't allowed on this network.
+     */
+    private fun preloadNextChapterAfter(chapters: ViewerChapters) {
+        val nextChapter = chapters.nextChapter ?: return
+        nextChapterPreloadJob?.cancel()
+        nextChapterPreloadJob = viewModelScope.launch(Dispatchers.IO) {
+            if (!canLoadAhead()) return@launch
+            val current = chapters.currChapter
+            // Only online pages load in the background; other loaders load a page once it's shown
+            if (current.pageLoader is HttpPageLoader) {
+                current.pages.orEmpty().forEach { page ->
+                    page.statusFlow.first { it == Page.State.Ready || it is Page.State.Error }
+                }
+            }
+            preload(nextChapter)
+        }
+    }
+
+    /**
+     * Whether online chapters may load past the pages near the one being read.
+     */
+    private fun canLoadAhead(): Boolean =
+        !readerPreferences.loadAheadOnlyOverWifi.get() || context.activeNetworkState().isWifi
+
     /**
      * Called when the viewers decide it's a good time to preload a [chapter] and improve the UX so
      * that the user doesn't have to wait too long to continue reading.

app/src/main/java/eu/kanade/tachiyomi/ui/reader/loader/ChapterLoader.kt

@@ -30,6 +30,7 @@ class ChapterLoader(
     private val downloadProvider: DownloadProvider,
     private val chapterCache: ChapterCache,
     private val appScope: CoroutineScope,
+    private val canLoadAhead: () -> Boolean,
     private val manga: Manga,
     private val source: Source,
 ) {
@@ -105,7 +106,7 @@ class ChapterLoader(
                     is Format.Epub -> EpubPageLoader(format.file.epubReader(context))
                 }
             }
-            source is HttpSource -> HttpPageLoader(chapter, source, chapterCache, appScope)
+            source is HttpSource -> HttpPageLoader(chapter, source, chapterCache, appScope, canLoadAhead)
             source is StubSource -> error(context.stringResource(MR.strings.source_not_installed, source.toString()))
             else -> error(context.stringResource(MR.strings.loader_not_implemented_error))
         }

app/src/main/java/eu/kanade/tachiyomi/ui/reader/loader/HttpPageLoader.kt

@@ -8,22 +8,15 @@ import eu.kanade.tachiyomi.ui.reader.model.ReaderChapter
 import eu.kanade.tachiyomi.ui.reader.model.ReaderPage
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
-import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.flow.filter
-import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.joinAll
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.runInterruptible
-import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlinx.coroutines.withContext
-import java.util.concurrent.PriorityBlockingQueue
-import kotlin.concurrent.atomics.AtomicInt
-import kotlin.concurrent.atomics.ExperimentalAtomicApi
-import kotlin.concurrent.atomics.incrementAndFetch
-import kotlin.math.min
+import java.util.concurrent.ConcurrentHashMap
 
 /**
  * Loader used to load chapters from an online source.
@@ -33,46 +26,13 @@ internal class HttpPageLoader(
     private val source: HttpSource,
     private val chapterCache: ChapterCache,
     private val appScope: CoroutineScope,
+    private val canLoadAhead: () -> Boolean,
 ) : PageLoader() {
 
     private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
-
-    /**
-     * A queue used to manage requests one by one while allowing priorities.
-     */
-    private val queue = PriorityBlockingQueue<PriorityPage>()
-
-    private val preloadSize = 4
-
-    // How many pending loadPage calls want each page index, counting their preloads. Pages are downloaded one
-    // at a time, so once nothing wants the page in flight its download is cancelled instead of holding up the
-    // pages being read. Guards inFlight too.
-    private val wantedPages = HashMap<Int, Int>()
-    private var inFlight: Pair<Int, Job>? = null
-
-    init {
-        scope.launch(Dispatchers.IO) {
-            flow {
-                while (true) {
-                    emit(runInterruptible { queue.take() })
-                }
-            }
-                .filter { it.page.status == Page.State.Queue }
-                .collect {
-                    coroutineScope {
-                        val load = launch {
-                            internalLoadPage(
-                                page = it.page,
-                                force = it.priority == PriorityPage.RETRY,
-                            )
-                        }
-                        synchronized(wantedPages) { inFlight = it.page.index to load }
-                        load.join()
-                        synchronized(wantedPages) { inFlight = null }
-                    }
-                }
-        }
-    }
+    private val chunkProcessLock = Any()
+    private var chunkProcessJob: Job? = null
+    private val downloadJobs = ConcurrentHashMap<Int, Job>()
 
     override var isLocal: Boolean = false
 
@@ -91,21 +51,24 @@ internal class HttpPageLoader(
         }
         return pages.mapIndexed { index, page ->
             // Don't trust sources and use our own indexing
-            ReaderPage(index, page.url, page.imageUrl)
+            ReaderPage(index, page.url, page.imageUrl).also {
+                // Loading starts below, before ChapterLoader gets to set this, and chapter is a lateinit
+                it.chapter = chapter
+            }
         }
+            .also {
+                it.loadByRollingChunked(0, 1, 1)
+            }
     }
 
     /**
      * Loads a page through the queue. Handles re-enqueueing pages if they were evicted from the cache.
      */
-    override suspend fun loadPage(page: ReaderPage) {
+    override suspend fun loadPage(page: ReaderPage) = withContext(Dispatchers.IO) {
         val imageUrl = page.imageUrl
 
-        // Check if the image has been deleted. Only this touches the disk; the queueing below
-        // stays on the caller's thread, so requests queue in the order they were made.
-        if (page.status == Page.State.Ready && imageUrl != null &&
-            !withContext(Dispatchers.IO) { chapterCache.isImageInCache(imageUrl) }
-        ) {
+        // Check if the image has been deleted
+        if (page.status == Page.State.Ready && imageUrl != null && !chapterCache.isImageInCache(imageUrl)) {
             page.status = Page.State.Queue
         }
 
@@ -114,39 +77,16 @@ internal class HttpPageLoader(
             page.status = Page.State.Queue
         }
 
-        val nextPages = nextPages(page, preloadSize)
-        val wanted = listOf(page) + nextPages
-        synchronized(wantedPages) {
-            wanted.forEach { wantedPages.merge(it.index, 1, Int::plus) }
-        }
+        // Start the page being read before anything else. Planning the chunks walks the whole chapter and takes
+        // chunkProcessLock, and the chain only reaches this page once it has, so don't make it wait on either.
+        launchLoadOnce(page)
 
-        val queuedPages = mutableListOf<PriorityPage>()
-        if (page.status == Page.State.Queue) {
-            queuedPages += PriorityPage(page, PriorityPage.DEFAULT).also { queue.offer(it) }
-        }
-        queuedPages += nextPages.mapNotNull {
-            if (it.status == Page.State.Queue) {
-                PriorityPage(it, PriorityPage.ADJACENT).apply { queue.offer(this) }
-            } else {
-                null
-            }
-        }
-
-        suspendCancellableCoroutine<Nothing> { continuation ->
-            continuation.invokeOnCancellation {
-                queuedPages.forEach {
-                    if (it.page.status == Page.State.Queue) {
-                        queue.remove(it)
-                    }
-                }
-                synchronized(wantedPages) {
-                    wanted.forEach {
-                        wantedPages.computeIfPresent(it.index) { _, count -> (count - 1).takeIf { it > 0 } }
-                    }
-                    // A cancelled load ends in Error, which the next loadPage of that page queues again
-                    inFlight?.let { (index, load) -> if (index !in wantedPages) load.cancel() }
-                }
-            }
+        val pages = page.chapter.pages.orEmpty()
+        if (page.index !in pages.indices) return@withContext
+        pages.loadByRollingChunked(page.index, 3, 5) { chunks ->
+            val continueLoading = chunks.take(2).flatten().map { it.index }.toSet()
+            downloadJobs.keys.filter { it !in continueLoading }
+                .forEach { downloadJobs.remove(it)?.cancel() }
         }
     }
 
@@ -157,13 +97,16 @@ internal class HttpPageLoader(
         if (page.status is Page.State.Error) {
             page.status = Page.State.Queue
         }
-        queue.offer(PriorityPage(page, PriorityPage.RETRY))
+
+        val pageIndex = page.index
+        downloadJobs.remove(pageIndex)?.cancel()
+        scope.launch { internalLoadPage(page, force = true) }.trackAsDownloadJob(pageIndex)
     }
 
     override fun recycle() {
         super.recycle()
         scope.cancel()
-        queue.clear()
+        downloadJobs.clear()
 
         // Cache current page list progress for online chapters to allow a faster reopen
         chapter.pages?.let { pages ->
@@ -182,14 +125,95 @@ internal class HttpPageLoader(
     }
 
     /**
-     * Returns the [amount] of pages after [currentPage], which are preloaded with a lower priority.
+     * Keeps [this] in [downloadJobs] under [pageIndex] until it completes. The entry is only removed if it still
+     * points at this job, so a newer job for the same page isn't evicted by an older one finishing.
      */
-    private fun nextPages(currentPage: ReaderPage, amount: Int): List<ReaderPage> {
-        val pageIndex = currentPage.index
-        val pages = currentPage.chapter.pages ?: return emptyList()
-        if (pageIndex == pages.lastIndex) return emptyList()
+    private fun Job.trackAsDownloadJob(pageIndex: Int) = also { job ->
+        downloadJobs[pageIndex] = job
+        job.invokeOnCompletion { downloadJobs.remove(pageIndex, job) }
+    }
 
-        return pages.subList(pageIndex + 1, min(pageIndex + 1 + amount, pages.size))
+    /**
+     * Starts loading [page] unless it's already loaded or a load is already in flight for it. Claiming the slot and
+     * starting the job has to be atomic, otherwise two chunk chains can both launch the same page.
+     */
+    private fun launchLoadOnce(page: ReaderPage): Job? {
+        val pageIndex = page.index
+        // A page another chain is already loading still counts towards this chunk, otherwise the chain moves on
+        // and starts the next chunk while it's in flight
+        downloadJobs[pageIndex]?.takeIf { it.isActive }?.let { return it }
+        if (page.status !is Page.State.Queue && page.status !is Page.State.Error) return null
+
+        val job = scope.launch(start = CoroutineStart.LAZY) { internalLoadPage(page) }
+        val running = downloadJobs.putIfAbsent(pageIndex, job)
+        if (running != null) {
+            job.cancel()
+            // Hand back the in flight job so the chunk still waits for this page and the chunk size keeps bounding
+            // how many downloads run at once
+            return running
+        }
+        job.invokeOnCompletion { downloadJobs.remove(pageIndex, job) }
+        job.start()
+        return job
+    }
+
+    private fun List<ReaderPage>.loadByRollingChunked(
+        index: Int,
+        chunkStartSize: Int,
+        chunkEndSize: Int,
+        onNewChunks: (List<List<ReaderPage>>) -> Unit = {},
+    ) {
+        val items = this
+        // When loading ahead isn't allowed, only the pages around the one being read load
+        val window = if (canLoadAhead()) items.indices else (index - 1)..(index + LIMITED_PAGES_AHEAD)
+        val chunks = buildList {
+            items.getOrNull(index)?.let(::add)
+            items.getOrNull(index - 1)?.let(::add)
+            items.getOrNull(index + 1)?.let(::add)
+            // `items.lastIndex`, not the list being built, whose lastIndex is at most 2 here
+            if (index < items.lastIndex - 1) {
+                items.subList(index + 2, items.size).let(::addAll)
+            }
+            if (index > 1) {
+                items.subList(0, index - 1).reversed().let(::addAll)
+            }
+        }
+            .filter { it.index in window }
+            .rollingChunked(chunkStartSize, chunkEndSize)
+            .also(onNewChunks)
+
+        // Every visible page holder calls loadPage, so this runs concurrently
+        synchronized(chunkProcessLock) {
+            chunkProcessJob?.cancel()
+            chunkProcessJob = scope.launch {
+                for (chunk in chunks) {
+                    val jobs = chunk.mapNotNull { page -> launchLoadOnce(page) }
+                    jobs.joinAll()
+                }
+            }
+        }
+    }
+
+    /**
+     * Splits the list into chunks with rolling sizes between a starting size and an ending size.
+     *
+     * The chunking process starts with the given `startSize` and increases by 1 until it reaches
+     * the `endSize`, at which point it continues with the `endSize` size for the remaining items.
+     */
+    private fun <T> List<T>.rollingChunked(startSize: Int, endSize: Int): List<List<T>> {
+        // A start size of 0 never advances the index
+        require(startSize >= 1) { "startSize must be at least 1, was $startSize" }
+        val thisSize = this.size
+        val result = ArrayList<List<T>>()
+        var chunkSize = startSize
+        var index = 0
+        while (index < thisSize) {
+            val localChunkSize = chunkSize.coerceAtMost(thisSize - index)
+            result.add(List(localChunkSize) { this[it + index] })
+            index += chunkSize
+            if (chunkSize < endSize) chunkSize += 1
+        }
+        return result
     }
 
     /**
@@ -198,7 +222,7 @@ internal class HttpPageLoader(
      *
      * @param page the page whose source image has to be downloaded.
      */
-    private suspend fun internalLoadPage(page: ReaderPage, force: Boolean) {
+    private suspend fun internalLoadPage(page: ReaderPage, force: Boolean = false) {
         try {
             if (page.imageUrl.isNullOrEmpty()) {
                 page.status = Page.State.LoadPage
@@ -224,34 +248,15 @@ internal class HttpPageLoader(
             page.status = Page.State.Ready
         } catch (e: Throwable) {
             page.downloadStream = null
-            page.status = Page.State.Error(e)
             if (e is CancellationException) {
+                // Chunk chains get cancelled on every page turn, so a cancelled page isn't an error. Put it back in
+                // the queue, otherwise it's stuck in a state no later pass will pick up.
+                page.status = Page.State.Queue
                 throw e
             }
+            page.status = Page.State.Error(e)
         }
     }
 }
 
-/**
- * Data class used to keep ordering of pages in order to maintain priority.
- */
-@OptIn(ExperimentalAtomicApi::class)
-private class PriorityPage(
-    val page: ReaderPage,
-    val priority: Int,
-) : Comparable<PriorityPage> {
-    companion object {
-        private val idGenerator = AtomicInt(0)
-
-        const val RETRY = 2
-        const val DEFAULT = 1
-        const val ADJACENT = 0
-    }
-
-    private val identifier = idGenerator.incrementAndFetch()
-
-    override fun compareTo(other: PriorityPage): Int {
-        val p = other.priority.compareTo(priority)
-        return if (p != 0) p else identifier.compareTo(other.identifier)
-    }
-}
+private const val LIMITED_PAGES_AHEAD = 4

app/src/main/java/eu/kanade/tachiyomi/ui/reader/setting/ReaderPreferences.kt

@@ -22,6 +22,11 @@ class ReaderPreferences(
 
     val pageTransitions: Preference<Boolean> = preferenceStore.getBoolean("pref_enable_transitions_key", true)
 
+    val loadAheadOnlyOverWifi: Preference<Boolean> = preferenceStore.getBoolean(
+        "pref_reader_load_ahead_only_over_wifi",
+        true,
+    )
+
     val flashOnPageChange: Preference<Boolean> = preferenceStore.getBoolean("pref_reader_flash", false)
 
     val flashDurationMillis: Preference<Int> = preferenceStore.getInt("pref_reader_flash_duration", MILLI_CONVERSION)

i18n/src/commonMain/moko-resources/base/strings.xml

@@ -409,6 +409,8 @@
     <string name="pref_double_tap_zoom">Double tap to zoom</string>
     <string name="pref_cutout_short">Show content in cutout area</string>
     <string name="pref_page_transitions">Animate page transitions</string>
+    <string name="pref_load_ahead_only_over_wifi">Load ahead only on Wi-Fi</string>
+    <string name="pref_load_ahead_only_over_wifi_summary">Saves mobile data by loading only the pages near the one you\'re reading</string>
     <string name="pref_flash_page">Flash on page change</string>
     <string name="pref_flash_page_summ">Reduces ghosting on e-ink displays</string>
     <string name="pref_flash_duration">Flash duration</string>