reader.go

   1 package torrent
   2
   3 import (
   4         "errors"
   5         "io"
   6         "os"
   7         "sync"
   8 )
   9
  10 // Accesses torrent data via a client.
  11 type Reader struct {
  12         t *Torrent
  13
  14         mu         sync.Mutex
  15         pos        int64
  16         responsive bool
  17         readahead  int64
  18 }
  19
  20 var _ io.ReadCloser = &Reader{}
  21
  22 // Don't wait for pieces to complete and be verified. Read calls return as
  23 // soon as they can when the underlying chunks become available.
  24 func (r *Reader) SetResponsive() {
  25         r.responsive = true
  26 }
  27
  28 // Configure the number of bytes ahead of a read that should also be
  29 // prioritized in preparation for further reads.
  30 func (r *Reader) SetReadahead(readahead int64) {
  31         r.mu.Lock()
  32         defer r.mu.Unlock()
  33         r.readahead = readahead
  34 }
  35
  36 func (r *Reader) readable(off int64) (ret bool) {
  37         // log.Println("readable", off)
  38         // defer func() {
  39         //      log.Println("readable", ret)
  40         // }()
  41         if r.t.torrent.isClosed() {
  42                 return true
  43         }
  44         req, ok := r.t.torrent.offsetRequest(off)
  45         if !ok {
  46                 panic(off)
  47         }
  48         if r.responsive {
  49                 return r.t.torrent.haveChunk(req)
  50         }
  51         return r.t.torrent.pieceComplete(int(req.Index))
  52 }
  53
  54 // How many bytes are available to read. Max is the most we could require.
  55 func (r *Reader) available(off, max int64) (ret int64) {
  56         for max > 0 {
  57                 req, ok := r.t.torrent.offsetRequest(off)
  58                 if !ok {
  59                         break
  60                 }
  61                 if !r.t.torrent.haveChunk(req) {
  62                         break
  63                 }
  64                 len1 := int64(req.Length) - (off - r.t.torrent.requestOffset(req))
  65                 max -= len1
  66                 ret += len1
  67                 off += len1
  68         }
  69         // Ensure that ret hasn't exceeded our original max.
  70         if max < 0 {
  71                 ret += max
  72         }
  73         return
  74 }
  75
  76 func (r *Reader) tickleClient() {
  77         r.t.torrent.readersChanged(r.t.cl)
  78 }
  79
  80 func (r *Reader) waitReadable(off int64) {
  81         // We may have been sent back here because we were told we could read but
  82         // it failed.
  83         r.tickleClient()
  84         r.t.cl.event.Wait()
  85 }
  86
  87 func (r *Reader) Read(b []byte) (n int, err error) {
  88         r.mu.Lock()
  89         pos := r.pos
  90         r.mu.Unlock()
  91         n, err = r.readAt(b, pos)
  92         r.mu.Lock()
  93         r.pos += int64(n)
  94         r.mu.Unlock()
  95         r.posChanged()
  96         return
  97 }
  98
  99 // Must only return EOF at the end of the torrent.
 100 func (r *Reader) readAt(b []byte, pos int64) (n int, err error) {
 101         // defer func() {
 102         //      log.Println(pos, n, err)
 103         // }()
 104         maxLen := r.t.torrent.Info.TotalLength() - pos
 105         if maxLen <= 0 {
 106                 err = io.EOF
 107                 return
 108         }
 109         if int64(len(b)) > maxLen {
 110                 b = b[:maxLen]
 111         }
 112 again:
 113         r.t.cl.mu.Lock()
 114         for !r.readable(pos) {
 115                 r.waitReadable(pos)
 116         }
 117         avail := r.available(pos, int64(len(b)))
 118         // log.Println("available", avail)
 119         r.t.cl.mu.Unlock()
 120         b1 := b[:avail]
 121         pi := int(pos / r.t.Info().PieceLength)
 122         tp := &r.t.torrent.Pieces[pi]
 123         ip := r.t.Info().Piece(pi)
 124         po := pos % ip.Length()
 125         if int64(len(b1)) > ip.Length()-po {
 126                 b1 = b1[:ip.Length()-po]
 127         }
 128         tp.waitNoPendingWrites()
 129         n, err = dataReadAt(r.t.torrent.data, b1, pos)
 130         if n != 0 {
 131                 err = nil
 132                 return
 133         }
 134         if r.t.torrent.isClosed() {
 135                 if err == nil {
 136                         err = errors.New("torrent closed")
 137                 }
 138                 return
 139         }
 140         if err == io.ErrUnexpectedEOF {
 141                 goto again
 142         }
 143         return
 144 }
 145
 146 func (r *Reader) Close() error {
 147         r.t.deleteReader(r)
 148         r.t = nil
 149         return nil
 150 }
 151
 152 func (r *Reader) posChanged() {
 153         r.t.cl.mu.Lock()
 154         defer r.t.cl.mu.Unlock()
 155         r.t.torrent.readersChanged(r.t.cl)
 156 }
 157
 158 func (r *Reader) Seek(off int64, whence int) (ret int64, err error) {
 159         r.mu.Lock()
 160         switch whence {
 161         case os.SEEK_SET:
 162                 r.pos = off
 163         case os.SEEK_CUR:
 164                 r.pos += off
 165         case os.SEEK_END:
 166                 r.pos = r.t.torrent.Info.TotalLength() + off
 167         default:
 168                 err = errors.New("bad whence")
 169         }
 170         ret = r.pos
 171         r.mu.Unlock()
 172         r.posChanged()
 173         return
 174 }