diff --git a/component/tiered_storage/lru_policy.go b/component/tiered_storage/lru_policy.go new file mode 100644 index 000000000..eface82a2 --- /dev/null +++ b/component/tiered_storage/lru_policy.go @@ -0,0 +1,371 @@ +package tiered_storage + +import ( + "fmt" + "os" + "path/filepath" + "sync" + "sync/atomic" + "time" + + "github.com/Seagate/cloudfuse/common" + "github.com/Seagate/cloudfuse/common/log" +) + +type lruNode struct { + prev *lruNode + next *lruNode + name string +} + +//upload 50 files and then check + +type lruQueue struct { + mu sync.Mutex + + nodeMap sync.Map + + wg sync.WaitGroup // tracks capacityChecker only + workerWg sync.WaitGroup // tracks worker goroutines + numWorkers int + activeWorkers int32 + + head *lruNode + tail *lruNode + + uploadChan chan string + doneChan chan struct{} + hallPassChan chan bool + + cachePath string + maxCacheSize float64 + totalUploadedSize int64 + + threshold float64 + targetRatio float64 + + tickerUnit time.Duration + + fileLocks *common.LockMap // uses object name (common.JoinUnixFilepath) + + //Functions to wire later into tiered_storage package + //upload function from tiered storage WIRE THIS LATER in tiered storage because we are using a function from there + uploadandCleanFn func(name string) error + + // //delete locally + // localPath := filepath.Join(c.tmpPath, options.Name) + // c.mu.Lock() + // delete(c.fileMap, options.Name) + // c.mu.Unlock() + // os.Remove(localPath) + +} + +func (q *lruQueue) StartPolicy() error { + if q.uploadandCleanFn == nil { + return fmt.Errorf("lruQueue: upload function not set") + } + if q.numWorkers <= 0 { + return fmt.Errorf("lruQueue: numWorkers must be > 0") + } + //initialize queue + q.head = nil + q.tail = nil + //channels + q.doneChan = make(chan struct{}) + q.hallPassChan = make(chan bool, 1) + q.hallPassChan <- true + + //go routines + q.wg.Add(1) + go q.capacityChecker() + + return nil +} + +func (q *lruQueue) StopPolicy() error { + close(q.doneChan) + // Wait for capacityChecker to exit — its deferred close(uploadChan) fires here, + // signalling workers that no more jobs are coming. + q.wg.Wait() + return nil +} + +func (q *lruQueue) Touch(name string) { + q.Enqueue(name) +} + +func (q *lruQueue) Enqueue(name string) { + //Maybe have a duplicate , that touches essentially + + //lock earlier + q.mu.Lock() + defer q.mu.Unlock() + + //search for new node first + val, found := q.nodeMap.Load(name) + if found { + // Node already exists, no need to create a new one, just touch + node := val.(*lruNode) + q.extractNode(node) + q.setHead(node) + } else { + // Node does not exist need to create a new one and put to top + newNode := &lruNode{name: name} + // if list is empty, set tail pointer to newNode + if q.tail == nil { + q.tail = newNode + } + q.nodeMap.Store(name, newNode) + q.setHead(newNode) + } +} + +func (q *lruQueue) Dequeue(name string) { + log.Trace("lruPolicy::removeNode : %s", name) + + q.mu.Lock() + defer q.mu.Unlock() + + val, found := q.nodeMap.LoadAndDelete(name) + if !found || val == nil { + return + } + + node := val.(*lruNode) + + q.extractNode(node) +} + +func (q *lruQueue) setHead(node *lruNode) { + // insert node at the head + node.prev = nil + node.next = q.head + if q.head != nil { + q.head.prev = node + } + q.head = node +} + +func (q *lruQueue) extractNode(node *lruNode) { + // remove the node from its position in the list + + // head case + if node == q.head { + q.head = node.next + } + //tail case + if node == q.tail { + q.tail = node.prev + } + + if node.next != nil { + node.next.prev = node.prev + } + if node.prev != nil { + node.prev.next = node.next + } + node.prev = nil + node.next = nil +} + +func (q *lruQueue) capacityChecker() { + defer q.wg.Done() + //defer close(q.uploadChan) + + ticker := time.NewTicker(q.tickerUnit) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + // eviction + select { + case <-q.hallPassChan: + + //1. Check if we need eviction + curSize, err := common.GetUsage(q.cachePath) + if err != nil { + log.Err("lruPolicy::capacityChecker : failed to get usage: %v", err) + q.hallPassChan <- true + continue + } + if curSize/q.maxCacheSize <= q.threshold { + q.hallPassChan <- true + break + } + //targetRatio should always be less than thresholdRatio + + //2. Find difference to evict down to target ratio + difference := curSize - q.maxCacheSize*q.targetRatio + curEvictedSpace := 0 + actualEvictedSpace := 0 + atomic.StoreInt64(&q.totalUploadedSize, 0) + evicFail := false + + //3. LRU Eviction to match difference + for actualEvictedSpace < int(difference) && !evicFail { + // Start nomination from what's already been confirmed uploaded, + // so we only nominate enough new files to cover the remaining gap. + curEvictedSpace = actualEvictedSpace + + //initialize channel + q.uploadChan = make(chan string, 1000) + + //start workers here + q.workerWg.Add(q.numWorkers) + for i := 0; i < q.numWorkers; i++ { + go q.worker() + } + + //populate upload channel for workers + for curEvictedSpace < int(difference) { + nodeSize, evicted := q.eviction() + if !evicted { + evicFail = true + break + } + curEvictedSpace += int(nodeSize) + } + + //close upload chan and wait for workers to process all remaining jobs + close(q.uploadChan) + q.workerWg.Wait() + + //update the actual evicted space here + actualEvictedSpace = int(q.totalUploadedSize) + + //if done is closed we need to check shutdown to prevent infinite loop + select { + case <-q.doneChan: + return + default: + } + + //we can also have a case here if no files were evicted then we can break on this tick if we so choose + + } + //give hall pass back when actual evicted space is satisfied + q.hallPassChan <- true + + case <-q.doneChan: + return + + //default return? + default: + } + case <-q.doneChan: + return + } + } +} + +func (q *lruQueue) eviction() (int64, bool) { + q.mu.Lock() + nodeToEvict := q.tail + if nodeToEvict == nil { + q.mu.Unlock() + return 0, false + } + + //1. loop through and find the first applicable node + for nodeToEvict != nil { + prevNode := nodeToEvict.prev + + flock := q.fileLocks.Get(nodeToEvict.name) + flock.RLock() + handleCount := flock.Count() + flock.RUnlock() + + if handleCount == 0 { + break + } + + //node has open handles touch node + q.extractNode(nodeToEvict) + q.setHead(nodeToEvict) + nodeToEvict = prevNode + } + + //all files are in use + if nodeToEvict == nil { + q.mu.Unlock() + return 0, false + } + name := nodeToEvict.name + + //2. Remove file from queue so not accidentally chosen again + q.extractNode(nodeToEvict) + q.nodeMap.Delete(name) + + q.mu.Unlock() + + //3. Get the node size that we evict + localPath := filepath.Join(q.cachePath, name) + + fileInfo, err := os.Stat(localPath) + if err != nil { + log.Err("lruPolicy::capacityChecker : failed to stat file: %v", err) + return 0, false + } + nodeSize := fileInfo.Size() + + //4. Send node to channel to be uploaded by workers + select { + case q.uploadChan <- name: + case <-q.doneChan: + return 0, false + } + + return nodeSize, true +} + +func (q *lruQueue) worker() { + defer q.workerWg.Done() + for fileName := range q.uploadChan { + + //1. Get handle count and file size + flock := q.fileLocks.Get(fileName) + flock.Lock() + handleCount := flock.Count() + + localPath := filepath.Join(q.cachePath, fileName) + fileInfo, err := os.Stat(localPath) + if err != nil { + log.Warn( + "lruPolicy::worker : file %s no longer exists or stat failed, skipping: %v", + fileName, + err, + ) + flock.Unlock() + continue + } + + fileSize := fileInfo.Size() + + //2. Check if file eligible to upload + if handleCount == 0 { + err := q.uploadandCleanFn(fileName) + flock.Unlock() + //handle when file doesn't exist during upload we do not requeue otherwise we do enqueue + if err != nil { + if os.IsNotExist(err) { + log.Warn( + "lruPolicy::worker : file %s was deleted during upload, skipping", + fileName, + ) + } else { + log.Err("lruPolicy::worker : failed to upload file %s: %v", fileName, err) + //if upload fails we have to put the file back to the queue and map to retry later + q.Touch(fileName) + } + } else { + atomic.AddInt64(&q.totalUploadedSize, fileSize) + } + //file is in use skip upload touch file back to top of queue + } else { + flock.Unlock() + q.Touch(fileName) + } + } +} diff --git a/component/tiered_storage/lru_policy_test.go b/component/tiered_storage/lru_policy_test.go new file mode 100644 index 000000000..d154f6ce5 --- /dev/null +++ b/component/tiered_storage/lru_policy_test.go @@ -0,0 +1,383 @@ +/* + Licensed under the MIT License . + + Copyright © 2023-2026 Seagate Technology LLC and/or its Affiliates + Copyright © 2020-2026 Microsoft Corporation. All rights reserved. + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in all + copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE +*/ + +package tiered_storage + +import ( + "fmt" + "io/fs" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/Seagate/cloudfuse/common" + "github.com/Seagate/cloudfuse/common/log" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/suite" +) + +type lruPolicyTestSuite struct { + suite.Suite + assert *assert.Assertions + policy *lruQueue +} + +var cache_path = filepath.Join(home_dir, "file_cache"+randomString(8)) + +func (suite *lruPolicyTestSuite) SetupTest() { + err := log.SetDefaultLogger("silent", common.LogConfig{Level: common.ELogLevel.LOG_DEBUG()}) + if err != nil { + panic(fmt.Sprintf("Unable to set silent logger as default: %v", err)) + } + suite.assert = assert.New(suite.T()) + + err = os.Mkdir(cache_path, fs.FileMode(0777)) + suite.assert.NoError(err) + + suite.setupTestHelper(cache_path, 1, 0.8, 0.6, 8) +} + +// setupTestHelper creates and starts an lruQueue for testing. +func (suite *lruPolicyTestSuite) setupTestHelper( + cachePath string, maxCacheMB float64, threshold float64, targetRatio float64, numWorkers int, +) { + suite.policy = &lruQueue{ + cachePath: cachePath, + maxCacheSize: maxCacheMB * 1024 * 1024, // convert MB to bytes + threshold: threshold, + targetRatio: targetRatio, + numWorkers: numWorkers, + tickerUnit: time.Millisecond, + fileLocks: common.NewLockMap(), // required by eviction() and worker() + + uploadandCleanFn: func(name string) error { + return nil + }, + } + + err := suite.policy.StartPolicy() + suite.assert.NoError(err) +} + +func (suite *lruPolicyTestSuite) cleanupTest() { + err := suite.policy.StopPolicy() + suite.assert.NoError(err) + + err = os.RemoveAll(cache_path) + suite.assert.NoError(err) +} + +// Test +// 1. Touch +func (suite *lruPolicyTestSuite) TestTouch() { + defer suite.cleanupTest() + //put one file in + name := "file1" + fileName := filepath.Join(cache_path, name) + suite.policy.Touch(fileName) + suite.assert.Equal(fileName, suite.policy.head.name) + suite.assert.Equal(fileName, suite.policy.tail.name) + + //put another file in + name2 := "file2" + fileName2 := filepath.Join(cache_path, name2) + suite.policy.Touch(fileName2) + suite.assert.Equal(fileName2, suite.policy.head.name) + suite.assert.Equal(fileName, suite.policy.tail.name) + + //touch file1 back to top + suite.policy.Touch(fileName) + suite.assert.Equal(fileName, suite.policy.head.name) + suite.assert.Equal(fileName2, suite.policy.tail.name) +} + +// 2. enqueueItem +func (suite *lruPolicyTestSuite) TestEnqueue() { + defer suite.cleanupTest() + //put one file in + name := "file1" + fileName := filepath.Join(cache_path, name) + suite.policy.Enqueue(fileName) + suite.assert.Equal(fileName, suite.policy.head.name) + suite.assert.Equal(fileName, suite.policy.tail.name) + + //put another file in + name2 := "file2" + fileName2 := filepath.Join(cache_path, name2) + suite.policy.Enqueue(fileName2) + suite.assert.Equal(fileName2, suite.policy.head.name) + suite.assert.Equal(fileName, suite.policy.tail.name) + + //put another file in + name3 := "file3" + fileName3 := filepath.Join(cache_path, name3) + suite.policy.Enqueue(fileName3) + suite.assert.Equal(fileName3, suite.policy.head.name) + suite.assert.Equal(fileName, suite.policy.tail.name) +} + +// 3. Dequeue +func (suite *lruPolicyTestSuite) TestDequeue() { + defer suite.cleanupTest() + //put one file in + name := "file1" + fileName := filepath.Join(cache_path, name) + suite.policy.Enqueue(fileName) + suite.assert.Equal(fileName, suite.policy.head.name) + suite.assert.Equal(fileName, suite.policy.tail.name) + + //put another file in + name2 := "file2" + fileName2 := filepath.Join(cache_path, name2) + suite.policy.Enqueue(fileName2) + suite.assert.Equal(fileName2, suite.policy.head.name) + suite.assert.Equal(fileName, suite.policy.tail.name) + + //remove + suite.policy.Dequeue(fileName) + suite.assert.Equal(fileName2, suite.policy.head.name) + suite.assert.Equal(fileName2, suite.policy.tail.name) +} + +// 5. Capacity checker, two cases +func (suite *lruPolicyTestSuite) TestCapacityCheckerEviction() { + defer suite.cleanupTest() + + var mu sync.Mutex + + //1. Define an arbitrary upload function to test the functionality of the channel + var uploaded []string + suite.policy.uploadandCleanFn = func(name string) error { + mu.Lock() + uploaded = append(uploaded, name) + os.Remove(filepath.Join(cache_path, name)) + mu.Unlock() + return nil + } + + //2. Create files that exceed the 80% threshold, max set at 1MB + data := make([]byte, 250*1024) + err := os.WriteFile(filepath.Join(cache_path, "file1"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file1") + + data = make([]byte, 250*1024) + err = os.WriteFile(filepath.Join(cache_path, "file2"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file2") + + data = make([]byte, 250*1024) + err = os.WriteFile(filepath.Join(cache_path, "file3"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file3") + + data = make([]byte, 250*1024) + err = os.WriteFile(filepath.Join(cache_path, "file4"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file4") + + _, ex1 := suite.policy.nodeMap.Load("file1") + _, ex2 := suite.policy.nodeMap.Load("file2") + _, ex3 := suite.policy.nodeMap.Load("file3") + _, ex4 := suite.policy.nodeMap.Load("file4") + + suite.assert.True(ex1) + suite.assert.True(ex2) + suite.assert.True(ex3) + suite.assert.True(ex4) + + //file4 should be in upload channel + time.Sleep(10 * time.Millisecond) + + mu.Lock() + snapshot := make([]string, len(uploaded)) + copy(snapshot, uploaded) + mu.Unlock() + + // file1 and file2 are the LRU tail so they should be evicted to reach targetRatio (60%) + suite.assert.Contains(snapshot, "file1") + suite.assert.Contains(snapshot, "file2") + // file3 and file4 are the most recently used, so they should NOT be evicted + suite.assert.NotContains(snapshot, "file3") + suite.assert.NotContains(snapshot, "file4") + + fmt.Println("=== nodeMap contents ===") + suite.policy.nodeMap.Range(func(key, value interface{}) bool { + lruNode := value.(*lruNode) + fmt.Printf(" key=%q, name=%q, next=%v, prev=%v\n", + key, + lruNode.name, + lruNode.next, + lruNode.prev, + ) + return true // continue iteration + }) + fmt.Println("=== end nodeMap ===") + + _, ex1 = suite.policy.nodeMap.Load("file1") + _, ex2 = suite.policy.nodeMap.Load("file2") + _, ex3 = suite.policy.nodeMap.Load("file3") + _, ex4 = suite.policy.nodeMap.Load("file4") + + suite.assert.False(ex1) + suite.assert.False(ex2) + suite.assert.True(ex3) + suite.assert.True(ex4) + +} + +// 6. Eviction, file with open handle, file with no open handle, +func (suite *lruPolicyTestSuite) TestCapacityCheckerEvictionOpenHandle() { + defer suite.cleanupTest() + var mu sync.Mutex + + //1. Define an arbitrary upload function to test the functionality of the channel + var uploaded []string + suite.policy.uploadandCleanFn = func(name string) error { + mu.Lock() + uploaded = append(uploaded, name) + os.Remove(filepath.Join(cache_path, name)) + mu.Unlock() + return nil + } + + //2. Create files that exceed the 80% threshold, max set at 1MB + data := make([]byte, 250*1024) + err := os.WriteFile(filepath.Join(cache_path, "file1"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file1") + + //open a file handle for file1 so it should get skipped and touched to the top, file1, file4, file3, file2 + flock := suite.policy.fileLocks.Get("file1") + flock.Lock() + flock.Inc() + flock.Unlock() + + data = make([]byte, 250*1024) + err = os.WriteFile(filepath.Join(cache_path, "file2"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file2") + + data = make([]byte, 250*1024) + err = os.WriteFile(filepath.Join(cache_path, "file3"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file3") + + data = make([]byte, 250*1024) + err = os.WriteFile(filepath.Join(cache_path, "file4"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file4") + + time.Sleep(10 * time.Millisecond) + + mu.Lock() + snapshot := make([]string, len(uploaded)) + copy(snapshot, uploaded) + mu.Unlock() + + fmt.Print(snapshot) + fmt.Print(suite.policy.head.name) + fmt.Print(suite.policy.tail.name) + + //file1 should be head, file4 should be tail + suite.assert.Equal("file1", suite.policy.head.name) + suite.assert.Equal("file4", suite.policy.tail.name) + + // file1 and file2 are the LRU tail so they should be evicted to reach targetRatio (60%) + suite.assert.Contains(snapshot, "file2") + suite.assert.Contains(snapshot, "file3") + // file3 and file4 are the most recently used, so they should NOT be evicted + suite.assert.NotContains(snapshot, "file1") + suite.assert.NotContains(snapshot, "file4") + +} + +// 7. Test done channel function +func (suite *lruPolicyTestSuite) TestStopPolicyMidUpload() { + //fill up upload chan + //call stop policy + //make sure all files in upload were indeed uploaded + + var mu sync.Mutex + + //1. Define an arbitrary upload function to test the functionality of the channel + var uploaded []string + suite.policy.uploadandCleanFn = func(name string) error { + mu.Lock() + //make upload super slow + time.Sleep(200 * time.Millisecond) + uploaded = append(uploaded, name) + os.Remove(filepath.Join(cache_path, name)) + mu.Unlock() + return nil + } + + data := make([]byte, 250*1024) + err := os.WriteFile(filepath.Join(cache_path, "file1"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file1") + + data = make([]byte, 250*1024) + err = os.WriteFile(filepath.Join(cache_path, "file2"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file2") + + data = make([]byte, 250*1024) + err = os.WriteFile(filepath.Join(cache_path, "file3"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file3") + + data = make([]byte, 250*1024) + err = os.WriteFile(filepath.Join(cache_path, "file4"), data, 0644) + suite.assert.NoError(err) + suite.policy.Enqueue("file4") + + time.Sleep(5 * time.Millisecond) + + //Stop policy + err = suite.policy.StopPolicy() + suite.assert.NoError(err) + + mu.Lock() + snapshot := make([]string, len(uploaded)) + copy(snapshot, uploaded) + mu.Unlock() + + fmt.Print(snapshot) + + suite.assert.Contains(snapshot, "file1") + suite.assert.Contains(snapshot, "file2") + suite.assert.NotContains(snapshot, "file3") + suite.assert.NotContains(snapshot, "file4") + + err = os.RemoveAll(cache_path) + suite.assert.NoError(err) +} + +func TestLRUPolicyTestSuite(t *testing.T) { + suite.Run(t, new(lruPolicyTestSuite)) +} diff --git a/component/tiered_storage/tiered_storage.go b/component/tiered_storage/tiered_storage.go index 96fd621bb..9fe0eb053 100644 --- a/component/tiered_storage/tiered_storage.go +++ b/component/tiered_storage/tiered_storage.go @@ -33,6 +33,7 @@ import ( "path/filepath" "sync" "syscall" + "time" "github.com/Seagate/cloudfuse/common" "github.com/Seagate/cloudfuse/common/config" @@ -51,8 +52,10 @@ import ( // Common structure for Component type TieredStorage struct { internal.BaseComponent - fileMap map[string]*FileNode - lruQueue *LRUQueue + //fileMap map[string]*FileNode + fileMap sync.Map + + policy *lruQueue //use LockMap instead of mutex to allow parallel access to different files fileLocks *common.LockMap // uses object name (common.JoinUnixFilepath) @@ -71,6 +74,7 @@ type FileNode struct { prev *FileNode next *FileNode cloudBacked bool + isDirty bool // Add more attributes as needed, e.g., last accessed time, etc. } @@ -85,7 +89,8 @@ type LRUQueue struct { // Structure defining your config parameters type TieredStorageOptions struct { // e.g. var1 uint32 `config:"var1"` - TmpPath string `config:"path" yaml:"path,omitempty"` + TmpPath string `config:"path" yaml:"path,omitempty"` + MaxSizeMB float64 `config:"max-size-mb" yaml:"max-size-mb,omitempty"` } const ( @@ -117,6 +122,14 @@ func (c *TieredStorage) Start(ctx context.Context) error { // TieredStorage : start code goes here + //Start the policy + if c.policy != nil { + if err := c.policy.StartPolicy(); err != nil { + log.Err("TieredStorage::Start : failed to start LRU policy [%v]", err) + return err + } + } + return nil } @@ -124,6 +137,10 @@ func (c *TieredStorage) Start(ctx context.Context) error { func (c *TieredStorage) Stop() error { log.Trace("TieredStorage::Stop : Stopping component %s", c.Name()) + if c.policy != nil { + return c.policy.StopPolicy() + } + return nil } @@ -153,6 +170,20 @@ func (c *TieredStorage) Configure(_ bool) error { return fmt.Errorf("TieredStorage: failed to create tmp path: %w", err) } + //figure out the maxCache size stuff here, there is just a bunch of configure stuff that we need to figure out + c.maxCacheSize = conf.MaxSizeMB * 1024 * 1024 + //Wire in the LRU Policy + c.policy = &lruQueue{ + cachePath: c.tmpPath, + maxCacheSize: c.maxCacheSize, + fileLocks: c.fileLocks, + threshold: 0.8, + targetRatio: 0.6, + numWorkers: 8, + tickerUnit: time.Millisecond, + uploadandCleanFn: c.uploadandCleanFile, + } + return nil } @@ -223,10 +254,13 @@ func (c *TieredStorage) createFileUnlocked( name: options.Name, size: uint64(0), cloudBacked: false, + isDirty: true, } - c.mu.Lock() - c.fileMap[options.Name] = node - c.mu.Unlock() + // c.mu.Lock() + // c.fileMap[options.Name] = node + // c.mu.Unlock() + + c.fileMap.Store(options.Name, node) //create handle handle := handlemap.NewHandle(options.Name) @@ -254,6 +288,46 @@ func (c *TieredStorage) CreateFile( } func (c *TieredStorage) DeleteFile(options internal.DeleteFileOptions) error { + log.Trace("TieredStorage::DeleteFile : name=%s", options.Name) + //Lock the file first + flock := c.fileLocks.Get(options.Name) + flock.Lock() + + //Unlock manually so we only hold one lock at a time + //defer flock.Unlock() + + val, exists := c.fileMap.Load(options.Name) + //Potential local or local + cloud state + if exists { + node := val.(*FileNode) + localPath := filepath.Join(c.tmpPath, options.Name) + + //Local and Cloud State + if node.cloudBacked { + //Both local and cloud state + //delete from cloud first + err := c.NextComponent().DeleteFile(internal.DeleteFileOptions{Name: options.Name}) + if err != nil { + flock.Unlock() + return err + } + } + //Local only State + //remove from LRU if it is in there already and delete local file + c.fileMap.Delete(options.Name) + flock.Unlock() + c.policy.Dequeue(options.Name) + os.Remove(localPath) + + //Cloud only state + } else { + //delete from cloud + flock.Unlock() + err := c.NextComponent().DeleteFile(internal.DeleteFileOptions{Name: options.Name}) + if err != nil { + return err + } + } return nil } @@ -269,9 +343,7 @@ func (c *TieredStorage) OpenFile(options internal.OpenFileOptions) (*handlemap.H //Case 1: OpenFile with O_Create if options.Flags&os.O_CREATE != 0 { //Check if file first exists, then proceed - c.mu.Lock() - _, exists := c.fileMap[options.Name] - c.mu.Unlock() + _, exists := c.fileMap.Load(options.Name) if !exists { handle, err := c.createFileUnlocked( internal.CreateFileOptions{Name: options.Name, Mode: options.Mode}, @@ -285,9 +357,7 @@ func (c *TieredStorage) OpenFile(options internal.OpenFileOptions) (*handlemap.H } //1. Initial Check Map - c.mu.Lock() - _, exists := c.fileMap[options.Name] - c.mu.Unlock() + _, exists := c.fileMap.Load(options.Name) //if exists skip to opening file since it should already be in local cache if !exists { @@ -304,9 +374,8 @@ func (c *TieredStorage) OpenFile(options internal.OpenFileOptions) (*handlemap.H size: uint64(info.Size()), cloudBacked: false, } - c.mu.Lock() - c.fileMap[options.Name] = node - c.mu.Unlock() + c.fileMap.Store(options.Name, node) + } else { //3. Check if File exists in Cloud info, err := c.GetAttr(internal.GetAttrOptions{Name: options.Name}) @@ -333,9 +402,8 @@ func (c *TieredStorage) OpenFile(options internal.OpenFileOptions) (*handlemap.H if err != nil { return nil, err } - c.mu.Lock() - c.fileMap[options.Name] = localCopyNode - c.mu.Unlock() + c.fileMap.Store(options.Name, localCopyNode) + } } @@ -415,11 +483,9 @@ func (c *TieredStorage) isOverLocalLimit( //find ExistingSize of file if exists existingSize := uint64(0) - c.mu.Lock() - if node, ok := c.fileMap[fileName]; ok { - existingSize = node.size + if val, ok := c.fileMap.Load(fileName); ok { + existingSize = val.(*FileNode).size } - c.mu.Unlock() addedFileSize := int64(newFileSize) - int64(existingSize) @@ -489,11 +555,12 @@ func (c *TieredStorage) WriteFile(options *internal.WriteFileOptions) (int, erro if err == nil { c.setHandleDirty(options.Handle) //update file node size in file map - c.mu.Lock() - if node, ok := c.fileMap[options.Handle.Path]; ok { + if val, ok := c.fileMap.Load(options.Handle.Path); ok { + node := val.(*FileNode) node.size = uint64(newSize) + node.isDirty = true } - c.mu.Unlock() + } else { log.Err( "TieredStorage::WriteFile : failed to write %s [%s]", @@ -506,10 +573,39 @@ func (c *TieredStorage) WriteFile(options *internal.WriteFileOptions) (int, erro } func (c *TieredStorage) SyncFile(options internal.SyncFileOptions) error { - return nil + log.Trace( + "TieredStorage::SyncFile : handle=%d, path=%s", + options.Handle.ID, + options.Handle.Path, + ) + return c.FlushFile(internal.FlushFileOptions{Handle: options.Handle}) } func (c *TieredStorage) FlushFile(options internal.FlushFileOptions) error { + //Ok so we just need to flush locally, which means just write it to the disc + log.Trace( + "TieredStorage::FlushFile : handle=%d, path=%s", + options.Handle.ID, + options.Handle.Path, + ) + + //1. Only need to flush dirty files + if !options.Handle.Dirty() { + return nil + } + //2. Check if there is local file object form handle + f := options.Handle.GetFileObject() + if f == nil { + log.Err("TieredStorage::FlushFile : %s no file object in handle", options.Handle.Path) + return syscall.EBADF + } + //3. Sync to Disk + err := f.Sync() + if err != nil { + log.Err("TieredStorage::FlushFile : %s sync failed [%v]", options.Handle.Path, err) + return syscall.EIO + } + return nil } @@ -517,32 +613,44 @@ func (c *TieredStorage) ReleaseFile(options internal.ReleaseFileOptions) error { // get the file lock, so only one open call can proceed for a file, other calls will wait here until lock is released flock := c.fileLocks.Get(options.Handle.Path) flock.Lock() - defer flock.Unlock() - //Dec Handle First + //Ok we have to manually unlock the file now instead + //defer flock.Unlock() + + //Dec Handle Count First flock.Dec() + //close file associated with handle + if f := options.Handle.GetFileObject(); f != nil { + f.Close() + } + + //clean handle state + c.clearHandleDirty(options.Handle) + options.Handle.Cleanup() + + //remove from global handle map + handlemap.Delete(options.Handle.ID) + //Check if this is the last file handle handleCount := flock.Count() //it is the last handle if handleCount == 0 { //is file cloudbacked - c.mu.Lock() - node, exists := c.fileMap[options.Handle.Path] - c.mu.Unlock() - - if !exists { + val, ok := c.fileMap.Load(options.Handle.Path) + if !ok { log.Err( "TieredStorage::ReleaseFile : internal error: file %s not found in map", options.Handle.Path, ) + flock.Unlock() return syscall.EBADF } - + node := val.(*FileNode) if node.cloudBacked { //File was modified - if options.Handle.Dirty() { + if node.isDirty { //Upload err := c.uploadCachedFile(options.Handle.Path) if err != nil { @@ -551,31 +659,26 @@ func (c *TieredStorage) ReleaseFile(options internal.ReleaseFileOptions) error { options.Handle.Path, err, ) - options.Handle.Cleanup() + flock.Unlock() return err } - //Delete local file copy - localPath := filepath.Join(c.tmpPath, options.Handle.Path) - c.mu.Lock() - delete(c.fileMap, options.Handle.Path) - c.mu.Unlock() - //Clean Handle - options.Handle.Cleanup() - os.Remove(localPath) - } else { - //File was not modified - localPath := filepath.Join(c.tmpPath, options.Handle.Path) - c.mu.Lock() - delete(c.fileMap, options.Handle.Path) - c.mu.Unlock() - options.Handle.Cleanup() - os.Remove(localPath) - } + //Whether File was modified or not, delete local file copy + localPath := filepath.Join(c.tmpPath, options.Handle.Path) + c.fileMap.Delete(options.Handle.Path) + os.Remove(localPath) + flock.Unlock() } else { - //local only then just close the file, update LRU add to queue, we will get to this later - options.Handle.Cleanup() + // update LRU add to queue because cleaning up the file should be handled once the file is uploaded in LRU policy logic + //Unlock the file first so we don't hold two locks at the same time + flock.Unlock() + if c.policy != nil { + c.policy.Enqueue(options.Handle.Path) + } } + } else { + //if not the last handle + flock.Unlock() } return nil } @@ -605,10 +708,116 @@ func (c *TieredStorage) uploadCachedFile(name string) error { return uploadErr } +func (c *TieredStorage) uploadandCleanFile(name string) error { + err := c.uploadCachedFile(name) + if err != nil { + return err + } + localPath := filepath.Join(c.tmpPath, name) + c.fileMap.Delete(name) + err = os.Remove(localPath) + if err != nil { + log.Err("TieredStorage::uploadandCleanFile : %s remove failed [%v]", name, err) + return err + } + return nil +} + func (c *TieredStorage) RenameFile(options internal.RenameFileOptions) error { + //Ok we are going to follow DeleteFile, we have to rename this File in the various states that its in + //So First we lock in alphabetical order + log.Trace("TieredStorage::RenameFile : src=%s, dst=%s", options.Src, options.Dst) + + sflock := c.fileLocks.Get(options.Src) + dflock := c.fileLocks.Get(options.Dst) + + if options.Src < options.Dst { + sflock.Lock() + dflock.Lock() + } else { + dflock.Lock() + sflock.Lock() + } + defer sflock.Unlock() + defer dflock.Unlock() + + //Ok now we have to consider all the states + //Rename File that is Local Only + //sync map in both local and in the LRU, also need to rename the node in the queue + //Check that it exists + val, exists := c.fileMap.Load(options.Src) + //Potential local or local + cloud state + if exists { + node := val.(*FileNode) + // //Local and Cloud State + if node.cloudBacked { + //just rename from the cloud + err := c.NextComponent().RenameFile(options) + if err != nil { + return err + } + } + //Local only State, this will happen anyways if it exists local + //Rename + err := os.Rename( + filepath.Join(c.tmpPath, options.Src), + filepath.Join(c.tmpPath, options.Dst), + ) + if err != nil { + return err + } + c.fileMap.Delete(options.Src) + node.name = options.Dst + c.fileMap.Store(options.Dst, node) + + //Check if it is in the LRU first + _, inLRU := c.policy.nodeMap.Load(options.Src) + if inLRU { + c.policy.Dequeue(options.Src) + c.policy.Enqueue(options.Dst) + } + //Change the handle and the lock counts + c.renameOpenHandles(options.Src, options.Dst, sflock, dflock) + + //Cloud only state + } else { + err := c.NextComponent().RenameFile(options) + if err != nil { + return err + } + } return nil } +// flock must be locked for both files +func (c *TieredStorage) renameOpenHandles( + srcName, dstName string, + sflock, dflock *common.LockMapItem, +) { + // update open handles + if sflock.Count() > 0 { + // update any open handles to the file with its new name + handlemap.GetHandles().Range(func(key, value any) bool { + handle := value.(*handlemap.Handle) + handle.Lock() + if handle.Path == srcName { + handle.Path = dstName + } + handle.Unlock() + return true + }) + // copy the number of open handles to the new name + for sflock.Count() > 0 { + sflock.Dec() + dflock.Inc() + } + for sflock.DirtyCount() > 0 { + sflock.DecDirty() + dflock.IncDirty() + } + } +} + func (c *TieredStorage) SyncDir(options internal.SyncDirOptions) error { return nil } @@ -679,8 +888,6 @@ func (c *TieredStorage) StatFs() (*common.Statfs_t, bool, error) { // << DO NOT DELETE ANY AUTO GENERATED CODE HERE >> func NewTieredStorageComponent() internal.Component { comp := &TieredStorage{ - fileMap: make(map[string]*FileNode), - lruQueue: &LRUQueue{}, fileLocks: common.NewLockMap(), } comp.SetName(compName) diff --git a/component/tiered_storage/tiered_storage_test.go b/component/tiered_storage/tiered_storage_test.go index fadf0afab..05972148d 100644 --- a/component/tiered_storage/tiered_storage_test.go +++ b/component/tiered_storage/tiered_storage_test.go @@ -9,6 +9,7 @@ import ( "strings" "syscall" "testing" + "time" "github.com/Seagate/cloudfuse/common" "github.com/Seagate/cloudfuse/common/config" @@ -71,7 +72,7 @@ func (suite *tieredStorageTestSuite) SetupTest() { suite.cache_path = filepath.Join(home_dir, "file_cache"+rand) suite.fake_storage_path = filepath.Join(home_dir, "fake_storage"+rand) defaultConfig := fmt.Sprintf( - "tiered_storage:\n path: %s\n offload-io: true\n\nloopbackfs:\n path: %s", + "tiered_storage:\n path: %s\n max-size-mb: 1.0\n offload-io: true\n\nloopbackfs:\n path: %s", suite.cache_path, suite.fake_storage_path, ) @@ -349,11 +350,15 @@ func (suite *tieredStorageTestSuite) TestReleaseCloudNoDirtyFile() { suite.assert.FileExists(filepath.Join(suite.cache_path, path)) //As of now, the file would be cloudbacked and exist in map - suite.tieredStorage.mu.Lock() - node, exists := suite.tieredStorage.fileMap[path] - suite.tieredStorage.mu.Unlock() + // suite.tieredStorage.mu.Lock() + // node, exists := suite.tieredStorage.fileMap[path] + // suite.tieredStorage.mu.Unlock() + + val, ok := suite.tieredStorage.fileMap.Load(path) + node := val.(*FileNode) + suite.assert.True(node.cloudBacked, "File should be marked as cloud-backed") - suite.assert.True(exists, "File should be tracked in the fileMap") + suite.assert.True(ok, "File should be tracked in the fileMap") //File should be "cloudBacked" and not dirty so on release the file should be deleted from local and the handle clean err = suite.tieredStorage.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) @@ -400,9 +405,13 @@ func (suite *tieredStorageTestSuite) TestReleaseCloudDirtyFile() { suite.assert.True(handle.Dirty()) //As of now, the file would be cloudbacked and exist in map - suite.tieredStorage.mu.Lock() - node, exists := suite.tieredStorage.fileMap[path] - suite.tieredStorage.mu.Unlock() + // suite.tieredStorage.mu.Lock() + // node, exists := suite.tieredStorage.fileMap[path] + // suite.tieredStorage.mu.Unlock() + + val, exists := suite.tieredStorage.fileMap.Load(path) + node := val.(*FileNode) + suite.assert.True(node.cloudBacked, "File should be marked as cloud-backed") suite.assert.True(exists, "File should be tracked in the fileMap") @@ -487,6 +496,396 @@ func (suite *tieredStorageTestSuite) TestReadInBufferErrorBadFd() { suite.assert.Equal(0, length) } +func (suite *tieredStorageTestSuite) TestWriteReadDirtyState() { + defer suite.cleanupTest() + path := "file16" + + //put file in cloud + handle, _ := suite.loopback.CreateFile(internal.CreateFileOptions{Name: path, Mode: 0777}) + err := suite.loopback.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) + suite.assert.NoError(err) + + //open file through tiered storage, should succeed and return a handle with correct path + handle, openErr := suite.tieredStorage.OpenFile( + internal.OpenFileOptions{ + Name: path, + Flags: os.O_RDWR, + Mode: 0666, //random mode, since we didn't do the other stuff yet + }, + ) + suite.assert.NoError(openErr) + + // Verify it was now downloaded to the local tiered storage cache + in map + suite.assert.FileExists(filepath.Join(suite.cache_path, path)) + // suite.tieredStorage.mu.Lock() + // node, exists := suite.tieredStorage.fileMap[path] + // suite.tieredStorage.mu.Unlock() + + val, exists := suite.tieredStorage.fileMap.Load(path) + node := val.(*FileNode) + + suite.assert.True(node.cloudBacked, "File should be marked as cloud-backed") + suite.assert.True(exists, "File should be tracked in the fileMap") + + //1. Write to handle + testData := "test data" + data := []byte(testData) + length, err := suite.tieredStorage.WriteFile( + &internal.WriteFileOptions{Handle: handle, Offset: 0, Data: data}, + ) + suite.assert.NoError(err) + suite.assert.Equal(len(data), length) + + //check the handle is dirty + suite.assert.True(handle.Dirty()) + + //2. New Read Handle to same file + handle2, openErr := suite.tieredStorage.OpenFile( + internal.OpenFileOptions{ + Name: path, + Flags: os.O_RDWR, + Mode: 0666, //random mode, since we didn't do the other stuff yet + }, + ) + suite.assert.NoError(openErr) + output := make([]byte, 9) + length, err = suite.tieredStorage.ReadInBuffer( + &internal.ReadInBufferOptions{Handle: handle2, Offset: 0, Data: output}, + ) + suite.assert.NoError(err) + suite.assert.Equal(data, output) + suite.assert.Equal(len(data), length) + + //3. Release The write handle, should still be in local with a dirty handle + err = suite.tieredStorage.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) + suite.assert.NoError(err) + _, err = os.Stat(filepath.Join(suite.cache_path, path)) + suite.assert.False(os.IsNotExist(err), "File should not be uploaded") + + //check still dirty + suite.assert.False(handle2.Dirty()) + suite.assert.False(handle.Dirty()) + + // suite.tieredStorage.mu.Lock() + // node, _ = suite.tieredStorage.fileMap[path] + // suite.tieredStorage.mu.Unlock() + + val, _ = suite.tieredStorage.fileMap.Load(path) + node = val.(*FileNode) + + suite.assert.True(node.isDirty, "File should be marked as dirty") + + //4. Release the read should upload to cloud + err = suite.tieredStorage.ReleaseFile(internal.ReleaseFileOptions{Handle: handle2}) + suite.assert.NoError(err) + _, err = os.Stat(filepath.Join(suite.cache_path, path)) + suite.assert.True(os.IsNotExist(err), "File should be deleted from cache after release") + + //5. Check data + //It just checks if the data is preserved + tmpFile, err := os.CreateTemp("", "cloud_verify") + suite.assert.NoError(err) + defer os.Remove(tmpFile.Name()) + defer tmpFile.Close() + + // 2. Copy from the cloud (loopback) to the temporary file + err = suite.loopback.CopyToFile(internal.CopyToFileOptions{ + Name: path, + Offset: 0, + Count: 0, // 0 usually means the whole file + File: tmpFile, + }) + suite.assert.NoError(err) + + // 3. Read the data back from the temp file and verify + dataFromCloud, err := os.ReadFile(tmpFile.Name()) + suite.assert.NoError(err) + suite.assert.Equal( + data, + dataFromCloud, + "The cloud version should match the modified local version", + ) +} + +func (suite *tieredStorageTestSuite) TestReleaseLocalToLRUQueue() { + //Ok this next test is to essentially go through an iteration of LRU, + + //1. Initialize a local only file + defer suite.cleanupTest() + path := "file17" + handle, err := suite.tieredStorage.OpenFile( + internal.OpenFileOptions{Name: path, Flags: os.O_CREATE, Mode: 0777}, + ) + suite.assert.NoError(err) + suite.assert.Equal(path, handle.Path) + suite.assert.True(handle.Dirty()) + // File should exist in cache + suite.assert.FileExists(filepath.Join(suite.cache_path, path)) + + val, exists := suite.tieredStorage.fileMap.Load(path) + node := val.(*FileNode) + + suite.assert.False(node.cloudBacked, "File should not be marked as cloud-backed") + suite.assert.True(exists, "File should be tracked in the fileMap") + + // 2. Release this local file + //File is local only so it shouldn't be deleted from local knowledge + err = suite.tieredStorage.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) + suite.assert.NoError(err) + + // 3. Check if its in the LRU Queue + suite.assert.Equal(path, suite.tieredStorage.policy.head.name) + suite.assert.Equal(path, suite.tieredStorage.policy.tail.name) + +} + +func (suite *tieredStorageTestSuite) TestReleaseToTriggerEviction() { + // Ok this next test is to essentially go through an iteration of LRU, + // 1. Initialize many local only file + //2. Create files that exceed the 80% threshold, max set at 1MB + data := make([]byte, 250*1024) + path1 := "file18" + handle, err := suite.tieredStorage.OpenFile( + internal.OpenFileOptions{Name: path1, Flags: os.O_CREATE, Mode: 0777}, + ) + suite.assert.NoError(err) + suite.assert.Equal(path1, handle.Path) + + suite.tieredStorage.WriteFile(&internal.WriteFileOptions{Handle: handle, Data: data}) + suite.assert.True(handle.Dirty()) + + err = suite.tieredStorage.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) + suite.assert.NoError(err) + + path2 := "file19" + handle, err = suite.tieredStorage.OpenFile( + internal.OpenFileOptions{Name: path2, Flags: os.O_CREATE, Mode: 0777}, + ) + suite.assert.NoError(err) + suite.assert.Equal(path2, handle.Path) + + suite.tieredStorage.WriteFile(&internal.WriteFileOptions{Handle: handle, Data: data}) + suite.assert.True(handle.Dirty()) + + err = suite.tieredStorage.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) + suite.assert.NoError(err) + + path3 := "file20" + handle, err = suite.tieredStorage.OpenFile( + internal.OpenFileOptions{Name: path3, Flags: os.O_CREATE, Mode: 0777}, + ) + suite.assert.NoError(err) + suite.assert.Equal(path3, handle.Path) + + suite.tieredStorage.WriteFile(&internal.WriteFileOptions{Handle: handle, Data: data}) + suite.assert.True(handle.Dirty()) + + err = suite.tieredStorage.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) + suite.assert.NoError(err) + + path4 := "file21" + handle, err = suite.tieredStorage.OpenFile( + internal.OpenFileOptions{Name: path4, Flags: os.O_CREATE, Mode: 0777}, + ) + suite.assert.NoError(err) + suite.assert.Equal(path4, handle.Path) + + suite.tieredStorage.WriteFile(&internal.WriteFileOptions{Handle: handle, Data: data}) + suite.assert.True(handle.Dirty()) + + err = suite.tieredStorage.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) + suite.assert.NoError(err) + + // 3. Check if all in the LRU Queue initially + suite.assert.Equal(path4, suite.tieredStorage.policy.head.name) + suite.assert.Equal(path3, suite.tieredStorage.policy.head.next.name) + suite.assert.Equal(path2, suite.tieredStorage.policy.head.next.next.name) + suite.assert.Equal(path1, suite.tieredStorage.policy.tail.name) + + _, exists1 := suite.tieredStorage.policy.nodeMap.Load(path1) + _, exists2 := suite.tieredStorage.policy.nodeMap.Load(path2) + _, exists3 := suite.tieredStorage.policy.nodeMap.Load(path3) + _, exists4 := suite.tieredStorage.policy.nodeMap.Load(path4) + + suite.assert.True(exists1) + suite.assert.True(exists2) + suite.assert.True(exists3) + suite.assert.True(exists4) + + //4. Sleep to wait for eviction to kick in + time.Sleep(100 * time.Millisecond) + + // 4. Some should then be released to the cloud essentially, the ones we wrote data to + //And the local files should be gone (uploaded and cleaned up), not in either map + + // 4a. Check state of NodeMap + _, exists1 = suite.tieredStorage.policy.nodeMap.Load(path1) + _, exists2 = suite.tieredStorage.policy.nodeMap.Load(path2) + _, exists3 = suite.tieredStorage.policy.nodeMap.Load(path3) + _, exists4 = suite.tieredStorage.policy.nodeMap.Load(path4) + + suite.assert.False(exists1) + suite.assert.False(exists2) + suite.assert.True(exists3) + suite.assert.True(exists4) + + //4b. Check state of fileMap + _, exists1 = suite.tieredStorage.fileMap.Load(path1) + _, exists2 = suite.tieredStorage.fileMap.Load(path2) + _, exists3 = suite.tieredStorage.fileMap.Load(path3) + _, exists4 = suite.tieredStorage.fileMap.Load(path4) + + suite.assert.False(exists1) + suite.assert.False(exists2) + suite.assert.True(exists3) + suite.assert.True(exists4) + + // 4c.Check files for files 1 and 2 no longer exist local + suite.assert.NoFileExists(filepath.Join(suite.cache_path, path1)) + suite.assert.NoFileExists(filepath.Join(suite.cache_path, path2)) + + //5. We have to check that the files exist in the cloud + //Must check that file is actually in the cloud + _, err = suite.tieredStorage.NextComponent().GetAttr( + internal.GetAttrOptions{Name: path1, RetrieveMetadata: true}) + suite.assert.NoError(err) + + //Must check that file is actually in the cloud + _, err = suite.tieredStorage.NextComponent().GetAttr( + internal.GetAttrOptions{Name: path2, RetrieveMetadata: true}) + suite.assert.NoError(err) + + //Validate the data matches what we have + //It just checks if the data is preserved + tmpFile, err := os.CreateTemp("", "cloud_verify") + suite.assert.NoError(err) + defer os.Remove(tmpFile.Name()) + defer tmpFile.Close() + + // 2. Copy from the cloud (loopback) to the temporary file + err = suite.loopback.CopyToFile(internal.CopyToFileOptions{ + Name: path1, + Offset: 0, + Count: 0, // 0 usually means the whole file + File: tmpFile, + }) + suite.assert.NoError(err) + + // 3. Read the data back from the temp file and verify + dataFromCloud, err := os.ReadFile(tmpFile.Name()) + suite.assert.NoError(err) + suite.assert.Equal( + data, + dataFromCloud, + "The cloud version should match the modified local version", + ) + + //It just checks if the data is preserved + tmpFile, err = os.CreateTemp("", "cloud_verify") + suite.assert.NoError(err) + defer os.Remove(tmpFile.Name()) + defer tmpFile.Close() + + // 2. Copy from the cloud (loopback) to the temporary file + err = suite.loopback.CopyToFile(internal.CopyToFileOptions{ + Name: path2, + Offset: 0, + Count: 0, // 0 usually means the whole file + File: tmpFile, + }) + suite.assert.NoError(err) + + // 3. Read the data back from the temp file and verify + dataFromCloud, err = os.ReadFile(tmpFile.Name()) + suite.assert.NoError(err) + suite.assert.Equal( + data, + dataFromCloud, + "The cloud version should match the modified local version", + ) + +} + +// ok we gonna do file in local, cloud, file doesnt exist +func (suite *tieredStorageTestSuite) TestDeleteFileCloud() { + defer suite.cleanupTest() + // Setup + file := "file22" + + //put file in cloud abd write to it + handle, err := suite.tieredStorage.CreateFile( + internal.CreateFileOptions{Name: file, Mode: 0777}, + ) + suite.assert.NoError(err) + err = suite.tieredStorage.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) + suite.assert.NoError(err) + + err = suite.tieredStorage.DeleteFile(internal.DeleteFileOptions{Name: file}) + suite.assert.NoError(err) + + // Path should not be in file cache + suite.assert.NoFileExists(filepath.Join(suite.cache_path, file)) + + //file should not exist in cloud + _, err = suite.tieredStorage.NextComponent().GetAttr( + internal.GetAttrOptions{Name: file, RetrieveMetadata: true}) + suite.assert.Error(err) + +} + +func (suite *tieredStorageTestSuite) TestDeleteFileLocal() { + defer suite.cleanupTest() + // Setup + file := "file23" + + //create local file + _, err := suite.tieredStorage.CreateFile( + internal.CreateFileOptions{Name: file, Mode: 0777}, + ) + suite.assert.NoError(err) + + err = suite.tieredStorage.DeleteFile(internal.DeleteFileOptions{Name: file}) + suite.assert.NoError(err) + + // Path should not be in file cache + suite.assert.NoFileExists(filepath.Join(suite.cache_path, file)) + +} + +func (suite *tieredStorageTestSuite) TestDeleteFileNotExists() { + defer suite.cleanupTest() + // Setup + file := "file24" + + err := suite.tieredStorage.DeleteFile(internal.DeleteFileOptions{Name: file}) + suite.assert.Error(err) + suite.assert.EqualValues(syscall.ENOENT, err) +} + +func (suite *tieredStorageTestSuite) TestFlushFile() { + defer suite.cleanupTest() + file := "file25" + handle, _ := suite.tieredStorage.CreateFile(internal.CreateFileOptions{Name: file, Mode: 0777}) + + testData := "test data" + data := []byte(testData) + _, err := suite.tieredStorage.WriteFile( + &internal.WriteFileOptions{Handle: handle, Offset: 0, Data: data}, + ) + suite.assert.NoError(err) + suite.assert.True(handle.Dirty()) + + err = suite.tieredStorage.FlushFile(internal.FlushFileOptions{Handle: handle}) + suite.assert.NoError(err) + + //Verify Data is still on the disk + d, _ := os.ReadFile(filepath.Join(suite.cache_path, file)) + suite.assert.Equal(data, d) + //Check that handle is still dirty + suite.assert.True(handle.Dirty()) + +} + func TestTieredStorageTestSuite(t *testing.T) { suite.Run(t, new(tieredStorageTestSuite)) }