rkrtechnology/litestream
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refactor repo

Browse Source
This commit is contained in:
Ben Johnson
2020-10-19 10:37:51 -06:00
parent d001df823d
commit 8fa4727358
6 changed files with 76 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
main.go → cmd/litestream/main.go
View File

@@ -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() {

18
file_system.go Normal file
View File

@@ -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
}

58
fs.go
View File

@@ -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 {}

152
handle.go
View File

@@ -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

52
node.go
View File

@@ -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
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")
// }

2
server.go
View File

@@ -1,4 +1,4 @@
package main
package litestream
/*
import (