Skip to content

Commit

Permalink
Fix concurrent writesto fs cache
Browse files Browse the repository at this point in the history
  • Loading branch information
td5r committed Dec 7, 2022
1 parent 50f741c commit 432e1f0
Showing 1 changed file with 75 additions and 49 deletions.
124 changes: 75 additions & 49 deletions pkg/storage/fs.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,13 @@ package storage

import (
"errors"
"fmt"
"io/ioutil"
"log"
"os"
"path/filepath"
"strings"
"sync"

"github.com/Bitspark/go-funk"
"github.com/Bitspark/slang/pkg/core"
Expand All @@ -18,9 +20,9 @@ import (
var FILE_ENDINGS = []string{".yaml", ".yml", ".json"} // Order of endings matters!

type FileSystem struct {
root string
cache map[uuid.UUID]*core.Blueprint
uuids []uuid.UUID
root string
cache map[uuid.UUID]*core.Blueprint
cacheLock sync.Mutex
}

type WritableFileSystem struct {
Expand All @@ -39,12 +41,22 @@ func cleanPath(p string) string {

func NewWritableFileSystem(root string) *WritableFileSystem {
p := cleanPath(root)
return &WritableFileSystem{FileSystem: FileSystem{p, make(map[uuid.UUID]*core.Blueprint), nil}}
return &WritableFileSystem{
FileSystem{
p,
make(map[uuid.UUID]*core.Blueprint),
sync.Mutex{},
},
}
}

func NewReadOnlyFileSystem(root string) *FileSystem {
p := cleanPath(root)
return &FileSystem{p, make(map[uuid.UUID]*core.Blueprint), nil}
return &FileSystem{
p,
make(map[uuid.UUID]*core.Blueprint),
sync.Mutex{},
}
}

func (fs *FileSystem) Has(opId uuid.UUID) bool {
Expand All @@ -53,44 +65,13 @@ func (fs *FileSystem) Has(opId uuid.UUID) bool {
}

func (fs *FileSystem) List() ([]uuid.UUID, error) {
if fs.uuids != nil {
return fs.uuids, nil
fmt.Println(">", fs.root)
fmt.Println(funk.Keys(fs.cache))
if len(fs.cache) == 0 {
fs.loadBlueprintFiles()
}

opsFilePathSet := make(map[uuid.UUID]bool)

_ = filepath.Walk(fs.root, func(path string, info os.FileInfo, err error) error {
if err != nil {
log.Printf("cannot read file %s: %s", path, err)
return nil
}

// Prevent recursive walk. Just read files within fs.root
if info.IsDir() && path != fs.root {
return filepath.SkipDir
}

if info.IsDir() ||
strings.HasPrefix(info.Name(), ".") ||
!fs.hasSupportedSuffix(info.Name()) {
return nil
}

blueprint, err := fs.readBlueprintFile(path)

if err != nil {
log.Printf("cannot read file %s: %s", path, err)
return nil
}

opsFilePathSet[blueprint.Id] = true

return nil
})

fs.uuids = funk.Keys(opsFilePathSet).([]uuid.UUID)

return fs.List()
return funk.Keys(fs.cache).([]uuid.UUID), nil
}

func (fs *FileSystem) Load(opId uuid.UUID) (*core.Blueprint, error) {
Expand All @@ -103,12 +84,15 @@ func (fs *FileSystem) Load(opId uuid.UUID) (*core.Blueprint, error) {
return nil, err
}

fs.cache[opId], err = fs.readBlueprintFile(blueprintFile)
blueprint, err := fs.readBlueprintFile(blueprintFile)

if err != nil {
return nil, err
}

return fs.Load(opId)
fs.cacheThis(blueprint)

return blueprint, nil
}

func (fs *WritableFileSystem) Save(blueprint core.Blueprint) (uuid.UUID, error) {
Expand All @@ -122,8 +106,7 @@ func (fs *WritableFileSystem) Save(blueprint core.Blueprint) (uuid.UUID, error)
return opId, err
}

delete(fs.cache, opId)
fs.uuids = append(fs.uuids, opId)
fs.cacheThis(&blueprint)

blueprintYaml, err := yaml.Marshal(&blueprint)

Expand All @@ -141,19 +124,31 @@ func (fs *WritableFileSystem) Save(blueprint core.Blueprint) (uuid.UUID, error)

func (fs *WritableFileSystem) List() ([]uuid.UUID, error) {
// force to reload writable/local blueprints
fs.clearCache()
fs.clearCache(nil)
return fs.FileSystem.List()
}

func (fs *WritableFileSystem) Load(opId uuid.UUID) (*core.Blueprint, error) {
// force to reload writable/local blueprints
delete(fs.cache, opId)
fs.clearCache(&opId)
return fs.FileSystem.Load(opId)
}

func (fs *WritableFileSystem) clearCache() {
func (fs *FileSystem) cacheThis(blueprint *core.Blueprint) {
fs.cacheLock.Lock()
fs.cache[blueprint.Id] = blueprint
fs.cacheLock.Unlock()
}

func (fs *WritableFileSystem) clearCache(blueprintId *uuid.UUID) {
fs.cacheLock.Lock()
if blueprintId != nil {
delete(fs.cache, *blueprintId)
fs.cacheLock.Unlock()
return
}
fs.cache = make(map[uuid.UUID]*core.Blueprint)
fs.uuids = nil
fs.cacheLock.Unlock()
}

func (fs *FileSystem) hasSupportedSuffix(filePath string) bool {
Expand All @@ -168,6 +163,37 @@ func (fs *FileSystem) getFilePath(opId uuid.UUID) (string, error) {
return utils.FileWithFileEnding(filepath.Join(fs.root, opId.String()), FILE_ENDINGS)
}

func (fs *FileSystem) loadBlueprintFiles() {
_ = filepath.Walk(fs.root, func(path string, info os.FileInfo, err error) error {
if err != nil {
log.Printf("cannot read file %s: %s", path, err)
return nil
}

// Prevent recursive walk. Just read files within fs.root
if info.IsDir() && path != fs.root {
return filepath.SkipDir
}

if info.IsDir() ||
strings.HasPrefix(info.Name(), ".") ||
!fs.hasSupportedSuffix(info.Name()) {
return nil
}

blueprint, err := fs.readBlueprintFile(path)

if err != nil {
log.Printf("cannot read file %s: %s", path, err)
return nil
}

fs.cacheThis(blueprint)

return nil
})
}

func (fs *FileSystem) readBlueprintFile(blueprintFile string) (*core.Blueprint, error) {
b, err := ioutil.ReadFile(blueprintFile)
if err != nil {
Expand Down

0 comments on commit 432e1f0

Please sign in to comment.