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