piece.go

   1 package torrent
   2
   3 import (
   4         "fmt"
   5         "sync"
   6
   7         "github.com/anacrolix/chansync"
   8         g "github.com/anacrolix/generics"
   9         "github.com/anacrolix/missinggo/v2/bitmap"
  10
  11         "github.com/anacrolix/torrent/metainfo"
  12         pp "github.com/anacrolix/torrent/peer_protocol"
  13         "github.com/anacrolix/torrent/storage"
  14 )
  15
  16 type Piece struct {
  17         // The completed piece SHA1 hash, from the metainfo "pieces" field. Nil if the info is not V1
  18         // compatible.
  19         hash   *metainfo.Hash
  20         hashV2 g.Option[[32]byte]
  21         t      *Torrent
  22         index  pieceIndex
  23         files  []*File
  24
  25         readerCond chansync.BroadcastCond
  26
  27         numVerifies         int64
  28         hashing             bool
  29         marking             bool
  30         storageCompletionOk bool
  31
  32         publicPieceState PieceState
  33         priority         PiecePriority
  34         // Availability adjustment for this piece relative to len(Torrent.connsWithAllPieces). This is
  35         // incremented for any piece a peer has when a peer has a piece, Torrent.haveInfo is true, and
  36         // the Peer isn't recorded in Torrent.connsWithAllPieces.
  37         relativeAvailability int
  38
  39         // This can be locked when the Client lock is taken, but probably not vice versa.
  40         pendingWritesMutex sync.Mutex
  41         pendingWrites      int
  42         noPendingWrites    sync.Cond
  43
  44         // Connections that have written data to this piece since its last check.
  45         // This can include connections that have closed.
  46         dirtiers map[*Peer]struct{}
  47 }
  48
  49 func (p *Piece) String() string {
  50         return fmt.Sprintf("%s/%d", p.t.canonicalShortInfohash().HexString(), p.index)
  51 }
  52
  53 func (p *Piece) Info() metainfo.Piece {
  54         return p.t.info.Piece(p.index)
  55 }
  56
  57 func (p *Piece) Storage() storage.Piece {
  58         var pieceHash g.Option[[]byte]
  59         if p.hash != nil {
  60                 pieceHash.Set(p.hash.Bytes())
  61         } else if p.hashV2.Ok {
  62                 pieceHash.Set(p.hashV2.Value[:])
  63         }
  64         return p.t.storage.PieceWithHash(p.Info(), pieceHash)
  65 }
  66
  67 func (p *Piece) Flush() {
  68         if p.t.storage.Flush != nil {
  69                 _ = p.t.storage.Flush()
  70         }
  71 }
  72
  73 func (p *Piece) pendingChunkIndex(chunkIndex chunkIndexType) bool {
  74         return !p.chunkIndexDirty(chunkIndex)
  75 }
  76
  77 func (p *Piece) pendingChunk(cs ChunkSpec, chunkSize pp.Integer) bool {
  78         return p.pendingChunkIndex(chunkIndexFromChunkSpec(cs, chunkSize))
  79 }
  80
  81 func (p *Piece) hasDirtyChunks() bool {
  82         return p.numDirtyChunks() != 0
  83 }
  84
  85 func (p *Piece) numDirtyChunks() chunkIndexType {
  86         return chunkIndexType(roaringBitmapRangeCardinality[RequestIndex](
  87                 &p.t.dirtyChunks,
  88                 p.requestIndexOffset(),
  89                 p.t.pieceRequestIndexOffset(p.index+1)))
  90 }
  91
  92 func (p *Piece) unpendChunkIndex(i chunkIndexType) {
  93         p.t.dirtyChunks.Add(p.requestIndexOffset() + i)
  94         p.t.updatePieceRequestOrderPiece(p.index)
  95         p.readerCond.Broadcast()
  96 }
  97
  98 func (p *Piece) pendChunkIndex(i RequestIndex) {
  99         p.t.dirtyChunks.Remove(p.requestIndexOffset() + i)
 100         p.t.updatePieceRequestOrderPiece(p.index)
 101 }
 102
 103 func (p *Piece) numChunks() chunkIndexType {
 104         return p.t.pieceNumChunks(p.index)
 105 }
 106
 107 func (p *Piece) incrementPendingWrites() {
 108         p.pendingWritesMutex.Lock()
 109         p.pendingWrites++
 110         p.pendingWritesMutex.Unlock()
 111 }
 112
 113 func (p *Piece) decrementPendingWrites() {
 114         p.pendingWritesMutex.Lock()
 115         if p.pendingWrites == 0 {
 116                 panic("assertion")
 117         }
 118         p.pendingWrites--
 119         if p.pendingWrites == 0 {
 120                 p.noPendingWrites.Broadcast()
 121         }
 122         p.pendingWritesMutex.Unlock()
 123 }
 124
 125 func (p *Piece) waitNoPendingWrites() {
 126         p.pendingWritesMutex.Lock()
 127         for p.pendingWrites != 0 {
 128                 p.noPendingWrites.Wait()
 129         }
 130         p.pendingWritesMutex.Unlock()
 131 }
 132
 133 func (p *Piece) chunkIndexDirty(chunk chunkIndexType) bool {
 134         return p.t.dirtyChunks.Contains(p.requestIndexOffset() + chunk)
 135 }
 136
 137 func (p *Piece) chunkIndexSpec(chunk chunkIndexType) ChunkSpec {
 138         return chunkIndexSpec(pp.Integer(chunk), p.length(), p.chunkSize())
 139 }
 140
 141 func (p *Piece) numDirtyBytes() (ret pp.Integer) {
 142         // defer func() {
 143         //      if ret > p.length() {
 144         //              panic("too many dirty bytes")
 145         //      }
 146         // }()
 147         numRegularDirtyChunks := p.numDirtyChunks()
 148         if p.chunkIndexDirty(p.numChunks() - 1) {
 149                 numRegularDirtyChunks--
 150                 ret += p.chunkIndexSpec(p.lastChunkIndex()).Length
 151         }
 152         ret += pp.Integer(numRegularDirtyChunks) * p.chunkSize()
 153         return
 154 }
 155
 156 func (p *Piece) length() pp.Integer {
 157         return p.t.pieceLength(p.index)
 158 }
 159
 160 func (p *Piece) chunkSize() pp.Integer {
 161         return p.t.chunkSize
 162 }
 163
 164 func (p *Piece) lastChunkIndex() chunkIndexType {
 165         return p.numChunks() - 1
 166 }
 167
 168 func (p *Piece) bytesLeft() (ret pp.Integer) {
 169         if p.t.pieceComplete(p.index) {
 170                 return 0
 171         }
 172         return p.length() - p.numDirtyBytes()
 173 }
 174
 175 // Forces the piece data to be rehashed.
 176 func (p *Piece) VerifyData() {
 177         p.t.cl.lock()
 178         defer p.t.cl.unlock()
 179         target := p.numVerifies + 1
 180         if p.hashing {
 181                 target++
 182         }
 183         // log.Printf("target: %d", target)
 184         p.t.queuePieceCheck(p.index)
 185         for {
 186                 // log.Printf("got %d verifies", p.numVerifies)
 187                 if p.numVerifies >= target {
 188                         break
 189                 }
 190                 p.t.cl.event.Wait()
 191         }
 192         // log.Print("done")
 193 }
 194
 195 func (p *Piece) queuedForHash() bool {
 196         return p.t.piecesQueuedForHash.Get(bitmap.BitIndex(p.index))
 197 }
 198
 199 func (p *Piece) torrentBeginOffset() int64 {
 200         return int64(p.index) * p.t.info.PieceLength
 201 }
 202
 203 func (p *Piece) torrentEndOffset() int64 {
 204         return p.torrentBeginOffset() + int64(p.t.usualPieceSize())
 205 }
 206
 207 func (p *Piece) SetPriority(prio PiecePriority) {
 208         p.t.cl.lock()
 209         defer p.t.cl.unlock()
 210         p.priority = prio
 211         p.t.updatePiecePriority(p.index, "Piece.SetPriority")
 212 }
 213
 214 // This is priority based only on piece, file and reader priorities.
 215 func (p *Piece) purePriority() (ret PiecePriority) {
 216         for _, f := range p.files {
 217                 ret.Raise(f.prio)
 218         }
 219         if p.t.readerNowPieces().Contains(bitmap.BitIndex(p.index)) {
 220                 ret.Raise(PiecePriorityNow)
 221         }
 222         // if t._readerNowPieces.Contains(piece - 1) {
 223         //      return PiecePriorityNext
 224         // }
 225         if p.t.readerReadaheadPieces().Contains(bitmap.BitIndex(p.index)) {
 226                 ret.Raise(PiecePriorityReadahead)
 227         }
 228         ret.Raise(p.priority)
 229         return
 230 }
 231
 232 func (p *Piece) ignoreForRequests() bool {
 233         return p.hashing || p.marking || !p.haveHash() || p.t.pieceComplete(p.index) || p.queuedForHash()
 234 }
 235
 236 // This is the priority adjusted for piece state like completion, hashing etc.
 237 func (p *Piece) effectivePriority() (ret PiecePriority) {
 238         if p.ignoreForRequests() {
 239                 return PiecePriorityNone
 240         }
 241         return p.purePriority()
 242 }
 243
 244 // Tells the Client to refetch the completion status from storage, updating priority etc. if
 245 // necessary. Might be useful if you know the state of the piece data has changed externally.
 246 func (p *Piece) UpdateCompletion() {
 247         p.t.cl.lock()
 248         defer p.t.cl.unlock()
 249         p.t.updatePieceCompletion(p.index)
 250 }
 251
 252 func (p *Piece) completion() (ret storage.Completion) {
 253         ret.Complete = p.t.pieceComplete(p.index)
 254         ret.Ok = p.storageCompletionOk
 255         return
 256 }
 257
 258 func (p *Piece) allChunksDirty() bool {
 259         return p.numDirtyChunks() == p.numChunks()
 260 }
 261
 262 func (p *Piece) State() PieceState {
 263         return p.t.PieceState(p.index)
 264 }
 265
 266 func (p *Piece) requestIndexOffset() RequestIndex {
 267         return p.t.pieceRequestIndexOffset(p.index)
 268 }
 269
 270 func (p *Piece) availability() int {
 271         return len(p.t.connsWithAllPieces) + p.relativeAvailability
 272 }
 273
 274 // For v2 torrents, files are aligned to pieces so there should always only be a single file for a
 275 // given piece.
 276 func (p *Piece) mustGetOnlyFile() *File {
 277         if len(p.files) != 1 {
 278                 panic(len(p.files))
 279         }
 280         return p.files[0]
 281 }
 282
 283 // Sets the v2 piece hash, queuing initial piece checks if appropriate.
 284 func (p *Piece) setV2Hash(v2h [32]byte) {
 285         // See Torrent.onSetInfo. We want to trigger an initial check if appropriate, if we didn't yet
 286         // have a piece hash (can occur with v2 when we don't start with piece layers).
 287         if !p.hashV2.Set(v2h).Ok && p.hash == nil {
 288                 p.t.updatePieceCompletion(p.index)
 289                 p.t.queueInitialPieceCheck(p.index)
 290         }
 291 }
 292
 293 // Can't do certain things if we don't know the piece hash.
 294 func (p *Piece) haveHash() bool {
 295         return p.hash != nil || p.hashV2.Ok
 296 }
 297
 298 func pieceStateAllowsMessageWrites(p *Piece, pc *PeerConn) bool {
 299         return (pc.shouldRequestHashes() && !p.haveHash()) || !p.t.ignorePieceForRequests(p.index)
 300 }