piece.go

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