From 8fa4727358f3e8d428e68fbec79816df6e903526 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Mon, 19 Oct 2020 10:37:51 -0600 Subject: [PATCH] Refactor repo --- main.go => cmd/litestream/main.go | 11 ++- file_system.go | 18 ++++ fs.go | 58 ------------ handle.go | 152 +++++++++--------------------- node.go | 54 ++--------- server.go | 2 +- 6 files changed, 76 insertions(+), 219 deletions(-) rename main.go => cmd/litestream/main.go (89%) create mode 100644 file_system.go delete mode 100644 fs.go diff --git a/main.go b/cmd/litestream/main.go similarity index 89% rename from main.go rename to cmd/litestream/main.go index bbb9a80..0c58d3a 100644 --- a/main.go +++ b/cmd/litestream/main.go @@ -11,6 +11,7 @@ import ( "bazil.org/fuse" "bazil.org/fuse/fs" + "github.com/middlemost/litestream" ) func main() { @@ -43,6 +44,8 @@ func (m *Main) Run(args []string) (err error) { if err := flagSet.Parse(args); err != nil { return err } + + // Ensure src & mount paths are specified. if m.SourcePath = flagSet.Arg(0); m.SourcePath == "" { return errors.New("source path required") } else if m.MountPath = flagSet.Arg(1); m.MountPath == "" { @@ -50,7 +53,9 @@ func (m *Main) Run(args []string) (err error) { } // Setup logging, if verbose specified. + var config fs.Config if *verbose { + config.Debug = debug m.logger = log.New(os.Stderr, "", log.LstdFlags) } @@ -64,10 +69,8 @@ func (m *Main) Run(args []string) (err error) { m.logger.Printf("mounted") - s := fs.New(conn, &fs.Config{ - Debug: debug, - }) - return s.Serve(&FS{SourcePath: m.SourcePath}) + s := fs.New(conn, &config) + return s.Serve(&litestream.FileSystem{SourcePath: m.SourcePath}) } func (m *Main) usage() { diff --git a/file_system.go b/file_system.go new file mode 100644 index 0000000..c0f5edd --- /dev/null +++ b/file_system.go @@ -0,0 +1,18 @@ +package litestream + +import ( + "bazil.org/fuse/fs" +) + +var _ fs.FS = (*FileSystem)(nil) + +// FileSystem represents the file system that is mounted. +// It returns a root node that represents the root directory. +type FileSystem struct { + SourcePath string +} + +// Root returns the file system root node. +func (f *FileSystem) Root() (fs.Node, error) { + return &Node{fs: f}, nil +} diff --git a/fs.go b/fs.go deleted file mode 100644 index 31edebf..0000000 --- a/fs.go +++ /dev/null @@ -1,58 +0,0 @@ -package main - -import ( - "context" - - "bazil.org/fuse" - "bazil.org/fuse/fs" -) - -var _ fs.FS = (*FS)(nil) -var _ fs.FSDestroyer = (*FS)(nil) -var _ fs.FSStatfser = (*FS)(nil) - -// var _ fs.FSInodeGenerator = (*FS)(nil) - -type FS struct { - SourcePath string -} - -// Root returns the file system root. -func (f *FS) Root() (fs.Node, error) { - return &Node{fs: f}, nil -} - -// Destroy is called when the file system is shutting down. -// -// Linux only sends this request for block device backed (fuseblk) -// filesystems, to allow them to flush writes to disk before the -// unmount completes. -func (f *FS) Destroy() { - // TODO: Flush writes? -} - -// Statfs is called to obtain file system metadata. -// It should write that data to resp. -func (f *FS) Statfs(ctx context.Context, req *fuse.StatfsRequest, resp *fuse.StatfsResponse) error { - panic("TODO") -} - -// GenerateInode is called to pick a dynamic inode number when it -// would otherwise be 0. -// -// Not all filesystems bother tracking inodes, but FUSE requires -// the inode to be set, and fewer duplicates in general makes UNIX -// tools work better. -// -// Operations where the nodes may return 0 inodes include Getattr, -// Setattr and ReadDir. -// -// If FS does not implement FSInodeGenerator, GenerateDynamicInode -// is used. -// -// Implementing this is useful to e.g. constrain the range of -// inode values used for dynamic inodes. -// -// Non-zero return values should be greater than 1, as that is -// always used for the root inode. -// func (f *FS) GenerateInode(parentInode uint64, name string) uint64 {} diff --git a/handle.go b/handle.go index af56579..6cf684c 100644 --- a/handle.go +++ b/handle.go @@ -1,4 +1,4 @@ -package main +package litestream import ( "context" @@ -11,11 +11,16 @@ import ( "bazil.org/fuse/fs" ) +const ( + F_OFD_GETLK = 0x24 + F_OFD_SETLK = 0x25 + F_OFD_SETLKW = 0x26 +) + var _ fs.HandleFlockLocker = (*Handle)(nil) var _ fs.HandleFlusher = (*Handle)(nil) var _ fs.HandleLocker = (*Handle)(nil) var _ fs.HandlePOSIXLocker = (*Handle)(nil) -var _ fs.HandlePoller = (*Handle)(nil) var _ fs.HandleReadDirAller = (*Handle)(nil) var _ fs.HandleReader = (*Handle)(nil) var _ fs.HandleReleaser = (*Handle)(nil) @@ -23,17 +28,39 @@ var _ fs.HandleWriter = (*Handle)(nil) // var _ fs.HandleReadAller = (*Handle)(nil) +// Handle represents a FUSE file handle. type Handle struct { f *os.File } -// Flush is called each time the file or directory is closed. -// Because there can be multiple file descriptors referring to a -// single opened file, Flush can be called multiple times. +// Release closes the underlying file descriptor. +func (h *Handle) Release(ctx context.Context, req *fuse.ReleaseRequest) (err error) { + return h.f.Close() +} + +// Read reads data from a given offset in the underlying file. +func (h *Handle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) (err error) { + buf := make([]byte, req.Size) + n, err := h.f.ReadAt(buf, req.Offset) + if err != nil && err != io.EOF { + return err + } + resp.Data = buf[:n] + return nil +} + +// Write writes data at a given offset to the underlying file. +func (h *Handle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) (err error) { + resp.Size, err = h.f.WriteAt(req.Data, req.Offset) + return err +} + +// Flush is called when a file handle is synced to disk. Implements fs.HandleFlusher. func (h *Handle) Flush(ctx context.Context, req *fuse.FlushRequest) (err error) { return h.f.Sync() } +// ReadDirAll returns a list of all entries in a directory. Implements fs.HandleReadDirAller. func (h *Handle) ReadDirAll(ctx context.Context) (ents []fuse.Dirent, err error) { fis, err := h.f.Readdir(-1) if err != nil { @@ -51,124 +78,33 @@ func (h *Handle) ReadDirAll(ctx context.Context) (ents []fuse.Dirent, err error) return ents, nil } -// Read requests to read data from the handle. -// -// There is a page cache in the kernel that normally submits only -// page-aligned reads spanning one or more pages. However, you -// should not rely on this. To see individual requests as -// submitted by the file system clients, set OpenDirectIO. -// -// Note that reads beyond the size of the file as reported by Attr -// are not even attempted (except in OpenDirectIO mode). -func (h *Handle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) (err error) { - // TODO: Flags ReadFlags - // TODO: LockOwner - // TODO: FileFlags OpenFlags - - buf := make([]byte, req.Size) - n, err := h.f.ReadAt(buf, req.Offset) - if err != nil && err != io.EOF { - return err - } - resp.Data = buf[:n] - - return nil -} - -// Write requests to write data into the handle at the given offset. -// Store the amount of data written in resp.Size. -// -// There is a writeback page cache in the kernel that normally submits -// only page-aligned writes spanning one or more pages. However, -// you should not rely on this. To see individual requests as -// submitted by the file system clients, set OpenDirectIO. -// -// Writes that grow the file are expected to update the file size -// (as seen through Attr). Note that file size changes are -// communicated also through Setattr. -func (h *Handle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) (err error) { - // Offset int64 - // Data []byte - // Flags WriteFlags - // LockOwner LockOwner - // FileFlags OpenFlags - - resp.Size, err = h.f.WriteAt(req.Data, req.Offset) - println("dbg/write!") - return err -} - -func (h *Handle) Release(ctx context.Context, req *fuse.ReleaseRequest) (err error) { - return h.f.Close() -} - -// Poll checks whether the handle is currently ready for I/O, and -// may request a wakeup when it is. -// -// Poll should always return quickly. Clients waiting for -// readiness can be woken up by passing the return value of -// PollRequest.Wakeup to fs.Server.NotifyPollWakeup or -// fuse.Conn.NotifyPollWakeup. -// -// To allow supporting poll for only some of your Nodes/Handles, -// the default behavior is to report immediate readiness. If your -// FS does not support polling and you want to minimize needless -// requests and log noise, implement NodePoller and return -// syscall.ENOSYS. -// -// The Go runtime uses epoll-based I/O whenever possible, even for -// regular files. -func (h *Handle) Poll(ctx context.Context, req *fuse.PollRequest, resp *fuse.PollResponse) error { - //type PollRequest struct { - // Header `json:"-"` - // Handle HandleID - // - // Flags PollFlags - // Events PollEvents - //} - - //type PollResponse struct { - // REvents PollEvents - //} - - panic("TODO") -} - // Lock tries to acquire a lock on a byte range of the node. If a // conflicting lock is already held, returns syscall.EAGAIN. -// -// LockRequest.LockOwner is a file-unique identifier for this -// lock, and will be seen in calls releasing this lock -// (UnlockRequest, ReleaseRequest, FlushRequest) and also -// in e.g. ReadRequest, WriteRequest. func (h *Handle) Lock(ctx context.Context, req *fuse.LockRequest) error { - // type LockRequest struct { - // Header - // Handle HandleID - // // LockOwner is a unique identifier for the originating client, to identify locks. - // LockOwner LockOwner - // Lock FileLock - // LockFlags LockFlags - // } - - panic("TODO") + return h.lock(ctx, req) } // LockWait acquires a lock on a byte range of the node, waiting // until the lock can be obtained (or context is canceled). func (h *Handle) LockWait(ctx context.Context, req *fuse.LockWaitRequest) error { - // type LockWaitRequest LockRequest - - panic("TODO") + return h.lock(ctx, (*fuse.LockRequest)(req)) } // Unlock releases the lock on a byte range of the node. Locks can // be released also implicitly, see HandleFlockLocker and // HandlePOSIXLocker. func (h *Handle) Unlock(ctx context.Context, req *fuse.UnlockRequest) error { - // type UnlockRequest LockRequest + return h.lock(ctx, (*fuse.LockRequest)(req)) - panic("TODO") +} + +func (h *Handle) lock(ctx context.Context, req *fuse.LockRequest) error { + return syscall.FcntlFlock(h.f.Fd(), F_OFD_SETLK, &syscall.Flock_t{ + Type: int16(req.Lock.Type), + Whence: io.SeekStart, + Start: int64(req.Lock.Start), + Len: int64(req.Lock.End) - int64(req.Lock.Start), + }) } // QueryLock returns the current state of locks held for the byte diff --git a/node.go b/node.go index d5c08e2..1c03a52 100644 --- a/node.go +++ b/node.go @@ -1,4 +1,4 @@ -package main +package litestream import ( "context" @@ -31,16 +31,13 @@ var _ fs.NodeSetxattrer = (*Node)(nil) var _ fs.NodeStringLookuper = (*Node)(nil) var _ fs.NodeSymlinker = (*Node)(nil) -// var _ fs.NodeRequestLookuper = (*Node)(nil) -// var _ fs.NodeForgetter = (*Node)(nil) -// var _ fs.NodePoller = (*Node)(nil) - +// Node represents a file or directory in the file system. type Node struct { - fs *FS // base filesystem - path string // path within file system + fs *FileSystem // base filesystem + path string // path within file system } -func NewNode(fs *FS, path string) *Node { +func NewNode(fs *FileSystem, path string) *Node { return &Node{fs: fs, path: path} } @@ -106,15 +103,6 @@ func (n *Node) ReadDirAll(ctx context.Context) (ents []fuse.Dirent, err error) { return ents, nil } -// Getattr obtains the standard metadata for the receiver. -// It should store that metadata in resp. -// -// If this method is not implemented, the attributes will be -// generated based on Attr(), with zero values filled in. -// func (n *Node) Getattr(ctx context.Context, req *fuse.GetattrRequest, resp *fuse.GetattrResponse) error { -// panic("TODO") -// } - // Setattr sets the standard metadata for the receiver. // // Note, this is also used to communicate changes in the size of @@ -249,12 +237,6 @@ func (n *Node) Access(ctx context.Context, req *fuse.AccessRequest) (err error) return syscall.Access(n.srcpath(), req.Mask) } -//type NodeRequestLookuper interface { -// // Lookup looks up a specific entry in the receiver. -// // See NodeStringLookuper for more. -// Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.LookupResponse) (Node, error) -//} - func (n *Node) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (_ fs.Node, err error) { if err := syscall.Mkdir(filepath.Join(n.srcpath(), req.Name), uint32(req.Mode^req.Umask)); err != nil { return nil, err @@ -264,14 +246,6 @@ func (n *Node) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (_ fs.Node, er // Open opens the receiver. After a successful open, a client // process has a file descriptor referring to this Handle. -// -// Open can also be also called on non-files. For example, -// directories are Opened for ReadDir or fchdir(2). -// -// If this method is not implemented, the open will always -// succeed, and the Node itself will be used as the Handle. -// -// XXX note about access. XXX OpenFlags. func (n *Node) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenResponse) (_ fs.Handle, err error) { // TODO(bbj): Where does mode come from? f, err := os.OpenFile(n.srcpath(), int(req.Flags), 0777) @@ -290,13 +264,6 @@ func (n *Node) Create(ctx context.Context, req *fuse.CreateRequest, resp *fuse.C return NewNode(n.fs, filepath.Join(n.path, req.Name)), &Handle{f: f}, nil } -// Forget about this node. This node will not receive further -// method calls. -// -// Forget is not necessarily seen on unmount, as all nodes are -// implicitly forgotten as part of the unmount. -// func (n *Node) Forget() { panic("TODO") } - func (n *Node) Rename(ctx context.Context, req *fuse.RenameRequest, _newDir fs.Node) (err error) { newDir := _newDir.(*Node) return os.Rename(filepath.Join(n.srcpath(), req.OldName), filepath.Join(newDir.srcpath(), req.NewName)) @@ -320,10 +287,7 @@ func (n *Node) Fsync(ctx context.Context, req *fuse.FsyncRequest) (err error) { return f.Sync() } -// Getxattr gets an extended attribute by the given name from the -// node. -// -// If there is no xattr by that name, returns fuse.ErrNoXattr. +// Getxattr gets an extended attribute by the given name from the node. func (n *Node) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *fuse.GetxattrResponse) (err error) { // TODO(bbj): Handle req.Size & returned syscall.Getxattr() size. if _, err = syscall.Getxattr(n.srcpath(), req.Name, resp.Xattr); err == syscall.ENODATA { @@ -351,9 +315,3 @@ func (n *Node) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) (err err func (n *Node) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) (err error) { return syscall.Removexattr(n.srcpath(), req.Name) } - -// Poll checks whether the node is currently ready for I/O, and -// may request a wakeup when it is. See HandlePoller. -// func (n *Node) Poll(ctx context.Context, req *fuse.PollRequest, resp *fuse.PollResponse) error { -// panic("TODO") -// } diff --git a/server.go b/server.go index 23bcd5e..4046b45 100644 --- a/server.go +++ b/server.go @@ -1,4 +1,4 @@ -package main +package litestream /* import (