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         numVerifies         int64
  52         hashing             bool
  53         marking             bool
  54         storageCompletionOk bool
  55
  56         publicPieceState PieceState
  57         priority         piecePriority
  58
  59         // This can be locked when the Client lock is taken, but probably not vice versa.
  60         pendingWritesMutex sync.Mutex
  61         pendingWrites      int
  62         noPendingWrites    sync.Cond
  63
  64         // Connections that have written data to this piece since its last check.
  65         // This can include connections that have closed.
  66         dirtiers map[*peer]struct{}
  67 }
  68
  69 func (p *Piece) String() string {
  70         return fmt.Sprintf("%s/%d", p.t.infoHash.HexString(), p.index)
  71 }
  72
  73 func (p *Piece) Info() metainfo.Piece {
  74         return p.t.info.Piece(int(p.index))
  75 }
  76
  77 func (p *Piece) Storage() storage.Piece {
  78         return p.t.storage.Piece(p.Info())
  79 }
  80
  81 func (p *Piece) pendingChunkIndex(chunkIndex int) bool {
  82         return !p._dirtyChunks.Contains(chunkIndex)
  83 }
  84
  85 func (p *Piece) pendingChunk(cs chunkSpec, chunkSize pp.Integer) bool {
  86         return p.pendingChunkIndex(chunkIndex(cs, chunkSize))
  87 }
  88
  89 func (p *Piece) hasDirtyChunks() bool {
  90         return p._dirtyChunks.Len() != 0
  91 }
  92
  93 func (p *Piece) numDirtyChunks() pp.Integer {
  94         return pp.Integer(p._dirtyChunks.Len())
  95 }
  96
  97 func (p *Piece) unpendChunkIndex(i int) {
  98         p._dirtyChunks.Add(i)
  99         p.t.tickleReaders()
 100 }
 101
 102 func (p *Piece) pendChunkIndex(i int) {
 103         p._dirtyChunks.Remove(i)
 104 }
 105
 106 func (p *Piece) numChunks() pp.Integer {
 107         return p.t.pieceNumChunks(p.index)
 108 }
 109
 110 func (p *Piece) incrementPendingWrites() {
 111         p.pendingWritesMutex.Lock()
 112         p.pendingWrites++
 113         p.pendingWritesMutex.Unlock()
 114 }
 115
 116 func (p *Piece) decrementPendingWrites() {
 117         p.pendingWritesMutex.Lock()
 118         if p.pendingWrites == 0 {
 119                 panic("assertion")
 120         }
 121         p.pendingWrites--
 122         if p.pendingWrites == 0 {
 123                 p.noPendingWrites.Broadcast()
 124         }
 125         p.pendingWritesMutex.Unlock()
 126 }
 127
 128 func (p *Piece) waitNoPendingWrites() {
 129         p.pendingWritesMutex.Lock()
 130         for p.pendingWrites != 0 {
 131                 p.noPendingWrites.Wait()
 132         }
 133         p.pendingWritesMutex.Unlock()
 134 }
 135
 136 func (p *Piece) chunkIndexDirty(chunk pp.Integer) bool {
 137         return p._dirtyChunks.Contains(bitmap.BitIndex(chunk))
 138 }
 139
 140 func (p *Piece) chunkIndexSpec(chunk pp.Integer) chunkSpec {
 141         return chunkIndexSpec(chunk, p.length(), p.chunkSize())
 142 }
 143
 144 func (p *Piece) chunkIndexRequest(chunkIndex pp.Integer) request {
 145         return request{
 146                 pp.Integer(p.index),
 147                 chunkIndexSpec(chunkIndex, p.length(), p.chunkSize()),
 148         }
 149 }
 150
 151 func (p *Piece) numDirtyBytes() (ret pp.Integer) {
 152         // defer func() {
 153         //      if ret > p.length() {
 154         //              panic("too many dirty bytes")
 155         //      }
 156         // }()
 157         numRegularDirtyChunks := p.numDirtyChunks()
 158         if p.chunkIndexDirty(p.numChunks() - 1) {
 159                 numRegularDirtyChunks--
 160                 ret += p.chunkIndexSpec(p.lastChunkIndex()).Length
 161         }
 162         ret += pp.Integer(numRegularDirtyChunks) * p.chunkSize()
 163         return
 164 }
 165
 166 func (p *Piece) length() pp.Integer {
 167         return p.t.pieceLength(p.index)
 168 }
 169
 170 func (p *Piece) chunkSize() pp.Integer {
 171         return p.t.chunkSize
 172 }
 173
 174 func (p *Piece) lastChunkIndex() pp.Integer {
 175         return p.numChunks() - 1
 176 }
 177
 178 func (p *Piece) bytesLeft() (ret pp.Integer) {
 179         if p.t.pieceComplete(p.index) {
 180                 return 0
 181         }
 182         return p.length() - p.numDirtyBytes()
 183 }
 184
 185 // Forces the piece data to be rehashed.
 186 func (p *Piece) VerifyData() {
 187         p.t.cl.lock()
 188         defer p.t.cl.unlock()
 189         target := p.numVerifies + 1
 190         if p.hashing {
 191                 target++
 192         }
 193         //log.Printf("target: %d", target)
 194         p.t.queuePieceCheck(p.index)
 195         for {
 196                 //log.Printf("got %d verifies", p.numVerifies)
 197                 if p.numVerifies >= target {
 198                         break
 199                 }
 200                 p.t.cl.event.Wait()
 201         }
 202         // log.Print("done")
 203 }
 204
 205 func (p *Piece) queuedForHash() bool {
 206         return p.t.piecesQueuedForHash.Get(bitmap.BitIndex(p.index))
 207 }
 208
 209 func (p *Piece) torrentBeginOffset() int64 {
 210         return int64(p.index) * p.t.info.PieceLength
 211 }
 212
 213 func (p *Piece) torrentEndOffset() int64 {
 214         return p.torrentBeginOffset() + int64(p.length())
 215 }
 216
 217 func (p *Piece) SetPriority(prio piecePriority) {
 218         p.t.cl.lock()
 219         defer p.t.cl.unlock()
 220         p.priority = prio
 221         p.t.updatePiecePriority(p.index)
 222 }
 223
 224 func (p *Piece) uncachedPriority() (ret piecePriority) {
 225         if p.t.pieceComplete(p.index) || p.t.pieceQueuedForHash(p.index) || p.t.hashingPiece(p.index) {
 226                 return PiecePriorityNone
 227         }
 228         for _, f := range p.files {
 229                 ret.Raise(f.prio)
 230         }
 231         if p.t.readerNowPieces().Contains(int(p.index)) {
 232                 ret.Raise(PiecePriorityNow)
 233         }
 234         // if t._readerNowPieces.Contains(piece - 1) {
 235         //      return PiecePriorityNext
 236         // }
 237         if p.t.readerReadaheadPieces().Contains(bitmap.BitIndex(p.index)) {
 238                 ret.Raise(PiecePriorityReadahead)
 239         }
 240         ret.Raise(p.priority)
 241         return
 242 }
 243
 244 // Tells the Client to refetch the completion status from storage, updating priority etc. if
 245 // necessary. Might be useful if you know the state of the piece data has changed externally.
 246 func (p *Piece) UpdateCompletion() {
 247         p.t.cl.lock()
 248         defer p.t.cl.unlock()
 249         p.t.updatePieceCompletion(p.index)
 250 }
 251
 252 func (p *Piece) completion() (ret storage.Completion) {
 253         ret.Complete = p.t.pieceComplete(p.index)
 254         ret.Ok = p.storageCompletionOk
 255         return
 256 }
 257
 258 func (p *Piece) allChunksDirty() bool {
 259         return p._dirtyChunks.Len() == int(p.numChunks())
 260 }
 261
 262 func (p *Piece) requestStrategyPiece() requestStrategyPiece {
 263         return p
 264 }
 265
 266 func (p *Piece) dirtyChunks() bitmap.Bitmap {
 267         return p._dirtyChunks
 268 }
 269
 270 func (p *Piece) State() PieceState {
 271         return p.t.PieceState(p.index)
 272 }