Tests passing but the performance degraded further

Author: smkniazi

File.go

@@ -47,6 +47,8 @@ func (file *FileINode) AbsolutePath() string {
 
 // Responds to the FUSE file attribute request
 func (file *FileINode) Attr(ctx context.Context, a *fuse.Attr) error {
+	file.fileMutex.Lock()
+	defer file.fileMutex.Unlock()
 	if file.FileSystem.Clock.Now().After(file.Attrs.Expires) {
 		err := file.Parent.LookupAttrs(file.Attrs.Name, &file.Attrs)
 		if err != nil {
@@ -100,6 +102,8 @@ func (file *FileINode) RemoveHandle(handle *FileHandle) {
 // Responds to the FUSE Fsync request
 func (file *FileINode) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
 	loginfo(fmt.Sprintf("Dispatching fsync request to all open handles: %d", len(file.activeHandles)), Fields{Operation: Fsync})
+	file.fileMutex.Lock()
+	defer file.fileMutex.Unlock()
 	var retErr error
 	for _, handle := range file.activeHandles {
 		err := handle.Fsync(ctx, req)
@@ -123,7 +127,7 @@ func (file *FileINode) Setattr(ctx context.Context, req *fuse.SetattrRequest, re
 	if req.Valid.Size() {
 		var retErr error
 		for _, handle := range file.activeHandles {
-			if handle.isWriteable() { // to only write enabled handles
+			if handle.dataChanged() { // to only write enabled handles
 				err := handle.Truncate(int64(req.Size))
 				if err != nil {
 					retErr = err
@@ -188,9 +192,9 @@ func (file *FileINode) countActiveHandles() int {
 	return len(file.activeHandles)
 }
 
-func (file *FileINode) createStagingFile(operation string, existsInDFS bool) error {
+func (file *FileINode) createStagingFile(operation string, existsInDFS bool) (*os.File, error) {
 	if file.handle != nil {
-		return nil // there is already an active handle.
+		return nil, nil // there is already an active handle.
 	}
 
 	//create staging file
@@ -200,7 +204,7 @@ func (file *FileINode) createStagingFile(operation string, existsInDFS bool) err
 		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
+			return nil, err
 		}
 		loginfo("Created an empty file in DFS", file.logInfo(Fields{Operation: operation}))
 		w.Close()
@@ -209,25 +213,24 @@ func (file *FileINode) createStagingFile(operation string, existsInDFS bool) err
 		_, err := hdfsAccessor.Stat(absPath)
 		if err != nil {
 			logerror("Failed to stat file in DFS", file.logInfo(Fields{Operation: operation, Error: err}))
-			return syscall.ENOENT
+			return nil, 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
+		return nil, 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
+			return nil, err
 		}
 	}
-	file.handle = &LocalFileProxy{localFile: stagingFile, file: file}
-	return nil
+	return stagingFile, nil
 }
 
 func (file *FileINode) downloadToStaging(stagingFile *os.File, operation string) error {
@@ -256,22 +259,77 @@ func (file *FileINode) NewFileHandle(existsInDFS bool, flags fuse.OpenFlags) (*F
 	file.lockFileHandle()
 	defer file.unLockFileHandle()
 
+	fh := &FileHandle{File: file, fileFlags: flags, fhID: int64(rand.Uint64())}
 	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 operation == Create {
+		// there must be no existing file handles for create operation
+		if file.handle != nil {
+			logpanic("Unexpected file state during creation", file.logInfo(Fields{Flags: flags}))
+		}
+		if err := file.checkDiskSpace(); err != nil {
+			return nil, err
+		}
+		stagingFile, err := file.createStagingFile(operation, existsInDFS)
+		if err != nil {
+			return nil, err
+		}
+		fh.File.handle = &LocalRWFileProxy{localFile: stagingFile, file: file}
+		loginfo("Opened file, RW handle", fh.logInfo(Fields{Operation: operation, Flags: fh.fileFlags}))
+	} else {
+		if file.handle != nil {
+			fh.File.handle = file.handle
+			loginfo("Opened file, Returning existing handle", fh.logInfo(Fields{Operation: operation, Flags: fh.fileFlags}))
+		} else {
+			// we alway open the file in RO mode. when the client writes to the file
+			// then we upgrade the handle. However, if the file is already opened in
+			// in RW state then we use the existing RW handle
+			// if file.handle
+			reader, _ := file.FileSystem.HdfsAccessor.OpenRead(file.AbsolutePath())
+			fh.File.handle = &RemoteROFileProxy{hdfsReader: reader, file: file}
+			loginfo("Opened file, RO handle", fh.logInfo(Fields{Operation: operation, Flags: fh.fileFlags}))
+		}
 	}
+	return fh, nil
+}
 
-	if err := file.createStagingFile(operation, existsInDFS); err != nil {
-		return nil, err
+// changes RO file handle to RW
+func (file *FileINode) upgradeHandleForWriting() error {
+	file.lockFileHandle()
+	defer file.unLockFileHandle()
+
+	var upgrade = false
+	if _, ok := file.handle.(*LocalRWFileProxy); ok {
+		upgrade = false
+	} else if _, ok := file.handle.(*RemoteROFileProxy); ok {
+		upgrade = true
+	} else {
+		logpanic("Unrecognized remote file proxy", nil)
 	}
 
-	loginfo("Opened file", fh.logInfo(Fields{Operation: operation, Flags: fh.fileFlags}))
-	return fh, nil
+	if !upgrade {
+		return nil
+	} else {
+		remoteROFileProxy, _ := file.handle.(*RemoteROFileProxy)
+		remoteROFileProxy.hdfsReader.Close() // close this read only handle
+		file.handle = nil
+
+		if err := file.checkDiskSpace(); err != nil {
+			return err
+		}
+
+		stagingFile, err := file.createStagingFile("Open", true)
+		if err != nil {
+			return err
+		}
+
+		file.handle = &LocalRWFileProxy{localFile: stagingFile, file: file}
+		loginfo("Open handle upgrade to support RW ", file.logInfo(Fields{Operation: "Open"}))
+		return nil
+	}
 }
 
 func (file *FileINode) checkDiskSpace() error {

FileHandleWriter_test.go

@@ -91,7 +91,7 @@ func TestFaultTolerantWriteFile(t *testing.T) {
 	assert.Equal(t, writeHandle.totalBytesWritten, int64(11))
 
 	binaryData := make([]byte, 65536)
-	writeHandle.File.handle.Seek(0, 0)
+	writeHandle.File.handle.SeekToStart()
 	nr, _ := writeHandle.File.handle.Read(binaryData)
 	binaryData = binaryData[:nr]
 

FileProxy.go

@@ -4,7 +4,7 @@ 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)
+	SeekToStart() (err error)
 	Read(b []byte) (n int, err error)
 	Close() error
 	Sync() error

FileSystemOperations_test.go

@@ -20,13 +20,33 @@ func TestSimple(t *testing.T) {
 	withMount(t, "/", func(mountPoint string, hdfsAccessor HdfsAccessor) {
 		//create a file, make sure that use and group information is correct
 		testFile := filepath.Join(mountPoint, "somefile")
+		os.Remove(testFile)
+
 		loginfo(fmt.Sprintf("New file: %s", testFile), nil)
 		createFile(t, testFile, "some data")
 		fi, _ := os.Stat(testFile)
 		fstat := fi.Sys().(*syscall.Stat_t)
 		grupInfo, _ := user.LookupGroupId(fmt.Sprintf("%d", fstat.Gid))
 		userInfo, _ := user.LookupId(fmt.Sprintf("%d", fstat.Uid))
-		loginfo(fmt.Sprintf("New file: %s, User %s, Gropu %s", testFile, userInfo.Name, grupInfo.Name), nil)
+		loginfo(fmt.Sprintf("---> New file: %s, User %s, Gropu %s", testFile, userInfo.Name, grupInfo.Name), nil)
+
+		loginfo("---> Reopening the file to write some more data", nil)
+		// append some more data
+		c, err := os.OpenFile(testFile, os.O_APPEND, 0600)
+		if err != nil {
+			t.Errorf("Reopening the file failed. File: %s. Error: %v", testFile, err)
+		}
+		c.WriteString("some more data")
+		c.Close()
+
+		loginfo("---> Reopening the file to read all the data", nil)
+		// read all the data again
+		c, _ = os.OpenFile(testFile, os.O_RDWR, 0600)
+		buffer := make([]byte, 1024)
+		c.Read(buffer)
+		c.Close()
+		logdebug(fmt.Sprintf("Data Read. %s", buffer), nil)
+
 		os.Remove(testFile)
 	})
 }

HopsFileHandle.go

@@ -29,8 +29,8 @@ var _ fs.HandleWriter = (*FileHandle)(nil)
 var _ fs.NodeFsyncer = (*FileHandle)(nil)
 var _ fs.HandleFlusher = (*FileHandle)(nil)
 
-func (fh *FileHandle) isWriteable() bool {
-	if fh.fileFlags.IsWriteOnly() || fh.fileFlags.IsReadWrite() {
+func (fh *FileHandle) dataChanged() bool {
+	if fh.totalBytesWritten > 0 {
 		return true
 	} else {
 		return false
@@ -66,14 +66,13 @@ func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fus
 	if err != nil {
 		if err == io.EOF {
 			// EOF isn't a error, reporting successful read to FUSE
-			logdebug("Finished reading from staging file. EOF", fh.logInfo(Fields{Operation: Read, Bytes: nr}))
+			logdebug("Completed reading", fh.logInfo(Fields{Operation: Read, Error: err, Bytes: nr}))
 			return nil
 		} else {
-			logerror("Failed to read from staging file", fh.logInfo(Fields{Operation: Read, Error: err, Bytes: nr}))
+			logerror("Failed to read", fh.logInfo(Fields{Operation: Read, Error: err, Bytes: nr}))
 			return err
 		}
 	}
-	logdebug("Read from staging file", fh.logInfo(Fields{Operation: Read, Bytes: nr, ReqOffset: req.Offset}))
 	return err
 }
 
@@ -82,6 +81,9 @@ func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *f
 	fh.Mutex.Lock()
 	defer fh.Mutex.Unlock()
 
+	// as an optimization the file is initially opened in readonly mode
+	fh.File.upgradeHandleForWriting()
+
 	nw, err := fh.File.handle.WriteAt(req.Data, req.Offset)
 	resp.Size = nw
 	fh.totalBytesWritten += int64(nw)
@@ -123,9 +125,9 @@ func (fh *FileHandle) FlushAttempt(operation string) error {
 	}
 
 	//open the file for reading and upload to DFS
-	offset, err := fh.File.handle.Seek(0, 0)
-	if err != nil || offset != 0 {
-		logerror("Unable to seek to the begenning of the temp file", fh.logInfo(Fields{Operation: operation, Offset: offset, Error: err}))
+	err = fh.File.handle.SeekToStart()
+	if err != nil {
+		logerror("Unable to seek to the begenning of the temp file", fh.logInfo(Fields{Operation: operation, Error: err}))
 		return err
 	}
 
@@ -164,7 +166,7 @@ func (fh *FileHandle) FlushAttempt(operation string) error {
 func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
 	fh.Mutex.Lock()
 	defer fh.Mutex.Unlock()
-	if fh.isWriteable() {
+	if fh.dataChanged() {
 		loginfo("Flush file", fh.logInfo(Fields{Operation: Flush}))
 		return fh.copyToDFS(Flush)
 	} else {
@@ -176,7 +178,7 @@ func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
 func (fh *FileHandle) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
 	fh.Mutex.Lock()
 	defer fh.Mutex.Unlock()
-	if fh.isWriteable() {
+	if fh.dataChanged() {
 		loginfo("Fsync file", fh.logInfo(Fields{Operation: Fsync}))
 		return fh.copyToDFS(Fsync)
 	} else {

LocalFileProxy.go

@@ -2,50 +2,53 @@ package main
 
 import "os"
 
-type LocalFileProxy struct {
+type LocalRWFileProxy struct {
 	localFile *os.File // handle to the temp file in staging dir
 	file      *FileINode
 }
 
-var _ FileProxy = (*LocalFileProxy)(nil)
+var _ FileProxy = (*LocalRWFileProxy)(nil)
 
-func (p *LocalFileProxy) Truncate(size int64) error {
+func (p *LocalRWFileProxy) Truncate(size int64) error {
 	p.file.lockFileHandle()
 	defer p.file.unLockFileHandle()
 	return p.localFile.Truncate(size)
 }
 
-func (p *LocalFileProxy) WriteAt(b []byte, off int64) (n int, err error) {
+func (p *LocalRWFileProxy) WriteAt(b []byte, off int64) (n int, err error) {
 	p.file.lockFileHandle()
 	defer p.file.unLockFileHandle()
 	return p.localFile.WriteAt(b, off)
 }
 
-func (p *LocalFileProxy) ReadAt(b []byte, off int64) (n int, err error) {
+func (p *LocalRWFileProxy) ReadAt(b []byte, off int64) (n int, err error) {
 	p.file.lockFileHandle()
 	defer p.file.unLockFileHandle()
-	return p.localFile.ReadAt(b, off)
+	n, err = p.localFile.ReadAt(b, off)
+	logdebug("LocalFileProxy ReadAt", p.file.logInfo(Fields{Operation: Read, Bytes: n, Error: err, Offset: off}))
+	return
 }
 
-func (p *LocalFileProxy) Seek(offset int64, whence int) (ret int64, err error) {
+func (p *LocalRWFileProxy) SeekToStart() (err error) {
 	p.file.lockFileHandle()
 	defer p.file.unLockFileHandle()
-	return p.localFile.Seek(offset, whence)
+	_, err = p.localFile.Seek(0, 0)
+	return
 }
 
-func (p *LocalFileProxy) Read(b []byte) (n int, err error) {
+func (p *LocalRWFileProxy) Read(b []byte) (n int, err error) {
 	p.file.lockFileHandle()
 	defer p.file.unLockFileHandle()
 	return p.localFile.Read(b)
 }
 
-func (p *LocalFileProxy) Close() error {
+func (p *LocalRWFileProxy) Close() error {
 	p.file.lockFileHandle()
 	defer p.file.unLockFileHandle()
 	return p.localFile.Close()
 }
 
-func (p *LocalFileProxy) Sync() error {
+func (p *LocalRWFileProxy) Sync() error {
 	p.file.lockFileHandle()
 	defer p.file.unLockFileHandle()
 	return p.localFile.Sync()

Log.go

@@ -66,7 +66,7 @@ const (
 var ReportCaller = true
 
 func init() {
-	initLogger("fatal", false, "")
+	initLogger("trace", false, "")
 }
 
 func initLogger(l string, reportCaller bool, lfile string) {