reader.go

   1 package torrent
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "io"
   8         "sync"
   9
  10         "github.com/anacrolix/log"
  11         "github.com/anacrolix/missinggo/v2"
  12 )
  13
  14 // Accesses Torrent data via a Client. Reads block until the data is available. Seeks and readahead
  15 // also drive Client behaviour.
  16 type Reader interface {
  17         io.ReadSeekCloser
  18         missinggo.ReadContexter
  19         // Configure the number of bytes ahead of a read that should also be prioritized in preparation
  20         // for further reads. Overridden by non-nil readahead func, see SetReadaheadFunc.
  21         SetReadahead(int64)
  22         // If non-nil, the provided function is called when the implementation needs to know the
  23         // readahead for the current reader. Calls occur during Reads and Seeks, and while the Client is
  24         // locked.
  25         SetReadaheadFunc(func() int64)
  26         // Don't wait for pieces to complete and be verified. Read calls return as soon as they can when
  27         // the underlying chunks become available.
  28         SetResponsive()
  29 }
  30
  31 // Piece range by piece index, [begin, end).
  32 type pieceRange struct {
  33         begin, end pieceIndex
  34 }
  35
  36 type reader struct {
  37         t *Torrent
  38         // Adjust the read/seek window to handle Readers locked to File extents and the like.
  39         offset, length int64
  40
  41         // Function to dynamically calculate readahead. If nil, readahead is static.
  42         readaheadFunc func() int64
  43
  44         // Required when modifying pos and readahead.
  45         mu sync.Locker
  46
  47         readahead, pos int64
  48         // Position that reads have continued contiguously from.
  49         contiguousReadStartPos int64
  50         // The cached piece range this reader wants downloaded. The zero value corresponds to nothing.
  51         // We cache this so that changes can be detected, and bubbled up to the Torrent only as
  52         // required.
  53         pieces pieceRange
  54
  55         // Reads have been initiated since the last seek. This is used to prevent readaheads occurring
  56         // after a seek or with a new reader at the starting position.
  57         reading    bool
  58         responsive bool
  59 }
  60
  61 var _ io.ReadSeekCloser = (*reader)(nil)
  62
  63 func (r *reader) SetResponsive() {
  64         r.responsive = true
  65         r.t.cl.event.Broadcast()
  66 }
  67
  68 // Disable responsive mode. TODO: Remove?
  69 func (r *reader) SetNonResponsive() {
  70         r.responsive = false
  71         r.t.cl.event.Broadcast()
  72 }
  73
  74 func (r *reader) SetReadahead(readahead int64) {
  75         r.mu.Lock()
  76         r.readahead = readahead
  77         r.readaheadFunc = nil
  78         r.posChanged()
  79         r.mu.Unlock()
  80 }
  81
  82 func (r *reader) SetReadaheadFunc(f func() int64) {
  83         r.mu.Lock()
  84         r.readaheadFunc = f
  85         r.posChanged()
  86         r.mu.Unlock()
  87 }
  88
  89 // How many bytes are available to read. Max is the most we could require.
  90 func (r *reader) available(off, max int64) (ret int64) {
  91         off += r.offset
  92         for max > 0 {
  93                 req, ok := r.t.offsetRequest(off)
  94                 if !ok {
  95                         break
  96                 }
  97                 if !r.responsive && !r.t.pieceComplete(pieceIndex(req.Index)) {
  98                         break
  99                 }
 100                 if !r.t.haveChunk(req) {
 101                         break
 102                 }
 103                 len1 := int64(req.Length) - (off - r.t.requestOffset(req))
 104                 max -= len1
 105                 ret += len1
 106                 off += len1
 107         }
 108         // Ensure that ret hasn't exceeded our original max.
 109         if max < 0 {
 110                 ret += max
 111         }
 112         return
 113 }
 114
 115 // Calculates the pieces this reader wants downloaded, ignoring the cached value at r.pieces.
 116 func (r *reader) piecesUncached() (ret pieceRange) {
 117         ra := r.readahead
 118         if r.readaheadFunc != nil {
 119                 ra = r.readaheadFunc()
 120         }
 121         if ra < 1 {
 122                 // Needs to be at least 1, because [x, x) means we don't want
 123                 // anything.
 124                 ra = 1
 125         }
 126         if !r.reading {
 127                 ra = 0
 128         }
 129         if ra > r.length-r.pos {
 130                 ra = r.length - r.pos
 131         }
 132         ret.begin, ret.end = r.t.byteRegionPieces(r.torrentOffset(r.pos), ra)
 133         return
 134 }
 135
 136 func (r *reader) Read(b []byte) (n int, err error) {
 137         return r.ReadContext(context.Background(), b)
 138 }
 139
 140 func (r *reader) ReadContext(ctx context.Context, b []byte) (n int, err error) {
 141         if len(b) > 0 {
 142                 r.reading = true
 143                 // TODO: Rework reader piece priorities so we don't have to push updates in to the Client
 144                 // and take the lock here.
 145                 r.mu.Lock()
 146                 r.posChanged()
 147                 r.mu.Unlock()
 148         }
 149         n, err = r.readOnceAt(ctx, b, r.pos)
 150         if n == 0 {
 151                 if err == nil && len(b) > 0 {
 152                         panic("expected error")
 153                 } else {
 154                         return
 155                 }
 156         }
 157
 158         r.mu.Lock()
 159         r.pos += int64(n)
 160         r.posChanged()
 161         r.mu.Unlock()
 162         if r.pos >= r.length {
 163                 err = io.EOF
 164         } else if err == io.EOF {
 165                 err = io.ErrUnexpectedEOF
 166         }
 167         return
 168 }
 169
 170 var closedChan = make(chan struct{})
 171
 172 func init() {
 173         close(closedChan)
 174 }
 175
 176 // Wait until some data should be available to read. Tickles the client if it isn't. Returns how
 177 // much should be readable without blocking.
 178 func (r *reader) waitAvailable(ctx context.Context, pos, wanted int64, wait bool) (avail int64, err error) {
 179         t := r.t
 180         for {
 181                 r.t.cl.rLock()
 182                 avail = r.available(pos, wanted)
 183                 readerCond := t.piece(int((r.offset + pos) / t.info.PieceLength)).readerCond.Signaled()
 184                 r.t.cl.rUnlock()
 185                 if avail != 0 {
 186                         return
 187                 }
 188                 var dontWait <-chan struct{}
 189                 if !wait || wanted == 0 {
 190                         dontWait = closedChan
 191                 }
 192                 select {
 193                 case <-r.t.closed.Done():
 194                         err = errors.New("torrent closed")
 195                         return
 196                 case <-ctx.Done():
 197                         err = ctx.Err()
 198                         return
 199                 case <-r.t.dataDownloadDisallowed.On():
 200                         err = errors.New("torrent data downloading disabled")
 201                 case <-r.t.networkingEnabled.Off():
 202                         err = errors.New("torrent networking disabled")
 203                         return
 204                 case <-dontWait:
 205                         return
 206                 case <-readerCond:
 207                 }
 208         }
 209 }
 210
 211 // Adds the reader's torrent offset to the reader object offset (for example the reader might be
 212 // constrainted to a particular file within the torrent).
 213 func (r *reader) torrentOffset(readerPos int64) int64 {
 214         return r.offset + readerPos
 215 }
 216
 217 // Performs at most one successful read to torrent storage.
 218 func (r *reader) readOnceAt(ctx context.Context, b []byte, pos int64) (n int, err error) {
 219         if pos >= r.length {
 220                 err = io.EOF
 221                 return
 222         }
 223         for {
 224                 var avail int64
 225                 avail, err = r.waitAvailable(ctx, pos, int64(len(b)), n == 0)
 226                 if avail == 0 {
 227                         return
 228                 }
 229                 firstPieceIndex := pieceIndex(r.torrentOffset(pos) / r.t.info.PieceLength)
 230                 firstPieceOffset := r.torrentOffset(pos) % r.t.info.PieceLength
 231                 b1 := missinggo.LimitLen(b, avail)
 232                 n, err = r.t.readAt(b1, r.torrentOffset(pos))
 233                 if n != 0 {
 234                         err = nil
 235                         return
 236                 }
 237                 r.t.cl.lock()
 238                 // TODO: Just reset pieces in the readahead window. This might help
 239                 // prevent thrashing with small caches and file and piece priorities.
 240                 r.log(log.Fstr("error reading torrent %s piece %d offset %d, %d bytes: %v",
 241                         r.t.infoHash.HexString(), firstPieceIndex, firstPieceOffset, len(b1), err))
 242                 if !r.t.updatePieceCompletion(firstPieceIndex) {
 243                         r.log(log.Fstr("piece %d completion unchanged", firstPieceIndex))
 244                 }
 245                 // Update the rest of the piece completions in the readahead window, without alerting to
 246                 // changes (since only the first piece, the one above, could have generated the read error
 247                 // we're currently handling).
 248                 if r.pieces.begin != firstPieceIndex {
 249                         panic(fmt.Sprint(r.pieces.begin, firstPieceIndex))
 250                 }
 251                 for index := r.pieces.begin + 1; index < r.pieces.end; index++ {
 252                         r.t.updatePieceCompletion(index)
 253                 }
 254                 r.t.cl.unlock()
 255         }
 256 }
 257
 258 // Hodor
 259 func (r *reader) Close() error {
 260         r.t.cl.lock()
 261         r.t.deleteReader(r)
 262         r.t.cl.unlock()
 263         return nil
 264 }
 265
 266 func (r *reader) posChanged() {
 267         to := r.piecesUncached()
 268         from := r.pieces
 269         if to == from {
 270                 return
 271         }
 272         r.pieces = to
 273         // log.Printf("reader pos changed %v->%v", from, to)
 274         r.t.readerPosChanged(from, to)
 275 }
 276
 277 func (r *reader) Seek(off int64, whence int) (newPos int64, err error) {
 278         switch whence {
 279         case io.SeekStart:
 280                 newPos = off
 281                 r.mu.Lock()
 282         case io.SeekCurrent:
 283                 r.mu.Lock()
 284                 newPos = r.pos + off
 285         case io.SeekEnd:
 286                 newPos = r.length + off
 287                 r.mu.Lock()
 288         default:
 289                 return 0, errors.New("bad whence")
 290         }
 291         if newPos != r.pos {
 292                 r.reading = false
 293                 r.pos = newPos
 294                 r.contiguousReadStartPos = newPos
 295                 r.posChanged()
 296         }
 297         r.mu.Unlock()
 298         return
 299 }
 300
 301 func (r *reader) log(m log.Msg) {
 302         r.t.logger.Log(m.Skip(1))
 303 }
 304
 305 // Implementation inspired by https://news.ycombinator.com/item?id=27019613.
 306 func (r *reader) defaultReadaheadFunc() int64 {
 307         return r.pos - r.contiguousReadStartPos
 308 }