Remove SyncPending flag from flock, since it's only used in file_cache.

Remove file_cache reference from policy, and pass pendingOps through conf instead.

Author: foodprocessor

common/lock_map.go

@@ -36,8 +36,7 @@ type LockMapItem struct {
 	mtx          sync.RWMutex
 	downloadTime time.Time
 	// track if file is in lazy open state
-	LazyOpen    bool
-	SyncPending bool
+	LazyOpen bool
 }
 
 // Map holding locks for all the files

component/file_cache/async.go

@@ -186,7 +186,6 @@ func (fc *FileCache) startScheduler() {
 func (fc *FileCache) addPendingOp(name string, flock *common.LockMapItem) {
 	log.Trace("FileCache::addPendingOp : %s", name)
 	fc.pendingOps.Store(name, struct{}{})
-	flock.SyncPending = true
 	select {
 	case fc.pendingOpAdded <- struct{}{}:
 	default: // do not block
@@ -262,7 +261,8 @@ func (fc *FileCache) uploadPendingFile(name string) error {
 	defer flock.Unlock()
 
 	// don't double upload
-	if !flock.SyncPending {
+	_, stillPending := fc.pendingOps.Load(name)
+	if !stillPending {
 		return nil
 	}
 
@@ -308,13 +308,17 @@ func (fc *FileCache) uploadPendingFile(name string) error {
 		}
 	}
 	// update state
-	flock.SyncPending = false
 	log.Info("FileCache::uploadPendingFile : File uploaded: %s", name)
 	fc.pendingOps.Delete(name)
 
 	return nil
 }
 
+func (fc *FileCache) IsScheduled(objName string) bool {
+	_, inSchedule := fc.pendingOps.Load(objName)
+	return inSchedule
+}
+
 // this returns true when offline access is enabled, and it's safe to access this object offline
 func (fc *FileCache) offlineOperationAllowed(name string) bool {
 	return fc.offlineAccess && fc.notInCloud(name)

component/file_cache/cache_policy.go

@@ -28,6 +28,7 @@ package file_cache
 import (
 	"fmt"
 	"os"
+	"sync"
 
 	"github.com/Seagate/cloudfuse/common"
 	"github.com/Seagate/cloudfuse/common/log"
@@ -45,7 +46,8 @@ type cachePolicyConfig struct {
 	highThreshold float64
 	lowThreshold  float64
 
-	fileLocks *common.LockMap // uses object name (common.JoinUnixFilepath)
+	fileLocks  *common.LockMap // uses object name (common.JoinUnixFilepath)
+	pendingOps *sync.Map
 
 	policyTrace bool
 }

component/file_cache/file_cache.go

@@ -453,6 +453,7 @@ func (fc *FileCache) GetPolicyConfig(conf FileCacheOptions) cachePolicyConfig {
 		maxSizeMB:     conf.MaxSizeMB,
 		fileLocks:     fc.fileLocks,
 		policyTrace:   conf.EnablePolicyTrace,
+		pendingOps:    &fc.pendingOps,
 	}
 
 	return cacheConfig
@@ -1191,7 +1192,6 @@ func (fc *FileCache) DeleteFile(options internal.DeleteFileOptions) error {
 
 	// update file state
 	flock.LazyOpen = false
-	flock.SyncPending = false
 	// remove deleted file from async upload map
 	fc.pendingOps.Delete(options.Name)
 
@@ -2118,7 +2118,6 @@ func (fc *FileCache) renamePendingOp(srcName, dstName string, dflock *common.Loc
 	_, operationPending := fc.pendingOps.LoadAndDelete(srcName)
 	if operationPending {
 		fc.pendingOps.Store(dstName, struct{}{})
-		dflock.SyncPending = true
 	}
 }
 
@@ -2144,7 +2143,6 @@ func (fc *FileCache) renameOpenHandles(
 		}
 		// copy flags
 		dflock.LazyOpen = sflock.LazyOpen
-		dflock.SyncPending = sflock.SyncPending
 	}
 }
 

component/file_cache/file_cache_test.go

@@ -1561,10 +1561,6 @@ loopbackfs:
 	suite.assert.FileExists(filepath.Join(suite.fake_storage_path, file))
 	_, exists = suite.fileCache.pendingOps.Load(file)
 	suite.assert.False(exists, "File should have been removed from pendingOps after upload")
-	suite.assert.False(
-		suite.fileCache.fileLocks.Get(file).SyncPending,
-		"SyncPending flag should be cleared after upload",
-	)
 }
 
 func (suite *fileCacheTestSuite) TestCronOnToOFFUpload() {
@@ -1630,8 +1626,6 @@ loopbackfs:
 	suite.assert.NoFileExists(filepath.Join(suite.fake_storage_path, file2))
 	_, scheduled := suite.fileCache.pendingOps.Load(file2)
 	suite.assert.True(scheduled, "File should be scheduled when scheduler is OFF")
-	flock := suite.fileCache.fileLocks.Get(file2)
-	suite.assert.True(flock.SyncPending, "SyncPending flag should be set")
 }
 
 func (suite *fileCacheTestSuite) TestNoScheduleAlwaysOn() {
@@ -1667,11 +1661,6 @@ loopbackfs:
 	uploadedData, err := os.ReadFile(filepath.Join(suite.fake_storage_path, file))
 	suite.assert.NoError(err)
 	suite.assert.Equal(data, uploadedData, "Uploaded file content should match original")
-
-	flock := suite.fileCache.fileLocks.Get(file)
-	if flock != nil {
-		suite.assert.False(flock.SyncPending, "SyncPending flag should be clear")
-	}
 }
 
 func (suite *fileCacheTestSuite) TestExistingCloudFileImmediateUpload() {
@@ -1799,12 +1788,6 @@ loopbackfs:
 
 	_, existsInScheduleNew := suite.fileCache.pendingOps.Load(dstFile)
 	suite.assert.True(existsInScheduleNew, "New file name should be in pendingOps after rename")
-
-	// Check that file lock status was properly transferred
-	flock := suite.fileCache.fileLocks.Get(dstFile)
-	if flock != nil {
-		suite.assert.True(flock.SyncPending, "SyncPending flag should be set on renamed file")
-	}
 }
 
 func (suite *fileCacheTestSuite) TestDeleteFileAndPendingOps() {

component/file_cache/lru_policy.go

@@ -73,16 +73,13 @@ type lruPolicy struct {
 
 	// DU utility was found on the path or not
 	duPresent bool
-
-	// Tracks scheduled files to skip during eviction
-	schedule *FileCache
 }
 
 // LRUPolicySnapshot represents the *persisted state* of lruPolicy.
 // It contains only the fields that need to be saved, and they are exported.
 type LRUPolicySnapshot struct {
 	NodeList           []string // Just node names, *without their fc.tmp prefix*, in linked list order
-	SyncPendingFlags   []bool   // the SyncPending flag for each node, in the same order as NodeList
+	SyncPendingFlags   []bool   // whether each file in NodeList belongs in the pendingOps map
 	CurrMarkerPosition uint64   // Node index of currMarker
 	LastMarkerPosition uint64   // Node index of lastMarker
 }
@@ -161,11 +158,6 @@ func (p *lruPolicy) ShutdownPolicy() error {
 	return p.createSnapshot().writeToFile(p.tmpPath)
 }
 
-func (fc *FileCache) IsScheduled(objName string) bool {
-	_, inSchedule := fc.pendingOps.Load(objName)
-	return inSchedule
-}
-
 func (p *lruPolicy) createSnapshot() *LRUPolicySnapshot {
 	log.Trace("lruPolicy::saveSnapshot")
 	var snapshot LRUPolicySnapshot
@@ -183,11 +175,10 @@ func (p *lruPolicy) createSnapshot() *LRUPolicySnapshot {
 			snapshot.LastMarkerPosition = index
 		case strings.HasPrefix(current.name, p.tmpPath):
 			relName := current.name[len(p.tmpPath):]
-			objName := common.NormalizeObjectName(relName[1:])
 			snapshot.NodeList = append(snapshot.NodeList, relName)
-			flock := p.fileLocks.Get(objName)
-			// don't acquire the locks - worst case scenario, we resync something unnecessarily
-			snapshot.SyncPendingFlags = append(snapshot.SyncPendingFlags, flock.SyncPending)
+			objName := common.NormalizeObjectName(relName[1:])
+			_, isPending := p.pendingOps.Load(objName)
+			snapshot.SyncPendingFlags = append(snapshot.SyncPendingFlags, isPending)
 		default:
 			log.Err("lruPolicy::saveSnapshot : %s Ignoring unrecognized cache path", current.name)
 		}
@@ -201,7 +192,7 @@ func (p *lruPolicy) loadSnapshot(snapshot *LRUPolicySnapshot) {
 		return
 	}
 	// maintain backward compatibility
-	loadSyncPending := len(snapshot.NodeList) == len(snapshot.SyncPendingFlags)
+	loadPendingOps := len(snapshot.NodeList) == len(snapshot.SyncPendingFlags)
 	p.Lock()
 	defer p.Unlock()
 	// walk the slice and write the entries into the policy
@@ -210,11 +201,10 @@ func (p *lruPolicy) loadSnapshot(snapshot *LRUPolicySnapshot) {
 	nextNode := p.head
 	tail := p.lastMarker
 	for i, v := range snapshot.NodeList {
-		// restore SyncPending flag
-		if loadSyncPending && snapshot.SyncPendingFlags[i] {
+		// populate pendingOps
+		if loadPendingOps && snapshot.SyncPendingFlags[i] {
 			objName := v[1:]
-			flock := p.fileLocks.Get(objName)
-			flock.SyncPending = true
+			p.pendingOps.Store(objName, struct{}{})
 		}
 		// recreate the node
 		fullPath := filepath.Join(p.tmpPath, v)
@@ -524,7 +514,7 @@ func (p *lruPolicy) deleteExpiredNodes() {
 		if objName[0] == '/' {
 			objName = objName[1:]
 		}
-		if p.schedule != nil && p.schedule.IsScheduled(objName) {
+		if _, syncPending := p.pendingOps.Load(objName); syncPending {
 			continue
 		}
 
@@ -594,7 +584,7 @@ func (p *lruPolicy) deleteItem(name string) {
 	}
 
 	// check if the file is pending upload (it was modified offline)
-	if flock.SyncPending {
+	if _, syncPending := p.pendingOps.Load(objName); syncPending {
 		log.Warn("lruPolicy::DeleteItem : %s File is not synchronized to cloud storage", name)
 		p.CacheValid(name)
 		return

component/file_cache/lru_policy_test.go

@@ -315,6 +315,10 @@ func (suite *lruPolicyTestSuite) verifyPolicy(expectedPolicy, actualPolicy *lruP
 			suite.assert.Same(actualPolicy.lastMarker, actual)
 		default:
 			suite.assert.Equal(expected.name, actual.name)
+			objName := expected.name[len(suite.policy.tmpPath)+1:]
+			_, expectedPending := expectedPolicy.pendingOps.Load(objName)
+			_, actualPending := actualPolicy.pendingOps.Load(objName)
+			suite.assert.Equal(expectedPending, actualPending)
 		}
 		suite.assert.NotNil(actual, "actual list is shorter than expected")
 		suite.assert.NotNil(expected, "actual list is longer than expected")
@@ -337,6 +341,32 @@ func (suite *lruPolicyTestSuite) TestCreateSnapshotEmpty() {
 	suite.verifyPolicy(originalPolicy, suite.policy)
 }
 
+func (suite *lruPolicyTestSuite) TestCreateSnapshot() {
+	defer suite.cleanupTest()
+	// setup
+	numFiles := 5
+	pathPrefix := filepath.Join(cache_path, "temp")
+	for i := 1; i <= numFiles; i++ {
+		suite.policy.CacheValid(pathPrefix + fmt.Sprint(i))
+		if i > 3 {
+			suite.policy.pendingOps.Store("temp"+fmt.Sprint(i), struct{}{})
+		}
+	}
+	originalPolicy := suite.policy
+	// test
+	snapshot := suite.policy.createSnapshot()
+	suite.cleanupTest()
+	suite.setupTestHelper(originalPolicy.cachePolicyConfig)
+	suite.policy.loadSnapshot(snapshot)
+	// assert
+	suite.assert.NotNil(snapshot)
+	suite.assert.Len(snapshot.NodeList, numFiles)
+	for i, v := range snapshot.NodeList {
+		suite.assert.Equal(pathPrefix+fmt.Sprint(numFiles-i), filepath.Join(cache_path, v))
+	}
+	suite.verifyPolicy(originalPolicy, suite.policy)
+}
+
 func (suite *lruPolicyTestSuite) TestCreateSnapshotWithTrailingMarkers() {
 	defer suite.cleanupTest()
 	// setup
@@ -481,6 +511,7 @@ func (suite *lruPolicyTestSuite) TestSnapshotSerialization() {
 		NodeList:           []string{"a", "b", "c"},
 		CurrMarkerPosition: 1,
 		LastMarkerPosition: 2,
+		SyncPendingFlags:   []bool{true, false, false},
 	}
 	// test
 	err := snapshot.writeToFile(cache_path)
@@ -494,12 +525,11 @@ func (suite *lruPolicyTestSuite) TestSnapshotSerialization() {
 func (suite *lruPolicyTestSuite) TestNoEvictionIfInPendingOps() {
 	defer suite.cleanupTest()
 
-	fileName := filepath.Join(cache_path, "scheduled_file")
+	name := "pending_file"
+	fileName := filepath.Join(cache_path, name)
 	suite.policy.CacheValid(fileName)
 
-	fakeSchedule := &FileCache{}
-	fakeSchedule.pendingOps.Store(common.NormalizeObjectName("scheduled_file"), struct{}{})
-	suite.policy.schedule = fakeSchedule
+	suite.policy.pendingOps.Store(name, struct{}{})
 
 	time.Sleep(2 * time.Second)
 
@@ -509,20 +539,19 @@ func (suite *lruPolicyTestSuite) TestNoEvictionIfInPendingOps() {
 func (suite *lruPolicyTestSuite) TestEvictionRespectsPendingOps() {
 	defer suite.cleanupTest()
 
+	objNames := []string{"File1", "file2", "file3", "file4"}
 	fileNames := []string{
-		filepath.Join(cache_path, "file1"),
-		filepath.Join(cache_path, "file2"),
-		filepath.Join(cache_path, "file3"),
-		filepath.Join(cache_path, "file4"),
+		filepath.Join(cache_path, objNames[0]),
+		filepath.Join(cache_path, objNames[1]),
+		filepath.Join(cache_path, objNames[2]),
+		filepath.Join(cache_path, objNames[3]),
 	}
 	for _, name := range fileNames {
 		suite.policy.CacheValid(name)
 	}
 
-	fakeSchedule := &FileCache{}
-	fakeSchedule.pendingOps.Store(common.NormalizeObjectName("file2"), struct{}{})
-	fakeSchedule.pendingOps.Store(common.NormalizeObjectName("file4"), struct{}{})
-	suite.policy.schedule = fakeSchedule
+	suite.policy.pendingOps.Store(objNames[1], struct{}{})
+	suite.policy.pendingOps.Store(objNames[3], struct{}{})
 
 	time.Sleep(3 * time.Second)