Fix TorrentDownloadController

torrent/api/src/commonMain/kotlin/api/pieces/TorrentDownloadController.kt

@@ -63,18 +63,20 @@ class TorrentDownloadController(
     private val footerSize: Long = headerSize,
     private val possibleFooterSize: Long = headerSize,
 ) : SynchronizedObject() {
-    private val totalSize: Long = pieces.sumOf { it.size }
+    private val totalPieceSize: Long = pieces.sumOf { it.size }
+    private val pieceOffsetStart = with(pieces) { pieces.first().dataStartOffset }
 
-    private val footerPieces = pieces.dropWhile { it.dataLastOffset < totalSize - footerSize }
-    private val possibleFooterRange = pieces.dropWhile { it.dataLastOffset < totalSize - possibleFooterSize }
+    private val footerPieces = pieces.dropWhile { it.dataLastOffset < pieceOffsetStart + totalPieceSize - footerSize }
+    private val possibleFooterRange = pieces
+        .dropWhile { it.dataLastOffset < pieceOffsetStart + totalPieceSize - possibleFooterSize }
         .let {
             if (it.isEmpty()) IntRange.EMPTY
             else it.first().pieceIndex..it.last().pieceIndex
         }
 
-    private val lastIndex = pieces.pieceIndexOfFirst { it.dataLastOffset >= totalSize - footerSize } - 1
+    private val lastIndex = pieces.last().pieceIndex
 
-    private var currentWindowStart = 0
+    private var currentWindowStart = pieces.initialPieceIndex
 
     // inclusive
     private var currentWindowEnd = (currentWindowStart + windowSize - 1).coerceAtMost(lastIndex)
@@ -88,7 +90,7 @@ class TorrentDownloadController(
     }
 
     fun onTorrentResumed() = synchronized(this) {
-        onSeek(0)
+        onSeek(pieces.initialPieceIndex)
     }
 
     fun onSeek(pieceIndex: Int) = synchronized(this) {
@@ -100,43 +102,57 @@ class TorrentDownloadController(
             return
         }
         downloadingPieces.clear()
-        currentWindowEnd = pieceIndex - 1
         fillWindow(pieceIndex)
         priorities.downloadOnly(downloadingPieces, possibleFooterRange)
     }
 
     /**
-     * 找接下来最近的还未完成的 piece, 如果没有, 返回 [startIndex]
+     * 找接下来最近的还未完成的 piece, 如果没有, 返回 null
      */
-    private fun findNextDownloadingPiece(startIndex: Int): Int {
-        for (index in (startIndex + 1)..lastIndex) {
+    private fun findNextDownloadingPiece(startIndex: Int): Int? {
+        for (index in startIndex..lastIndex) {
             if (with(pieces) { pieces.getByPieceIndex(index).state } != PieceState.FINISHED) {
                 return index
             }
         }
-        return startIndex
+        return null
     }
 
     fun onPieceDownloaded(pieceIndex: Int) = synchronized(this) {
         if (!downloadingPieces.remove(pieceIndex)) {
             return
         }
 
-        val newWindowEnd = findNextDownloadingPiece(currentWindowEnd)
-        if (newWindowEnd != currentWindowEnd) {
+        val newWindowEnd = findNextDownloadingPiece(currentWindowEnd + 1)
+        if (newWindowEnd != null && newWindowEnd != currentWindowEnd) {
             downloadingPieces.add(newWindowEnd)
             currentWindowEnd = newWindowEnd
         }
         priorities.downloadOnly(downloadingPieces, possibleFooterRange)
     }
 
+    /**
+     * seek 到 pieceIndex 不一定会重构从 pieceIndex 开始的 window
+     * 要从 pieceIndex 开始找接下来 window 大小个未完成的 piece 构成 window
+     */
     private fun fillWindow(pieceIndex: Int) {
-        val nextStartIndex = downloadingPieces.firstOrNull() ?: (currentWindowEnd + 1)
-        val nextEndIndex = (nextStartIndex + windowSize - 1).coerceAtMost(lastIndex)
-        downloadingPieces = (nextStartIndex..nextEndIndex).toMutableList()
-        currentWindowStart = downloadingPieces.firstOrNull() ?: -1
-        currentWindowEnd = downloadingPieces.lastOrNull() ?: -1
-        if (pieceIndex <= 1) {
+        // 如果 pieceIndex 以后的 piece 都完成了, 那就没有 piece 要填充到 window 了
+        currentWindowStart = findNextDownloadingPiece(pieceIndex) ?: return
+        currentWindowEnd = currentWindowStart
+
+        downloadingPieces.add(currentWindowStart)
+        for (i in 0..<windowSize - 1) {
+            val next = findNextDownloadingPiece(currentWindowEnd + 1)
+            if (next != null) {
+                downloadingPieces.add(next)
+                currentWindowEnd = next
+            } else {
+                // findNext 没找到, 说明所有的 piece 都下完了
+                break
+            }
+        }
+
+        if (pieceIndex <= pieces.initialPieceIndex + 1) {
             // 正在下载第 0-1 个 piece, 说明我们刚刚开始下载视频, 需要额外请求尾部元数据
             addFooterPieces()
         }

torrent/api/src/commonTest/kotlin/api/pieces/TorrentDownloadControllerTest.kt

@@ -0,0 +1,112 @@
+/*
+ * Copyright (C) 2024 OpenAni and contributors.
+ *
+ * 此源代码的使用受 GNU AFFERO GENERAL PUBLIC LICENSE version 3 许可证的约束, 可以在以下链接找到该许可证.
+ * Use of this source code is governed by the GNU AGPLv3 license, which can be found at the following link.
+ *
+ * https://github.com/open-ani/ani/blob/main/LICENSE
+ */
+
+package me.him188.ani.app.torrent.api.pieces
+
+import me.him188.ani.datasources.api.topic.FileSize.Companion.kiloBytes
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+internal class TorrentDownloadControllerTest {
+    internal val Int.kb: Long get() = kiloBytes.inBytes
+
+    @Test
+    fun `test sequence download and seek`() {
+        var currentDownloadingPieces: List<Int> = emptyList()
+        var currentPossibleFooterRange: IntRange = IntRange.EMPTY
+
+        val priorities = object : PiecePriorities {
+            override fun downloadOnly(pieceIndexes: List<Int>, possibleFooterRange: IntRange) {
+                currentDownloadingPieces = pieceIndexes
+                currentPossibleFooterRange = possibleFooterRange
+            }
+        }
+
+        val pieceList = PieceList.create(
+            totalSize = 1000.kb,
+            pieceSize = 1.kb,
+            initialDataOffset = 0,
+            initialPieceIndex = 0,
+        )
+
+        val controller = TorrentDownloadController(
+            pieces = pieceList,
+            priorities = priorities,
+            windowSize = 10,
+            headerSize = 5.kb,
+            footerSize = 3.kb,
+            possibleFooterSize = 12.kb,
+        )
+
+        fun finishPiece(index: Int) {
+            with(pieceList) { getByPieceIndex(index).state = PieceState.FINISHED }
+            controller.onPieceDownloaded(index)
+        }
+
+        controller.onTorrentResumed()
+        // resume 后立刻请求 windowSize 大小的 header 和 footer
+        assertEquals(13, currentDownloadingPieces.size)
+        assertEquals(listOf(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 997, 998, 999), currentDownloadingPieces)
+        assertEquals(988..999, currentPossibleFooterRange)
+
+        // resume window 内的 piece 会使 window 向后滑动
+        finishPiece(0)
+        assertEquals(listOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 997, 998, 999), currentDownloadingPieces.sorted())
+
+        finishPiece(1)
+        assertEquals(listOf(2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 997, 998, 999), currentDownloadingPieces.sorted())
+
+        finishPiece(5)
+        assertEquals(listOf(2, 3, 4, 6, 7, 8, 9, 10, 11, 12, 997, 998, 999), currentDownloadingPieces.sorted())
+
+        finishPiece(12)
+        assertEquals(listOf(2, 3, 4, 6, 7, 8, 9, 10, 11, 13, 997, 998, 999), currentDownloadingPieces.sorted())
+
+        // resume window 外的 piece 不会使 window 向后滑动
+        finishPiece(100)
+        assertEquals(listOf(2, 3, 4, 6, 7, 8, 9, 10, 11, 13, 997, 998, 999), currentDownloadingPieces.sorted())
+
+        // 不会重复请求已经完成的 piece
+        (10..99).forEach { finishPiece(it) }
+        assertEquals(listOf(2, 3, 4, 6, 7, 8, 9, 101, 102, 103, 997, 998, 999), currentDownloadingPieces.sorted())
+
+        // seek 到 200, seek 后面的地方不会请求 footer
+        controller.onSeek(200)
+        assertEquals(listOf(200, 201, 202, 203, 204, 205, 206, 207, 208, 209), currentDownloadingPieces.sorted())
+
+        (200..220).forEach { finishPiece(it) }
+        assertEquals(listOf(221, 222, 223, 224, 225, 226, 227, 228, 229, 230), currentDownloadingPieces.sorted())
+
+        // seek 到前面已经完成的部分, window 应该填充 50 后的前十个未完成的 piece
+        controller.onSeek(50)
+        assertEquals(listOf(101, 102, 103, 104, 105, 106, 107, 108, 109, 110), currentDownloadingPieces.sorted())
+
+        // 200 - 220 已经完成了
+        controller.onSeek(200)
+        assertEquals(listOf(221, 222, 223, 224, 225, 226, 227, 228, 229, 230), currentDownloadingPieces.sorted())
+
+        (101..109 step 2).forEach { finishPiece(it) }
+        controller.onSeek(50)
+        assertEquals(listOf(102, 104, 106, 108, 110, 111, 112, 113, 114, 115), currentDownloadingPieces.sorted())
+
+        // 完成很多 piece
+        (0..994).forEach { finishPiece(it) }
+        controller.onSeek(100)
+        assertEquals(listOf(995, 996, 997, 998, 999), currentDownloadingPieces.sorted())
+
+        // 测试边界 piece
+        (998..999).forEach { finishPiece(it) }
+        controller.onSeek(100)
+        assertEquals(listOf(995, 996, 997), currentDownloadingPieces.sorted())
+
+        finishPiece(996)
+        controller.onSeek(100)
+        assertEquals(listOf(995, 997), currentDownloadingPieces.sorted())
+    }
+}