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) waitReadable(off int64) {
  77         r.t.cl.event.Wait()
  78 }
  79
  80 func (r *Reader) Read(b []byte) (n int, err error) {
  81         r.mu.Lock()
  82         pos := r.pos
  83         r.mu.Unlock()
  84         n, err = r.readAt(b, pos)
  85         r.mu.Lock()
  86         r.pos += int64(n)
  87         r.mu.Unlock()
  88         r.posChanged()
  89         return
  90 }
  91
  92 // Must only return EOF at the end of the torrent.
  93 func (r *Reader) readAt(b []byte, pos int64) (n int, err error) {
  94         // defer func() {
  95         //      log.Println(pos, n, err)
  96         // }()
  97         maxLen := r.t.torrent.Info.TotalLength() - pos
  98         if maxLen <= 0 {
  99                 err = io.EOF
 100                 return
 101         }
 102         if int64(len(b)) > maxLen {
 103                 b = b[:maxLen]
 104         }
 105 again:
 106         r.t.cl.mu.Lock()
 107         for !r.readable(pos) {
 108                 r.waitReadable(pos)
 109         }
 110         avail := r.available(pos, int64(len(b)))
 111         // log.Println("available", avail)
 112         r.t.cl.mu.Unlock()
 113         b1 := b[:avail]
 114         pi := int(pos / r.t.Info().PieceLength)
 115         tp := &r.t.torrent.Pieces[pi]
 116         ip := r.t.Info().Piece(pi)
 117         po := pos % ip.Length()
 118         if int64(len(b1)) > ip.Length()-po {
 119                 b1 = b1[:ip.Length()-po]
 120         }
 121         tp.waitNoPendingWrites()
 122         n, err = dataReadAt(r.t.torrent.data, b1, pos)
 123         if n != 0 {
 124                 err = nil
 125                 return
 126         }
 127         if r.t.torrent.isClosed() {
 128                 if err == nil {
 129                         err = errors.New("torrent closed")
 130                 }
 131                 return
 132         }
 133         if err == io.ErrUnexpectedEOF {
 134                 goto again
 135         }
 136         return
 137 }
 138
 139 func (r *Reader) Close() error {
 140         r.t.deleteReader(r)
 141         r.t = nil
 142         return nil
 143 }
 144
 145 func (r *Reader) posChanged() {
 146         r.t.cl.mu.Lock()
 147         defer r.t.cl.mu.Unlock()
 148         r.t.torrent.readersChanged(r.t.cl)
 149 }
 150
 151 func (r *Reader) Seek(off int64, whence int) (ret int64, err error) {
 152         r.mu.Lock()
 153         switch whence {
 154         case os.SEEK_SET:
 155                 r.pos = off
 156         case os.SEEK_CUR:
 157                 r.pos += off
 158         case os.SEEK_END:
 159                 r.pos = r.t.torrent.Info.TotalLength() + off
 160         default:
 161                 err = errors.New("bad whence")
 162         }
 163         ret = r.pos
 164         r.mu.Unlock()
 165         r.posChanged()
 166         return
 167 }