Separating file handles for RO and RW operations

Author: smkniazi

Dir.go

@@ -235,7 +235,7 @@ func (dir *DirINode) Create(ctx context.Context, req *fuse.CreateRequest, resp *
 
 	loginfo("Creating a new file", Fields{Operation: Create, Path: dir.AbsolutePathForChild(req.Name), Mode: req.Mode, Flags: req.Flags})
 	file := dir.NodeFromAttrs(Attrs{Name: req.Name, Mode: req.Mode}).(*FileINode)
-	handle, err := NewFileHandle(file, false, req.Flags)
+	handle, err := file.NewFileHandle(false, req.Flags)
 	if err != nil {
 		logerror("File creation failed", Fields{Operation: Create, Path: dir.AbsolutePathForChild(req.Name), Mode: req.Mode, Flags: req.Flags, Error: err})
 		//TODO remove the entry from the cache

File.go

@@ -4,15 +4,20 @@ package main
 
 import (
 	"fmt"
+	"io"
+	"io/ioutil"
+	"math/rand"
 	"os"
 	"os/user"
 	"path"
 	"sync"
+	"syscall"
 	"time"
 
 	"bazil.org/fuse"
 	"bazil.org/fuse/fs"
 	"golang.org/x/net/context"
+	"golang.org/x/sys/unix"
 )
 
 type FileINode struct {
@@ -22,7 +27,7 @@ type FileINode struct {
 
 	activeHandles []*FileHandle // list of opened file handles
 	fileMutex     sync.Mutex    // mutex for activeHandles
-	handle        *os.File      // handle to the temp file in staging dir
+	handle        FileProxy     // handle to the temp file in staging dir
 }
 
 // Verify that *File implements necesary FUSE interfaces
@@ -56,7 +61,7 @@ func (file *FileINode) Open(ctx context.Context, req *fuse.OpenRequest, resp *fu
 	defer file.fileMutex.Unlock()
 
 	logdebug("Opening file", Fields{Operation: Open, Path: file.AbsolutePath(), Flags: req.Flags})
-	handle, err := NewFileHandle(file, true, req.Flags)
+	handle, err := file.NewFileHandle(true, req.Flags)
 	if err != nil {
 		return nil, err
 	}
@@ -181,3 +186,110 @@ func (file *FileINode) Setattr(ctx context.Context, req *fuse.SetattrRequest, re
 func (file *FileINode) countActiveHandles() int {
 	return len(file.activeHandles)
 }
+
+func (file *FileINode) createStagingFile(operation string, existsInDFS bool) error {
+	if file.handle != nil {
+		return nil // there is already an active handle.
+	}
+
+	//create staging file
+	absPath := file.AbsolutePath()
+	hdfsAccessor := file.FileSystem.HdfsAccessor
+	if !existsInDFS { // it  is a new file so create it in the DFS
+		w, err := hdfsAccessor.CreateFile(absPath, file.Attrs.Mode, false)
+		if err != nil {
+			logerror("Failed to create file in DFS", file.logInfo(Fields{Operation: operation, Error: err}))
+			return err
+		}
+		loginfo("Created an empty file in DFS", file.logInfo(Fields{Operation: operation}))
+		w.Close()
+	} else {
+		// Request to write to existing file
+		_, err := hdfsAccessor.Stat(absPath)
+		if err != nil {
+			logerror("Failed to stat file in DFS", file.logInfo(Fields{Operation: operation, Error: err}))
+			return syscall.ENOENT
+		}
+	}
+
+	stagingFile, err := ioutil.TempFile(stagingDir, "stage")
+	if err != nil {
+		logerror("Failed to create staging file", file.logInfo(Fields{Operation: operation, Error: err}))
+		return err
+	}
+	os.Remove(stagingFile.Name())
+	loginfo("Created staging file", file.logInfo(Fields{Operation: operation, TmpFile: stagingFile.Name()}))
+
+	if existsInDFS {
+		if err := file.downloadToStaging(stagingFile, operation); err != nil {
+			return err
+		}
+	}
+	file.handle = &LocalFileProxy{localFile: stagingFile}
+	return nil
+}
+
+func (file *FileINode) downloadToStaging(stagingFile *os.File, operation string) error {
+	hdfsAccessor := file.FileSystem.HdfsAccessor
+	absPath := file.AbsolutePath()
+
+	reader, err := hdfsAccessor.OpenRead(absPath)
+	if err != nil {
+		logerror("Failed to open file in DFS", file.logInfo(Fields{Operation: operation, Error: err}))
+		// TODO remove the staging file if there are no more active handles
+		return err
+	}
+
+	nc, err := io.Copy(stagingFile, reader)
+	if err != nil {
+		logerror("Failed to copy content to staging file", file.logInfo(Fields{Operation: operation, Error: err}))
+		return err
+	}
+	reader.Close()
+	loginfo(fmt.Sprintf("Downloaded a copy to stating dir. %d bytes copied", nc), file.logInfo(Fields{Operation: operation}))
+	return nil
+}
+
+// Creates new file handle
+func (file *FileINode) NewFileHandle(existsInDFS bool, flags fuse.OpenFlags) (*FileHandle, error) {
+	operation := Create
+	if existsInDFS {
+		operation = Open
+	}
+
+	fh := &FileHandle{File: file, fileFlags: flags, fhID: int64(rand.Uint64())}
+	if err := file.checkDiskSpace(); err != nil {
+		return nil, err
+	}
+
+	if err := file.createStagingFile(operation, existsInDFS); err != nil {
+		return nil, err
+	}
+
+	loginfo("Opened file", fh.logInfo(Fields{Operation: operation, Flags: fh.fileFlags}))
+	return fh, nil
+}
+
+func (file *FileINode) checkDiskSpace() error {
+	var stat unix.Statfs_t
+	wd, err := os.Getwd()
+	if err != nil {
+		return err
+	}
+	unix.Statfs(wd, &stat)
+	// Available blocks * size per block = available space in bytes
+	bytesAvailable := stat.Bavail * uint64(stat.Bsize)
+	if bytesAvailable < 64*1024*1024 {
+		return syscall.ENOSPC
+	} else {
+		return nil
+	}
+}
+
+func (file *FileINode) logInfo(fields Fields) Fields {
+	f := Fields{Path: file.AbsolutePath()}
+	for k, e := range fields {
+		f[k] = e
+	}
+	return f
+}

FileProxy.go

@@ -0,0 +1,11 @@
+package main
+
+type FileProxy interface {
+	Truncate(size int64) error
+	WriteAt(b []byte, off int64) (n int, err error)
+	ReadAt(b []byte, off int64) (n int, err error)
+	Seek(offset int64, whence int) (ret int64, err error)
+	Read(b []byte) (n int, err error)
+	Close() error
+	Sync() error
+}

HopsFileHandle.go

@@ -3,18 +3,12 @@
 package main
 
 import (
-	"fmt"
 	"io"
-	"io/ioutil"
-	"math/rand"
-	"os"
 	"sync"
-	"syscall"
 
 	"bazil.org/fuse"
 	"bazil.org/fuse/fs"
 	"golang.org/x/net/context"
-	"golang.org/x/sys/unix"
 )
 
 // Represends a handle to an open file
@@ -35,90 +29,6 @@ var _ fs.HandleWriter = (*FileHandle)(nil)
 var _ fs.NodeFsyncer = (*FileHandle)(nil)
 var _ fs.HandleFlusher = (*FileHandle)(nil)
 
-func (fh *FileHandle) createStagingFile(operation string, existsInDFS bool) error {
-	if fh.File.handle != nil {
-		return nil // there is already an active handle.
-	}
-
-	//create staging file
-	absPath := fh.File.AbsolutePath()
-	hdfsAccessor := fh.File.FileSystem.HdfsAccessor
-	if !existsInDFS { // it  is a new file so create it in the DFS
-		w, err := hdfsAccessor.CreateFile(absPath, fh.File.Attrs.Mode, false)
-		if err != nil {
-			logerror("Failed to create file in DFS", fh.logInfo(Fields{Operation: operation, Error: err}))
-			return err
-		}
-		loginfo("Created an empty file in DFS", fh.logInfo(Fields{Operation: operation}))
-		w.Close()
-	} else {
-		// Request to write to existing file
-		_, err := hdfsAccessor.Stat(absPath)
-		if err != nil {
-			logerror("Failed to stat file in DFS", fh.logInfo(Fields{Operation: operation, Error: err}))
-			return syscall.ENOENT
-		}
-	}
-
-	stagingFile, err := ioutil.TempFile(stagingDir, "stage")
-	if err != nil {
-		logerror("Failed to create staging file", fh.logInfo(Fields{Operation: operation, Error: err}))
-		return err
-	}
-	os.Remove(stagingFile.Name())
-	loginfo("Created staging file", fh.logInfo(Fields{Operation: operation, TmpFile: stagingFile.Name()}))
-
-	if existsInDFS {
-		if err := fh.downloadToStaging(stagingFile, operation); err != nil {
-			return err
-		}
-	}
-	fh.File.handle = stagingFile
-	return nil
-}
-
-func (fh *FileHandle) downloadToStaging(stagingFile *os.File, operation string) error {
-	hdfsAccessor := fh.File.FileSystem.HdfsAccessor
-	absPath := fh.File.AbsolutePath()
-
-	reader, err := hdfsAccessor.OpenRead(absPath)
-	if err != nil {
-		logerror("Failed to open file in DFS", fh.logInfo(Fields{Operation: operation, Error: err}))
-		// TODO remove the staging file if there are no more active handles
-		return err
-	}
-
-	nc, err := io.Copy(stagingFile, reader)
-	if err != nil {
-		logerror("Failed to copy content to staging file", fh.logInfo(Fields{Operation: operation, Error: err}))
-		return err
-	}
-	reader.Close()
-	loginfo(fmt.Sprintf("Downloaded a copy to stating dir. %d bytes copied", nc), fh.logInfo(Fields{Operation: operation}))
-	return nil
-}
-
-// Creates new file handle
-func NewFileHandle(file *FileINode, existsInDFS bool, flags fuse.OpenFlags) (*FileHandle, error) {
-
-	operation := Create
-	if existsInDFS {
-		operation = Open
-	}
-
-	fh := &FileHandle{File: file, fileFlags: flags, fhID: int64(rand.Uint64())}
-	if err := checkDiskSpace(); err != nil {
-		return nil, err
-	}
-
-	if err := fh.createStagingFile(operation, existsInDFS); err != nil {
-		return nil, err
-	}
-
-	loginfo("Opened file", fh.logInfo(Fields{Operation: operation, Flags: fh.fileFlags}))
-	return fh, nil
-}
-
 func (fh *FileHandle) isWriteable() bool {
 	if fh.fileFlags.IsWriteOnly() || fh.fileFlags.IsReadWrite() {
 		return true
@@ -136,22 +46,6 @@ func (fh *FileHandle) Truncate(size int64) error {
 	return nil
 }
 
-func checkDiskSpace() error {
-	var stat unix.Statfs_t
-	wd, err := os.Getwd()
-	if err != nil {
-		return err
-	}
-	unix.Statfs(wd, &stat)
-	// Available blocks * size per block = available space in bytes
-	bytesAvailable := stat.Bavail * uint64(stat.Bsize)
-	if bytesAvailable < 64*1024*1024 {
-		return syscall.ENOSPC
-	} else {
-		return nil
-	}
-}
-
 // Returns attributes of the file associated with this handle
 func (fh *FileHandle) Attr(ctx context.Context, a *fuse.Attr) error {
 	fh.Mutex.Lock()

LocalFileProxy.go

@@ -0,0 +1,37 @@
+package main
+
+import "os"
+
+type LocalFileProxy struct {
+	localFile *os.File // handle to the temp file in staging dir
+}
+
+var _ FileProxy = (*LocalFileProxy)(nil)
+
+func (p *LocalFileProxy) Truncate(size int64) error {
+	return p.localFile.Truncate(size)
+}
+
+func (p *LocalFileProxy) WriteAt(b []byte, off int64) (n int, err error) {
+	return p.localFile.WriteAt(b, off)
+}
+
+func (p *LocalFileProxy) ReadAt(b []byte, off int64) (n int, err error) {
+	return p.localFile.ReadAt(b, off)
+}
+
+func (p *LocalFileProxy) Seek(offset int64, whence int) (ret int64, err error) {
+	return p.localFile.Seek(offset, whence)
+}
+
+func (p *LocalFileProxy) Read(b []byte) (n int, err error) {
+	return p.localFile.Read(b)
+}
+
+func (p *LocalFileProxy) Close() error {
+	return p.localFile.Close()
+}
+
+func (p *LocalFileProxy) Sync() error {
+	return p.localFile.Sync()
+}

RemoteFileProxy.go

@@ -0,0 +1,35 @@
+package main
+
+type RemoteFileProxy struct {
+}
+
+var _ FileProxy = (*RemoteFileProxy)(nil)
+
+func (p *RemoteFileProxy) Truncate(size int64) error {
+	logfatal("Not implemented yet", nil)
+	return nil
+}
+func (p *RemoteFileProxy) WriteAt(b []byte, off int64) (n int, err error) {
+	logfatal("Not implemented yet", nil)
+	return 0, nil
+}
+func (p *RemoteFileProxy) ReadAt(b []byte, off int64) (n int, err error) {
+	logfatal("Not implemented yet", nil)
+	return 0, nil
+}
+func (p *RemoteFileProxy) Seek(offset int64, whence int) (ret int64, err error) {
+	logfatal("Not implemented yet", nil)
+	return 0, nil
+}
+func (p *RemoteFileProxy) Read(b []byte) (n int, err error) {
+	logfatal("Not implemented yet", nil)
+	return 0, nil
+}
+func (p *RemoteFileProxy) Close() error {
+	logfatal("Not implemented yet", nil)
+	return nil
+}
+func (p *RemoteFileProxy) Sync() error {
+	logfatal("Not implemented yet", nil)
+	return nil
+}