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.hasPieceLayer() {
  62                 pieceHash.Set(p.mustGetOnlyFile().piecesRoot.UnwrapPtr()[:])
  63         } else if p.hashV2.Ok {
  64                 pieceHash.Set(p.hashV2.Value[:])
  65         }
  66         return p.t.storage.PieceWithHash(p.Info(), pieceHash)
  67 }
  68
  69 func (p *Piece) Flush() {
  70         if p.t.storage.Flush != nil {
  71                 _ = p.t.storage.Flush()
  72         }
  73 }
  74
  75 func (p *Piece) pendingChunkIndex(chunkIndex chunkIndexType) bool {
  76         return !p.chunkIndexDirty(chunkIndex)
  77 }
  78
  79 func (p *Piece) pendingChunk(cs ChunkSpec, chunkSize pp.Integer) bool {
  80         return p.pendingChunkIndex(chunkIndexFromChunkSpec(cs, chunkSize))
  81 }
  82
  83 func (p *Piece) hasDirtyChunks() bool {
  84         return p.numDirtyChunks() != 0
  85 }
  86
  87 func (p *Piece) numDirtyChunks() chunkIndexType {
  88         return chunkIndexType(roaringBitmapRangeCardinality[RequestIndex](
  89                 &p.t.dirtyChunks,
  90                 p.requestIndexOffset(),
  91                 p.t.pieceRequestIndexOffset(p.index+1)))
  92 }
  93
  94 func (p *Piece) unpendChunkIndex(i chunkIndexType) {
  95         p.t.dirtyChunks.Add(p.requestIndexOffset() + i)
  96         p.t.updatePieceRequestOrderPiece(p.index)
  97         p.readerCond.Broadcast()
  98 }
  99
 100 func (p *Piece) pendChunkIndex(i RequestIndex) {
 101         p.t.dirtyChunks.Remove(p.requestIndexOffset() + i)
 102         p.t.updatePieceRequestOrderPiece(p.index)
 103 }
 104
 105 func (p *Piece) numChunks() chunkIndexType {
 106         return p.t.pieceNumChunks(p.index)
 107 }
 108
 109 func (p *Piece) incrementPendingWrites() {
 110         p.pendingWritesMutex.Lock()
 111         p.pendingWrites++
 112         p.pendingWritesMutex.Unlock()
 113 }
 114
 115 func (p *Piece) decrementPendingWrites() {
 116         p.pendingWritesMutex.Lock()
 117         if p.pendingWrites == 0 {
 118                 panic("assertion")
 119         }
 120         p.pendingWrites--
 121         if p.pendingWrites == 0 {
 122                 p.noPendingWrites.Broadcast()
 123         }
 124         p.pendingWritesMutex.Unlock()
 125 }
 126
 127 func (p *Piece) waitNoPendingWrites() {
 128         p.pendingWritesMutex.Lock()
 129         for p.pendingWrites != 0 {
 130                 p.noPendingWrites.Wait()
 131         }
 132         p.pendingWritesMutex.Unlock()
 133 }
 134
 135 func (p *Piece) chunkIndexDirty(chunk chunkIndexType) bool {
 136         return p.t.dirtyChunks.Contains(p.requestIndexOffset() + chunk)
 137 }
 138
 139 func (p *Piece) chunkIndexSpec(chunk chunkIndexType) ChunkSpec {
 140         return chunkIndexSpec(pp.Integer(chunk), p.length(), p.chunkSize())
 141 }
 142
 143 func (p *Piece) numDirtyBytes() (ret pp.Integer) {
 144         // defer func() {
 145         //      if ret > p.length() {
 146         //              panic("too many dirty bytes")
 147         //      }
 148         // }()
 149         numRegularDirtyChunks := p.numDirtyChunks()
 150         if p.chunkIndexDirty(p.numChunks() - 1) {
 151                 numRegularDirtyChunks--
 152                 ret += p.chunkIndexSpec(p.lastChunkIndex()).Length
 153         }
 154         ret += pp.Integer(numRegularDirtyChunks) * p.chunkSize()
 155         return
 156 }
 157
 158 func (p *Piece) length() pp.Integer {
 159         return p.t.pieceLength(p.index)
 160 }
 161
 162 func (p *Piece) chunkSize() pp.Integer {
 163         return p.t.chunkSize
 164 }
 165
 166 func (p *Piece) lastChunkIndex() chunkIndexType {
 167         return p.numChunks() - 1
 168 }
 169
 170 func (p *Piece) bytesLeft() (ret pp.Integer) {
 171         if p.t.pieceComplete(p.index) {
 172                 return 0
 173         }
 174         return p.length() - p.numDirtyBytes()
 175 }
 176
 177 // Forces the piece data to be rehashed.
 178 func (p *Piece) VerifyData() {
 179         p.t.cl.lock()
 180         defer p.t.cl.unlock()
 181         target := p.numVerifies + 1
 182         if p.hashing {
 183                 target++
 184         }
 185         // log.Printf("target: %d", target)
 186         p.t.queuePieceCheck(p.index)
 187         for {
 188                 // log.Printf("got %d verifies", p.numVerifies)
 189                 if p.numVerifies >= target {
 190                         break
 191                 }
 192                 p.t.cl.event.Wait()
 193         }
 194         // log.Print("done")
 195 }
 196
 197 func (p *Piece) queuedForHash() bool {
 198         return p.t.piecesQueuedForHash.Get(bitmap.BitIndex(p.index))
 199 }
 200
 201 func (p *Piece) torrentBeginOffset() int64 {
 202         return int64(p.index) * p.t.info.PieceLength
 203 }
 204
 205 func (p *Piece) torrentEndOffset() int64 {
 206         return p.torrentBeginOffset() + int64(p.t.usualPieceSize())
 207 }
 208
 209 func (p *Piece) SetPriority(prio PiecePriority) {
 210         p.t.cl.lock()
 211         defer p.t.cl.unlock()
 212         p.priority = prio
 213         p.t.updatePiecePriority(p.index, "Piece.SetPriority")
 214 }
 215
 216 // This is priority based only on piece, file and reader priorities.
 217 func (p *Piece) purePriority() (ret PiecePriority) {
 218         for _, f := range p.files {
 219                 ret.Raise(f.prio)
 220         }
 221         if p.t.readerNowPieces().Contains(bitmap.BitIndex(p.index)) {
 222                 ret.Raise(PiecePriorityNow)
 223         }
 224         // if t._readerNowPieces.Contains(piece - 1) {
 225         //      return PiecePriorityNext
 226         // }
 227         if p.t.readerReadaheadPieces().Contains(bitmap.BitIndex(p.index)) {
 228                 ret.Raise(PiecePriorityReadahead)
 229         }
 230         ret.Raise(p.priority)
 231         return
 232 }
 233
 234 func (p *Piece) ignoreForRequests() bool {
 235         return p.hashing || p.marking || !p.haveHash() || p.t.pieceComplete(p.index) || p.queuedForHash()
 236 }
 237
 238 // This is the priority adjusted for piece state like completion, hashing etc.
 239 func (p *Piece) effectivePriority() (ret PiecePriority) {
 240         if p.ignoreForRequests() {
 241                 return PiecePriorityNone
 242         }
 243         return p.purePriority()
 244 }
 245
 246 // Tells the Client to refetch the completion status from storage, updating priority etc. if
 247 // necessary. Might be useful if you know the state of the piece data has changed externally.
 248 func (p *Piece) UpdateCompletion() {
 249         p.t.cl.lock()
 250         defer p.t.cl.unlock()
 251         p.t.updatePieceCompletion(p.index)
 252 }
 253
 254 func (p *Piece) completion() (ret storage.Completion) {
 255         ret.Complete = p.t.pieceComplete(p.index)
 256         ret.Ok = p.storageCompletionOk
 257         return
 258 }
 259
 260 func (p *Piece) allChunksDirty() bool {
 261         return p.numDirtyChunks() == p.numChunks()
 262 }
 263
 264 func (p *Piece) State() PieceState {
 265         return p.t.PieceState(p.index)
 266 }
 267
 268 func (p *Piece) requestIndexOffset() RequestIndex {
 269         return p.t.pieceRequestIndexOffset(p.index)
 270 }
 271
 272 func (p *Piece) availability() int {
 273         return len(p.t.connsWithAllPieces) + p.relativeAvailability
 274 }
 275
 276 // For v2 torrents, files are aligned to pieces so there should always only be a single file for a
 277 // given piece.
 278 func (p *Piece) mustGetOnlyFile() *File {
 279         if len(p.files) != 1 {
 280                 panic(len(p.files))
 281         }
 282         return p.files[0]
 283 }
 284
 285 // Sets the v2 piece hash, queuing initial piece checks if appropriate.
 286 func (p *Piece) setV2Hash(v2h [32]byte) {
 287         // See Torrent.onSetInfo. We want to trigger an initial check if appropriate, if we didn't yet
 288         // have a piece hash (can occur with v2 when we don't start with piece layers).
 289         p.t.storageLock.Lock()
 290         oldV2Hash := p.hashV2.Set(v2h)
 291         p.t.storageLock.Unlock()
 292         if !oldV2Hash.Ok && p.hash == nil {
 293                 p.t.updatePieceCompletion(p.index)
 294                 p.t.queueInitialPieceCheck(p.index)
 295         }
 296 }
 297
 298 // Can't do certain things if we don't know the piece hash.
 299 func (p *Piece) haveHash() bool {
 300         if p.hash != nil {
 301                 return true
 302         }
 303         if !p.hasPieceLayer() {
 304                 return true
 305         }
 306         return p.hashV2.Ok
 307 }
 308
 309 func (p *Piece) hasPieceLayer() bool {
 310         return int64(p.length()) > p.t.info.PieceLength
 311 }