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