fs/file_handle.go

   1 package torrentfs
   2
   3 import (
   4         "context"
   5         "io"
   6         "log"
   7
   8         "bazil.org/fuse"
   9         "bazil.org/fuse/fs"
  10         "github.com/anacrolix/missinggo"
  11
  12         "github.com/anacrolix/torrent"
  13 )
  14
  15 type fileHandle struct {
  16         fn fileNode
  17         r  torrent.Reader
  18 }
  19
  20 var _ interface {
  21         fs.HandleReader
  22         fs.HandleReleaser
  23 } = fileHandle{}
  24
  25 func (me fileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) error {
  26         torrentfsReadRequests.Add(1)
  27         if req.Dir {
  28                 panic("read on directory")
  29         }
  30         r := me.r
  31         pos, err := r.Seek(req.Offset, io.SeekStart)
  32         if err != nil {
  33                 panic(err)
  34         }
  35         if pos != req.Offset {
  36                 panic("seek failed")
  37         }
  38         resp.Data = resp.Data[:req.Size]
  39         readDone := make(chan struct{})
  40         ctx, cancel := context.WithCancel(ctx)
  41         var readErr error
  42         go func() {
  43                 defer close(readDone)
  44                 me.fn.FS.mu.Lock()
  45                 me.fn.FS.blockedReads++
  46                 me.fn.FS.event.Broadcast()
  47                 me.fn.FS.mu.Unlock()
  48                 var n int
  49                 r := missinggo.ContextedReader{r, ctx}
  50                 log.Printf("reading %v bytes at %v", len(resp.Data), req.Offset)
  51                 if true {
  52                         n, readErr = io.ReadFull(r, resp.Data)
  53                 } else {
  54                         n, readErr = r.Read(resp.Data)
  55                         if readErr == io.EOF {
  56                                 readErr = nil
  57                         }
  58                 }
  59                 resp.Data = resp.Data[:n]
  60         }()
  61         defer func() {
  62                 <-readDone
  63                 me.fn.FS.mu.Lock()
  64                 me.fn.FS.blockedReads--
  65                 me.fn.FS.event.Broadcast()
  66                 me.fn.FS.mu.Unlock()
  67         }()
  68         defer cancel()
  69
  70         select {
  71         case <-readDone:
  72                 return readErr
  73         case <-me.fn.FS.destroyed:
  74                 return fuse.EIO
  75         case <-ctx.Done():
  76                 return fuse.EINTR
  77         }
  78 }
  79
  80 func (me fileHandle) Release(context.Context, *fuse.ReleaseRequest) error {
  81         return me.r.Close()
  82 }