piece.go

   1 package torrent
   2
   3 import (
   4         "fmt"
   5         "sync"
   6
   7         "github.com/anacrolix/missinggo/v2/bitmap"
   8
   9         "github.com/anacrolix/torrent/metainfo"
  10         pp "github.com/anacrolix/torrent/peer_protocol"
  11         "github.com/anacrolix/torrent/storage"
  12 )
  13
  14 // Describes the importance of obtaining a particular piece.
  15 type piecePriority byte
  16
  17 func (pp *piecePriority) Raise(maybe piecePriority) bool {
  18         if maybe > *pp {
  19                 *pp = maybe
  20                 return true
  21         }
  22         return false
  23 }
  24
  25 // Priority for use in PriorityBitmap
  26 func (me piecePriority) BitmapPriority() int {
  27         return -int(me)
  28 }
  29
  30 const (
  31         PiecePriorityNone      piecePriority = iota // Not wanted. Must be the zero value.
  32         PiecePriorityNormal                         // Wanted.
  33         PiecePriorityHigh                           // Wanted a lot.
  34         PiecePriorityReadahead                      // May be required soon.
  35         // Succeeds a piece where a read occurred. Currently the same as Now,
  36         // apparently due to issues with caching.
  37         PiecePriorityNext
  38         PiecePriorityNow // A Reader is reading in this piece. Highest urgency.
  39 )
  40
  41 type Piece struct {
  42         // The completed piece SHA1 hash, from the metainfo "pieces" field.
  43         hash  *metainfo.Hash
  44         t     *Torrent
  45         index pieceIndex
  46         files []*File
  47         // Chunks we've written to since the last check. The chunk offset and
  48         // length can be determined by the request chunkSize in use.
  49         _dirtyChunks bitmap.Bitmap
  50
  51         hashing             bool
  52         numVerifies         int64
  53         storageCompletionOk bool
  54
  55         publicPieceState PieceState
  56         priority         piecePriority
  57
  58         pendingWritesMutex sync.Mutex
  59         pendingWrites      int
  60         noPendingWrites    sync.Cond
  61
  62         // Connections that have written data to this piece since its last check.
  63         // This can include connections that have closed.
  64         dirtiers map[*connection]struct{}
  65 }
  66
  67 func (p *Piece) String() string {
  68         return fmt.Sprintf("%s/%d", p.t.infoHash.HexString(), p.index)
  69 }
  70
  71 func (p *Piece) Info() metainfo.Piece {
  72         return p.t.info.Piece(int(p.index))
  73 }
  74
  75 func (p *Piece) Storage() storage.Piece {
  76         return p.t.storage.Piece(p.Info())
  77 }
  78
  79 func (p *Piece) pendingChunkIndex(chunkIndex int) bool {
  80         return !p._dirtyChunks.Contains(chunkIndex)
  81 }
  82
  83 func (p *Piece) pendingChunk(cs chunkSpec, chunkSize pp.Integer) bool {
  84         return p.pendingChunkIndex(chunkIndex(cs, chunkSize))
  85 }
  86
  87 func (p *Piece) hasDirtyChunks() bool {
  88         return p._dirtyChunks.Len() != 0
  89 }
  90
  91 func (p *Piece) numDirtyChunks() pp.Integer {
  92         return pp.Integer(p._dirtyChunks.Len())
  93 }
  94
  95 func (p *Piece) unpendChunkIndex(i int) {
  96         p._dirtyChunks.Add(i)
  97         p.t.tickleReaders()
  98 }
  99
 100 func (p *Piece) pendChunkIndex(i int) {
 101         p._dirtyChunks.Remove(i)
 102 }
 103
 104 func (p *Piece) numChunks() pp.Integer {
 105         return p.t.pieceNumChunks(p.index)
 106 }
 107
 108 func (p *Piece) undirtiedChunkIndices() (ret bitmap.Bitmap) {
 109         ret = p._dirtyChunks.Copy()
 110         ret.FlipRange(0, bitmap.BitIndex(p.numChunks()))
 111         return
 112 }
 113
 114 func (p *Piece) incrementPendingWrites() {
 115         p.pendingWritesMutex.Lock()
 116         p.pendingWrites++
 117         p.pendingWritesMutex.Unlock()
 118 }
 119
 120 func (p *Piece) decrementPendingWrites() {
 121         p.pendingWritesMutex.Lock()
 122         if p.pendingWrites == 0 {
 123                 panic("assertion")
 124         }
 125         p.pendingWrites--
 126         if p.pendingWrites == 0 {
 127                 p.noPendingWrites.Broadcast()
 128         }
 129         p.pendingWritesMutex.Unlock()
 130 }
 131
 132 func (p *Piece) waitNoPendingWrites() {
 133         p.pendingWritesMutex.Lock()
 134         for p.pendingWrites != 0 {
 135                 p.noPendingWrites.Wait()
 136         }
 137         p.pendingWritesMutex.Unlock()
 138 }
 139
 140 func (p *Piece) chunkIndexDirty(chunk pp.Integer) bool {
 141         return p._dirtyChunks.Contains(bitmap.BitIndex(chunk))
 142 }
 143
 144 func (p *Piece) chunkIndexSpec(chunk pp.Integer) chunkSpec {
 145         return chunkIndexSpec(chunk, p.length(), p.chunkSize())
 146 }
 147
 148 func (p *Piece) chunkIndexRequest(chunkIndex pp.Integer) request {
 149         return request{
 150                 pp.Integer(p.index),
 151                 chunkIndexSpec(chunkIndex, p.length(), p.chunkSize()),
 152         }
 153 }
 154
 155 func (p *Piece) numDirtyBytes() (ret pp.Integer) {
 156         // defer func() {
 157         //      if ret > p.length() {
 158         //              panic("too many dirty bytes")
 159         //      }
 160         // }()
 161         numRegularDirtyChunks := p.numDirtyChunks()
 162         if p.chunkIndexDirty(p.numChunks() - 1) {
 163                 numRegularDirtyChunks--
 164                 ret += p.chunkIndexSpec(p.lastChunkIndex()).Length
 165         }
 166         ret += pp.Integer(numRegularDirtyChunks) * p.chunkSize()
 167         return
 168 }
 169
 170 func (p *Piece) length() pp.Integer {
 171         return p.t.pieceLength(p.index)
 172 }
 173
 174 func (p *Piece) chunkSize() pp.Integer {
 175         return p.t.chunkSize
 176 }
 177
 178 func (p *Piece) lastChunkIndex() pp.Integer {
 179         return p.numChunks() - 1
 180 }
 181
 182 func (p *Piece) bytesLeft() (ret pp.Integer) {
 183         if p.t.pieceComplete(p.index) {
 184                 return 0
 185         }
 186         return p.length() - p.numDirtyBytes()
 187 }
 188
 189 // Forces the piece data to be rehashed.
 190 func (p *Piece) VerifyData() {
 191         p.t.cl.lock()
 192         defer p.t.cl.unlock()
 193         target := p.numVerifies + 1
 194         if p.hashing {
 195                 target++
 196         }
 197         //log.Printf("target: %d", target)
 198         p.t.queuePieceCheck(p.index)
 199         for {
 200                 //log.Printf("got %d verifies", p.numVerifies)
 201                 if p.numVerifies >= target {
 202                         break
 203                 }
 204                 p.t.cl.event.Wait()
 205         }
 206         // log.Print("done")
 207 }
 208
 209 func (p *Piece) queuedForHash() bool {
 210         return p.t.piecesQueuedForHash.Get(bitmap.BitIndex(p.index))
 211 }
 212
 213 func (p *Piece) torrentBeginOffset() int64 {
 214         return int64(p.index) * p.t.info.PieceLength
 215 }
 216
 217 func (p *Piece) torrentEndOffset() int64 {
 218         return p.torrentBeginOffset() + int64(p.length())
 219 }
 220
 221 func (p *Piece) SetPriority(prio piecePriority) {
 222         p.t.cl.lock()
 223         defer p.t.cl.unlock()
 224         p.priority = prio
 225         p.t.updatePiecePriority(p.index)
 226 }
 227
 228 func (p *Piece) uncachedPriority() (ret piecePriority) {
 229         if p.t.pieceComplete(p.index) || p.t.pieceQueuedForHash(p.index) || p.t.hashingPiece(p.index) {
 230                 return PiecePriorityNone
 231         }
 232         for _, f := range p.files {
 233                 ret.Raise(f.prio)
 234         }
 235         if p.t.readerNowPieces().Contains(int(p.index)) {
 236                 ret.Raise(PiecePriorityNow)
 237         }
 238         // if t._readerNowPieces.Contains(piece - 1) {
 239         //      return PiecePriorityNext
 240         // }
 241         if p.t.readerReadaheadPieces().Contains(bitmap.BitIndex(p.index)) {
 242                 ret.Raise(PiecePriorityReadahead)
 243         }
 244         ret.Raise(p.priority)
 245         return
 246 }
 247
 248 func (p *Piece) completion() (ret storage.Completion) {
 249         ret.Complete = p.t.pieceComplete(p.index)
 250         ret.Ok = p.storageCompletionOk
 251         return
 252 }
 253
 254 func (p *Piece) allChunksDirty() bool {
 255         return p._dirtyChunks.Len() == int(p.numChunks())
 256 }
 257
 258 func (p *Piece) requestStrategyPiece() requestStrategyPiece {
 259         return p
 260 }
 261
 262 func (p *Piece) dirtyChunks() bitmap.Bitmap {
 263         return p._dirtyChunks
 264 }
 265
 266 func (rs requestStrategyThree) wouldDuplicateRecent(r request) bool {
 267         // This piece has been requested on another connection, and the duplicate request timer is still
 268         // running.
 269         _, ok := rs.lastRequested[r]
 270         return ok
 271 }