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.pendingWritesMutex.Lock()
 122         for tp.pendingWrites != 0 {
 123                 tp.noPendingWrites.Wait()
 124         }
 125         tp.pendingWritesMutex.Unlock()
 126         n, err = dataReadAt(r.t.torrent.data, b1, pos)
 127         if n != 0 {
 128                 err = nil
 129                 return
 130         }
 131         if r.t.torrent.isClosed() {
 132                 if err == nil {
 133                         err = errors.New("torrent closed")
 134                 }
 135                 return
 136         }
 137         if err == io.ErrUnexpectedEOF {
 138                 goto again
 139         }
 140         return
 141 }
 142
 143 func (r *Reader) Close() error {
 144         r.t.deleteReader(r)
 145         r.t = nil
 146         return nil
 147 }
 148
 149 func (r *Reader) posChanged() {
 150         r.t.cl.mu.Lock()
 151         defer r.t.cl.mu.Unlock()
 152         r.t.torrent.readersChanged(r.t.cl)
 153 }
 154
 155 func (r *Reader) Seek(off int64, whence int) (ret int64, err error) {
 156         r.mu.Lock()
 157         switch whence {
 158         case os.SEEK_SET:
 159                 r.pos = off
 160         case os.SEEK_CUR:
 161                 r.pos += off
 162         case os.SEEK_END:
 163                 r.pos = r.t.torrent.Info.TotalLength() + off
 164         default:
 165                 err = errors.New("bad whence")
 166         }
 167         ret = r.pos
 168         r.mu.Unlock()
 169         r.posChanged()
 170         return
 171 }