client.go | 16 ++++++++-------- connection.go | 10 +++++----- connection_test.go | 18 +++++++++--------- file.go | 9 +++++---- piece.go | 10 +++++----- piecestate.go | 18 ++++++++++++++++++ t.go | 9 +++++++++ torrent.go | 124 +++++++++++++++++++++++------------------------------ diff --git a/client.go b/client.go index ee15ee0fb7164fe31f14d1f904ba3408e3009f72..1b8dbe0d11b318e9c9ab0d8450f3c1894724f8f8 100644 --- a/client.go +++ b/client.go @@ -279,18 +279,18 @@ } func (cl *Client) readRaisePiecePriorities(t *torrent, off, readaheadBytes int64) { index := int(off / int64(t.usualPieceSize())) - cl.raisePiecePriority(t, index, piecePriorityNow) + cl.raisePiecePriority(t, index, PiecePriorityNow) index++ if index >= t.numPieces() { return } - cl.raisePiecePriority(t, index, piecePriorityNext) + cl.raisePiecePriority(t, index, PiecePriorityNext) for range iter.N(readaheadPieces(readaheadBytes, t.Info.PieceLength)) { index++ if index >= t.numPieces() { break } - cl.raisePiecePriority(t, index, piecePriorityReadahead) + cl.raisePiecePriority(t, index, PiecePriorityReadahead) } } @@ -1981,7 +1981,7 @@ t.cl.mu.Lock() defer t.cl.mu.Unlock() pieceSize := int64(t.usualPieceSize()) for i := off / pieceSize; i*pieceSize < off+len; i++ { - t.cl.prioritizePiece(t.torrent, int(i), piecePriorityNormal) + t.cl.raisePiecePriority(t.torrent, int(i), PiecePriorityNormal) } } @@ -1999,12 +1999,12 @@ func (t Torrent) DownloadAll() { t.cl.mu.Lock() defer t.cl.mu.Unlock() for i := range iter.N(t.numPieces()) { - t.cl.raisePiecePriority(t.torrent, i, piecePriorityNormal) + t.cl.raisePiecePriority(t.torrent, i, PiecePriorityNormal) } // Nice to have the first and last pieces sooner for various interactive // purposes. - t.cl.raisePiecePriority(t.torrent, 0, piecePriorityReadahead) - t.cl.raisePiecePriority(t.torrent, t.numPieces()-1, piecePriorityReadahead) + t.cl.raisePiecePriority(t.torrent, 0, PiecePriorityReadahead) + t.cl.raisePiecePriority(t.torrent, t.numPieces()-1, PiecePriorityReadahead) } // Returns nil metainfo if it isn't in the cache. Checks that the retrieved @@ -2543,7 +2543,7 @@ func (me *Client) pieceChanged(t *torrent, piece int) { correct := t.pieceComplete(piece) p := t.Pieces[piece] if correct { - p.Priority = piecePriorityNone + p.Priority = PiecePriorityNone p.PendingChunkSpecs = nil for req := range t.urgent { if int(req.Index) == piece { diff --git a/connection.go b/connection.go index 1bf69b15f7b6c9b68fafe459778ffe75b6d1fa13..51034e9319927cf5cb33eb6dcdd81d94e22dd971 100644 --- a/connection.go +++ b/connection.go @@ -102,7 +102,7 @@ // Adjust piece position in the request order for this connection based on the // given piece priority. func (cn *connection) pendPiece(piece int, priority piecePriority) { - if priority == piecePriorityNone { + if priority == PiecePriorityNone { cn.pieceRequestOrder.DeletePiece(piece) return } @@ -117,13 +117,13 @@ // [ Readahead ] // [ Normal ] key := func() int { switch priority { - case piecePriorityNow: + case PiecePriorityNow: return -3*len(cn.piecePriorities) + 3*pp - case piecePriorityNext: + case PiecePriorityNext: return -2*len(cn.piecePriorities) + 2*pp - case piecePriorityReadahead: + case PiecePriorityReadahead: return -len(cn.piecePriorities) + pp - case piecePriorityNormal: + case PiecePriorityNormal: return pp default: panic(priority) diff --git a/connection_test.go b/connection_test.go index d93b7d05e2a627ed188ebdf18efdaf357e2df062..1ffb9aef3fb65005d1482f136e40f43867a161e4 100644 --- a/connection_test.go +++ b/connection_test.go @@ -77,22 +77,22 @@ pieceRequestOrder: pieceordering.New(), piecePriorities: []int{1, 4, 0, 3, 2}, } testRequestOrder(nil, c.pieceRequestOrder, t) - c.pendPiece(2, piecePriorityNone) + c.pendPiece(2, PiecePriorityNone) testRequestOrder(nil, c.pieceRequestOrder, t) - c.pendPiece(1, piecePriorityNormal) - c.pendPiece(2, piecePriorityNormal) + c.pendPiece(1, PiecePriorityNormal) + c.pendPiece(2, PiecePriorityNormal) testRequestOrder([]int{2, 1}, c.pieceRequestOrder, t) - c.pendPiece(0, piecePriorityNormal) + c.pendPiece(0, PiecePriorityNormal) testRequestOrder([]int{2, 0, 1}, c.pieceRequestOrder, t) - c.pendPiece(1, piecePriorityReadahead) + c.pendPiece(1, PiecePriorityReadahead) testRequestOrder([]int{1, 2, 0}, c.pieceRequestOrder, t) - c.pendPiece(4, piecePriorityNow) + c.pendPiece(4, PiecePriorityNow) // now(4), r(1), normal(0, 2) testRequestOrder([]int{4, 1, 2, 0}, c.pieceRequestOrder, t) - c.pendPiece(2, piecePriorityReadahead) + c.pendPiece(2, PiecePriorityReadahead) // N(4), R(1, 2), N(0) testRequestOrder([]int{4, 2, 1, 0}, c.pieceRequestOrder, t) - c.pendPiece(1, piecePriorityNow) + c.pendPiece(1, PiecePriorityNow) // now(4, 1), readahead(2), normal(0) // in the same order, the keys will be: -15+6, -15+12, -5, 1 // so we test that a very low priority (for this connection), "now" @@ -100,7 +100,7 @@ // piece has been placed after a readahead piece. testRequestOrder([]int{4, 2, 1, 0}, c.pieceRequestOrder, t) // Note this intentially sets to None a piece that's not in the order. for i := range iter.N(5) { - c.pendPiece(i, piecePriorityNone) + c.pendPiece(i, PiecePriorityNone) } testRequestOrder(nil, c.pieceRequestOrder, t) } diff --git a/file.go b/file.go index d663a2f64d659fe7a186e02d4e09d8c011e84905..2efbaeaaf8d0149e1e0b865492d1f1470a763a8f 100644 --- a/file.go +++ b/file.go @@ -29,11 +29,12 @@ return f.length } type FilePieceState struct { - Length int64 - State byte + Bytes int64 // Bytes within the piece that are part of this File. + PieceState } -func (f *File) Progress() (ret []FilePieceState) { +// Returns the state of pieces in this file. +func (f *File) State() (ret []FilePieceState) { pieceSize := int64(f.t.usualPieceSize()) off := f.offset % pieceSize remaining := f.length @@ -45,7 +46,7 @@ len1 := pieceSize - off if len1 > remaining { len1 = remaining } - ret = append(ret, FilePieceState{len1, f.t.pieceStatusChar(i)}) + ret = append(ret, FilePieceState{len1, f.t.pieceState(i)}) off = 0 remaining -= len1 } diff --git a/piece.go b/piece.go index c45e7a423bedbe99b318362b268b675ea0faaeb2..bae74094da2d1f618ab095507750450dd9683f23 100644 --- a/piece.go +++ b/piece.go @@ -12,11 +12,11 @@ type piecePriority byte const ( - piecePriorityNone piecePriority = iota - piecePriorityNormal - piecePriorityReadahead - piecePriorityNext - piecePriorityNow + PiecePriorityNone piecePriority = iota // Not wanted. + PiecePriorityNormal // Wanted. + PiecePriorityReadahead // May be required soon. + PiecePriorityNext // Succeeds a piece where a read occurred. + PiecePriorityNow // A read occurred in this piece. ) type piece struct { diff --git a/piecestate.go b/piecestate.go new file mode 100644 index 0000000000000000000000000000000000000000..358ffcf2d015a57b3dfc5cfe364a08a6b63e7560 --- /dev/null +++ b/piecestate.go @@ -0,0 +1,18 @@ +package torrent + +// The current state of a piece. +type PieceState struct { + Priority piecePriority + // The piece is available in its entirety. + Complete bool + // The piece is being hashed, or is queued for hash. + Checking bool + // Some of the piece has been obtained. + Partial bool +} + +// Represents a series of consecutive pieces with the same state. +type PieceStateRun struct { + PieceState + Length int // How many consecutive pieces have this state. +} diff --git a/t.go b/t.go index 70e6f8dd42affed6990582404a9b24693eaf8e5a..2f0ddbee230c11da7bc2fd3efe42d475272bdbdf 100644 --- a/t.go +++ b/t.go @@ -30,3 +30,12 @@ readahead: 5 * 1024 * 1024, } return } + +// Returns the state of pieces of the torrent. They are grouped into runs of +// same state. The sum of the state run lengths is the number of pieces +// in the torrent. +func (t *Torrent) PieceStateRuns() []PieceStateRun { + t.stateMu.Lock() + defer t.stateMu.Unlock() + return t.torrent.pieceStateRuns() +} diff --git a/torrent.go b/torrent.go index a764a662f83e5b1532b4f2dc509ae7842c6f3f86..2fa69be3b545610a64a4fbdd40292578e4049fcb 100644 --- a/torrent.go +++ b/torrent.go @@ -10,6 +10,7 @@ "sort" "sync" "time" + "github.com/anacrolix/missinggo" "github.com/bradfitz/iter" "github.com/anacrolix/torrent/bencode" @@ -289,38 +290,19 @@ } return "" } -func (t *torrent) pieceStatusChar(index int) byte { +func (t *torrent) pieceState(index int) (ret PieceState) { p := t.Pieces[index] - switch { - case t.pieceComplete(index): - return 'C' - case p.QueuedForHash: - return 'Q' - case p.Hashing: - return 'H' - case !p.EverHashed: - return '?' - case t.piecePartiallyDownloaded(index): - switch p.Priority { - case piecePriorityNone: - return 'F' // Forgotten - default: - return 'P' - } - default: - switch p.Priority { - case piecePriorityNone: - return 'z' - case piecePriorityNow: - return '!' - case piecePriorityReadahead: - return 'R' - case piecePriorityNext: - return 'N' - default: - return '.' - } + ret.Priority = p.Priority + if t.pieceComplete(index) { + ret.Complete = true + } + if p.QueuedForHash || p.Hashing { + ret.Checking = true + } + if t.piecePartiallyDownloaded(index) { + ret.Partial = true } + return } func (t *torrent) metadataPieceSize(piece int) int { @@ -346,45 +328,45 @@ ExtendedPayload: append(p, data...), } } -type PieceStatusCharSequence struct { - Char byte // The state of this sequence of pieces. - Count int // How many consecutive pieces have this state. -} - -// Returns the state of pieces of the torrent. They are grouped into runs of -// same state. The sum of the Counts of the sequences is the number of pieces -// in the torrent. See the function torrent.pieceStatusChar for the possible -// states. -func (t *torrent) PieceStatusCharSequences() []PieceStatusCharSequence { - t.stateMu.Lock() - defer t.stateMu.Unlock() - return t.pieceStatusCharSequences() +func (t *torrent) pieceStateRuns() (ret []PieceStateRun) { + rle := missinggo.NewRunLengthEncoder(func(el interface{}, count uint64) { + ret = append(ret, PieceStateRun{ + PieceState: el.(PieceState), + Length: int(count), + }) + }) + for index := range t.Pieces { + rle.Append(t.pieceState(index), 1) + } + rle.Flush() + return } -// Returns the length of sequences of identical piece status chars. -func (t *torrent) pieceStatusCharSequences() (ret []PieceStatusCharSequence) { - var ( - char byte - count int - ) - writeSequence := func() { - ret = append(ret, PieceStatusCharSequence{char, count}) +// Produces a small string representing a PieceStateRun. +func pieceStateRunStatusChars(psr PieceStateRun) (ret string) { + ret = fmt.Sprintf("%d", psr.Length) + ret += func() string { + switch psr.Priority { + case PiecePriorityNext: + return "N" + case PiecePriorityNormal: + return "." + case PiecePriorityReadahead: + return "R" + case PiecePriorityNow: + return "!" + default: + return "" + } + }() + if psr.Checking { + ret += "H" } - if len(t.Pieces) != 0 { - char = t.pieceStatusChar(0) + if psr.Partial { + ret += "P" } - for index := range t.Pieces { - char1 := t.pieceStatusChar(index) - if char1 == char { - count++ - } else { - writeSequence() - char = char1 - count = 1 - } - } - if count != 0 { - writeSequence() + if psr.Complete { + ret += "C" } return } @@ -411,9 +393,10 @@ return "?" } }()) if t.haveInfo() { - fmt.Fprint(w, "Pieces: ") - for _, seq := range t.pieceStatusCharSequences() { - fmt.Fprintf(w, "%d%c ", seq.Count, seq.Char) + fmt.Fprint(w, "Pieces:") + for _, psr := range t.pieceStateRuns() { + w.Write([]byte(" ")) + w.Write([]byte(pieceStateRunStatusChars(psr))) } fmt.Fprintln(w) } @@ -494,7 +477,8 @@ return } func (t *torrent) piecePartiallyDownloaded(index int) bool { - return t.pieceNumPendingBytes(index) != t.pieceLength(index) + pendingBytes := t.pieceNumPendingBytes(index) + return pendingBytes != 0 && pendingBytes != t.pieceLength(index) } func numChunksForPiece(chunkSize int, pieceSize int) int { @@ -713,7 +697,7 @@ } if p.Hashing { return false } - if p.Priority == piecePriorityNone { + if p.Priority == PiecePriorityNone { if !t.urgentChunkInPiece(index) { return false }