piece.go

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