Consistency for File State

component/file_cache/cache_policy.go

@@ -57,7 +57,7 @@ type cachePolicy interface {
 	UpdateConfig(cachePolicyConfig) error
 
 	CacheValid(name string) // Mark the file as hit
-	CachePurge(name string) // Schedule the file for deletion
+	CachePurge(name string) // Delete the file from cache
 
 	IsCached(name string) bool // Whether or not the cache policy considers this file cached
 

component/file_cache/file_cache.go

@@ -442,7 +442,11 @@ func (fc *FileCache) invalidateDirectory(name string) {
 		if err == nil && d != nil {
 			if !d.IsDir() {
 				log.Debug("FileCache::invalidateDirectory : removing file %s from cache", path)
+				objPath := common.JoinUnixFilepath(name, d.Name())
+				flock := fc.fileLocks.Get(objPath)
+				flock.Lock()
 				fc.policy.CachePurge(path)
+				flock.Unlock()
 			} else {
 				// remember to delete the directory later (after its children)
 				directoriesToPurge = append(directoriesToPurge, path)
@@ -541,14 +545,19 @@ func (fc *FileCache) StreamDir(options internal.StreamDirOptions) ([]*internal.O
 	if token == "" {
 		for _, entry := range dirents {
 			entryPath := common.JoinUnixFilepath(options.Name, entry.Name())
-			if !entry.IsDir() && !fc.fileLocks.Locked(entryPath) {
+			if !entry.IsDir() {
 				// This is an overhead for streamdir for now
 				// As list is paginated we have no way to know whether this particular item exists both in local cache
 				// and container or not. So we rely on getAttr to tell if entry was cached then it exists in cloud storage too
 				// If entry does not exists on storage then only return a local item here.
 				_, err := fc.NextComponent().GetAttr(internal.GetAttrOptions{Name: entryPath})
 				if err != nil && (err == syscall.ENOENT || os.IsNotExist(err)) {
-					info, err := entry.Info() // Grab local cache attributes
+					// get the lock on the file, to allow any pending operation to complete
+					flock := fc.fileLocks.Get(entryPath)
+					flock.Lock()
+					// use os.Stat instead of entry.Info() to be sure we get good info (with flock locked)
+					info, err := os.Stat(filepath.Join(localPath, entry.Name())) // Grab local cache attributes
+					flock.Unlock()
 					// If local file is not locked then only use its attributes otherwise rely on container attributes
 					if err == nil {
 						// Case 2 (file only in local cache) so create a new attributes and add them to the storage attributes
@@ -621,7 +630,13 @@ func (fc *FileCache) RenameDir(options internal.RenameDirOptions) error {
 			newPath := strings.Replace(path, localSrcPath, localDstPath, 1)
 			if !d.IsDir() {
 				log.Debug("FileCache::RenameDir : Renaming local file %s -> %s", path, newPath)
+				srcFlock := fc.fileLocks.Get(path)
+				dstFlock := fc.fileLocks.Get(newPath)
+				srcFlock.Lock()
+				dstFlock.Lock()
 				fc.renameCachedFile(path, newPath)
+				srcFlock.Unlock()
+				dstFlock.Unlock()
 			} else {
 				log.Debug("FileCache::RenameDir : Creating local destination directory %s", newPath)
 				// create the new directory
@@ -755,7 +770,6 @@ func (fc *FileCache) validateStorageError(path string, err error, method string,
 	return nil
 }
 
-// DeleteFile: Invalidate the file in local cache.
 func (fc *FileCache) DeleteFile(options internal.DeleteFileOptions) error {
 	log.Trace("FileCache::DeleteFile : name=%s", options.Name)
 
@@ -771,11 +785,6 @@ func (fc *FileCache) DeleteFile(options internal.DeleteFileOptions) error {
 	}
 
 	localPath := filepath.Join(fc.tmpPath, options.Name)
-	err = deleteFile(localPath)
-	if err != nil && !os.IsNotExist(err) {
-		log.Err("FileCache::DeleteFile : failed to delete local file %s [%s]", localPath, err.Error())
-	}
-
 	fc.policy.CachePurge(localPath)
 
 	return nil
@@ -1026,12 +1035,6 @@ func (fc *FileCache) closeFileInternal(options internal.CloseFileOptions, flock
 	if options.Handle.Fsynced() {
 		log.Trace("FileCache::closeFileInternal : fsync/sync op, purging %s", options.Handle.Path)
 		localPath := filepath.Join(fc.tmpPath, options.Handle.Path)
-
-		err = deleteFile(localPath)
-		if err != nil && !os.IsNotExist(err) {
-			log.Err("FileCache::closeFileInternal : failed to delete local file %s [%s]", localPath, err.Error())
-		}
-
 		fc.policy.CachePurge(localPath)
 		return nil
 	}
@@ -1173,6 +1176,11 @@ func (fc *FileCache) FlushFile(options internal.FlushFileOptions) error {
 	//defer exectime.StatTimeCurrentBlock("FileCache::FlushFile")()
 	log.Trace("FileCache::FlushFile : handle=%d, path=%s", options.Handle.ID, options.Handle.Path)
 
+	// lock the file state
+	flock := fc.fileLocks.Get(options.Handle.Path)
+	flock.Lock()
+	defer flock.Unlock()
+
 	// The file should already be in the cache since CreateFile/OpenFile was called before and a shared lock was acquired.
 	localPath := filepath.Join(fc.tmpPath, options.Handle.Path)
 	fc.policy.CacheValid(localPath)
@@ -1312,28 +1320,20 @@ func (fc *FileCache) GetAttr(options internal.GetAttrOptions) (*internal.ObjAttr
 
 	// To cover cases 2 and 3, grab the attributes from the local cache
 	localPath := filepath.Join(fc.tmpPath, options.Name)
+	flock := fc.fileLocks.Get(options.Name)
+	flock.Lock()
 	info, err := os.Stat(localPath)
+	flock.Unlock()
 	// All directory operations are guaranteed to be synced with storage so they cannot be in a case 2 or 3 state.
 	if err == nil && !info.IsDir() {
 		if exists { // Case 3 (file in cloud storage and in local cache) so update the relevant attributes
-			// Return from local cache only if file is not under download or deletion
-			// If file is under download then taking size or mod time from it will be incorrect.
-			if !fc.fileLocks.Locked(options.Name) {
-				log.Debug("FileCache::GetAttr : updating %s from local cache", options.Name)
-				attrs.Size = info.Size()
-				attrs.Mtime = info.ModTime()
-			} else {
-				log.Debug("FileCache::GetAttr : %s is locked, use storage attributes", options.Name)
-			}
+			log.Debug("FileCache::GetAttr : updating %s from local cache", options.Name)
+			attrs.Size = info.Size()
+			attrs.Mtime = info.ModTime()
 		} else { // Case 2 (file only in local cache) so create a new attributes and add them to the storage attributes
-			if !strings.Contains(localPath, fc.tmpPath) {
-				// Here if the path is going out of the temp directory then return ENOENT
-				exists = false
-			} else {
-				log.Debug("FileCache::GetAttr : serving %s attr from local cache", options.Name)
-				exists = true
-				attrs = newObjAttr(options.Name, info)
-			}
+			log.Debug("FileCache::GetAttr : serving %s attr from local cache", options.Name)
+			exists = true
+			attrs = newObjAttr(options.Name, info)
 		}
 	}
 
@@ -1375,6 +1375,7 @@ func (fc *FileCache) RenameFile(options internal.RenameFileOptions) error {
 	return nil
 }
 
+// both src and dst file locks should be locked before calling renameCachedFile
 func (fc *FileCache) renameCachedFile(localSrcPath string, localDstPath string) {
 	err := os.Rename(localSrcPath, localDstPath)
 	if err != nil {
@@ -1453,6 +1454,10 @@ func (fc *FileCache) TruncateFile(options internal.TruncateFileOptions) error {
 func (fc *FileCache) Chmod(options internal.ChmodOptions) error {
 	log.Trace("FileCache::Chmod : Change mode of path %s", options.Name)
 
+	flock := fc.fileLocks.Get(options.Name)
+	flock.Lock()
+	defer flock.Unlock()
+
 	// Update the file in cloud storage
 	err := fc.NextComponent().Chmod(options)
 	err = fc.validateStorageError(options.Name, err, "Chmod", false)
@@ -1487,6 +1492,10 @@ func (fc *FileCache) Chmod(options internal.ChmodOptions) error {
 func (fc *FileCache) Chown(options internal.ChownOptions) error {
 	log.Trace("FileCache::Chown : Change owner of path %s", options.Name)
 
+	flock := fc.fileLocks.Get(options.Name)
+	flock.Lock()
+	defer flock.Unlock()
+
 	// Update the file in cloud storage
 	err := fc.NextComponent().Chown(options)
 	err = fc.validateStorageError(options.Name, err, "Chown", false)

component/file_cache/file_cache_linux_test.go

@@ -62,7 +62,7 @@ func (suite *fileCacheLinuxTestSuite) SetupTest() {
 	rand := randomString(8)
 	suite.cache_path = common.JoinUnixFilepath(home_dir, "file_cache"+rand)
 	suite.fake_storage_path = common.JoinUnixFilepath(home_dir, "fake_storage"+rand)
-	defaultConfig := fmt.Sprintf("file_cache:\n  path: %s\n  offload-io: true\n  timeout-sec: 0\n\nloopbackfs:\n  path: %s", suite.cache_path, suite.fake_storage_path)
+	defaultConfig := fmt.Sprintf("file_cache:\n  path: %s\n  offload-io: true\n  timeout-sec: 1\n\nloopbackfs:\n  path: %s", suite.cache_path, suite.fake_storage_path)
 	log.Debug(defaultConfig)
 
 	// Delete the temp directories created
@@ -111,23 +111,15 @@ func (suite *fileCacheLinuxTestSuite) cleanupTest() {
 
 func (suite *fileCacheLinuxTestSuite) TestChmodNotInCache() {
 	defer suite.cleanupTest()
-	// Setup
+	// Setup - create file directly in fake storage
 	path := "file33"
-	handle, _ := suite.fileCache.CreateFile(internal.CreateFileOptions{Name: path, Mode: 0777})
-	suite.fileCache.CloseFile(internal.CloseFileOptions{Handle: handle})
-
-	_, err := os.Stat(suite.cache_path + "/" + path)
-	for i := 0; i < 1000 && !os.IsNotExist(err); i++ {
-		time.Sleep(10 * time.Millisecond)
-		_, err = os.Stat(suite.cache_path + "/" + path)
-	}
-	suite.assert.True(os.IsNotExist(err))
+	suite.loopback.CreateFile(internal.CreateFileOptions{Name: path, Mode: 0777})
 
 	// Path should be in fake storage
 	suite.assert.FileExists(suite.fake_storage_path + "/" + path)
 
 	// Chmod
-	err = suite.fileCache.Chmod(internal.ChmodOptions{Name: path, Mode: os.FileMode(0666)})
+	err := suite.fileCache.Chmod(internal.ChmodOptions{Name: path, Mode: os.FileMode(0666)})
 	suite.assert.NoError(err)
 
 	// Path in fake storage should be updated
@@ -178,7 +170,6 @@ func (suite *fileCacheLinuxTestSuite) TestChmodCase2() {
 	err = suite.fileCache.FlushFile(internal.FlushFileOptions{Handle: createHandle})
 	suite.assert.NoError(err)
 
-	// Path should be in the file cache with old mode (since we failed the operation)
 	info, err := os.Stat(suite.cache_path + "/" + path)
 	suite.assert.NoError(err)
 	suite.assert.EqualValues(info.Mode(), newMode)
@@ -187,6 +178,7 @@ func (suite *fileCacheLinuxTestSuite) TestChmodCase2() {
 	suite.assert.NoError(err)
 
 	// loop until file does not exist - done due to async nature of eviction
+	time.Sleep(time.Second)
 	_, err = os.Stat(suite.cache_path + "/" + path)
 	for i := 0; i < 1000 && !os.IsNotExist(err); i++ {
 		time.Sleep(10 * time.Millisecond)
@@ -206,23 +198,15 @@ func (suite *fileCacheLinuxTestSuite) TestChownNotInCache() {
 	defer suite.cleanupTest()
 	// Setup
 	path := "file36"
-	handle, _ := suite.fileCache.CreateFile(internal.CreateFileOptions{Name: path, Mode: 0777})
-	suite.fileCache.CloseFile(internal.CloseFileOptions{Handle: handle})
-
-	_, err := os.Stat(suite.cache_path + "/" + path)
-	for i := 0; i < 1000 && !os.IsNotExist(err); i++ {
-		time.Sleep(10 * time.Millisecond)
-		_, err = os.Stat(suite.cache_path + "/" + path)
-	}
-	suite.assert.True(os.IsNotExist(err))
+	suite.loopback.CreateFile(internal.CreateFileOptions{Name: path, Mode: 0777})
 
 	// Path should be in fake storage
 	suite.assert.FileExists(suite.fake_storage_path + "/" + path)
 
 	// Chown
 	owner := os.Getuid()
 	group := os.Getgid()
-	err = suite.fileCache.Chown(internal.ChownOptions{Name: path, Owner: owner, Group: group})
+	err := suite.fileCache.Chown(internal.ChownOptions{Name: path, Owner: owner, Group: group})
 	suite.assert.NoError(err)
 
 	// Path in fake storage should be updated