fs/file_handle.go

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