reader.go

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