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