mirrors/stfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement disk-based IO cache

Browse Source
This commit is contained in:
Felicitas Pojtinger
2021-12-23 22:12:29 +01:00
parent 70aad4db5d
commit 31459e0efd
4 changed files with 93 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
cmd/stfs/cmd/serve_ftp.go
View File

@@ -3,6 +3,7 @@ package cmd
import (
"context"
"fmt"
"io/ioutil"
"log"
"os"
"path/filepath"
@@ -18,6 +19,7 @@ import (
"github.com/pojntfx/stfs/pkg/config"
"github.com/pojntfx/stfs/pkg/operations"
"github.com/pojntfx/stfs/pkg/tape"
"github.com/spf13/afero"
"github.com/spf13/cobra"
"github.com/spf13/viper"
)
@@ -32,6 +34,10 @@ const (
signatureRecipientFlag = "signature-recipient"
)
var (
cacheDir = filepath.Join(os.TempDir(), "stfs")
)
var serveFTPCmd = &cobra.Command{
Use: "ftp",
Aliases: []string{"f"},
@@ -187,6 +193,22 @@ var serveFTPCmd = &cobra.Command{
},
viper.GetString(compressionLevelFlag),
func() (afero.File, func() error, error) {
tmpdir := filepath.Join(viper.GetString(cacheDirFlag), "io")
if err := os.MkdirAll(tmpdir, os.ModePerm); err != nil {
return nil, nil, err
}
f, err := ioutil.TempFile(tmpdir, "*")
if err != nil {
return nil, nil, err
}
return f, func() error {
return os.Remove(filepath.Join(tmpdir, f.Name()))
}, nil
},
logger.PrintHeader,
)
@@ -196,7 +218,7 @@ var serveFTPCmd = &cobra.Command{
root,
viper.GetString(cacheFlag),
viper.GetDuration(cacheDurationFlag),
viper.GetString(cacheDirFlag),
filepath.Join(viper.GetString(cacheDirFlag), "cache"),
)
if err != nil {
return err
@@ -236,7 +258,7 @@ func init() {
serveFTPCmd.PersistentFlags().StringP(laddrFlag, "a", "localhost:1337", "Listen address")
serveFTPCmd.PersistentFlags().StringP(cacheFlag, "n", config.NoneKey, fmt.Sprintf("Cache to use (default %v, available are %v)", config.NoneKey, cache.KnownCacheTypes))
serveFTPCmd.PersistentFlags().DurationP(cacheDurationFlag, "u", time.Hour, "Duration until cache is invalidated")
serveFTPCmd.PersistentFlags().StringP(cacheDirFlag, "w", filepath.Join(os.TempDir(), "stfs", "cache"), "Directory to use if dir cache is enabled")
serveFTPCmd.PersistentFlags().StringP(cacheDirFlag, "w", cacheDir, "Directory to use if dir cache is enabled")
viper.AutomaticEnv()

21
cmd/stfs/cmd/serve_http.go
View File

@@ -3,6 +3,7 @@ package cmd
import (
"context"
"fmt"
"io/ioutil"
"log"
"net/http"
"os"
@@ -127,6 +128,22 @@ var serveHTTPCmd = &cobra.Command{
},
"", // We never write
func() (afero.File, func() error, error) {
tmpdir := filepath.Join(viper.GetString(cacheDirFlag), "io")
if err := os.MkdirAll(tmpdir, os.ModePerm); err != nil {
return nil, nil, err
}
f, err := ioutil.TempFile(tmpdir, "*")
if err != nil {
return nil, nil, err
}
return f, func() error {
return os.Remove(filepath.Join(tmpdir, f.Name()))
}, nil
},
logger.PrintHeader,
)
@@ -136,7 +153,7 @@ var serveHTTPCmd = &cobra.Command{
root,
viper.GetString(cacheFlag),
viper.GetDuration(cacheDurationFlag),
viper.GetString(cacheDirFlag),
filepath.Join(viper.GetString(cacheDirFlag), "cache"),
)
if err != nil {
return err
@@ -163,7 +180,7 @@ func init() {
serveHTTPCmd.PersistentFlags().StringP(laddrFlag, "a", "localhost:1337", "Listen address")
serveHTTPCmd.PersistentFlags().StringP(cacheFlag, "n", config.NoneKey, fmt.Sprintf("Cache to use (default %v, available are %v)", config.NoneKey, cache.KnownCacheTypes))
serveHTTPCmd.PersistentFlags().DurationP(cacheDurationFlag, "u", time.Hour, "Duration until cache is invalidated")
serveHTTPCmd.PersistentFlags().StringP(cacheDirFlag, "w", filepath.Join(os.TempDir(), "stfs", "cache"), "Directory to use if dir cache is enabled")
serveHTTPCmd.PersistentFlags().StringP(cacheDirFlag, "w", cacheDir, "Directory to use if dir cache is enabled")
viper.AutomaticEnv()

48
internal/fs/file.go
View File

@@ -9,7 +9,6 @@ import (
"os"
"sync"
"github.com/mattetti/filebuffer"
models "github.com/pojntfx/stfs/internal/db/sqlite/models/metadata"
"github.com/pojntfx/stfs/internal/ioext"
"github.com/pojntfx/stfs/pkg/config"
@@ -32,7 +31,9 @@ type File struct {
path string
link string
compressionLevel string
getFileBuffer func() (afero.File, func() error, error)
name string
info os.FileInfo
@@ -43,7 +44,8 @@ type File struct {
readOpWriter io.WriteCloser
// TODO: Find a non-in-memory method to do this
writeBuf *filebuffer.Buffer
writeBuf afero.File
cleanWriteBuf func() error
onHeader func(hdr *models.Header)
}
@@ -56,7 +58,9 @@ func NewFile(
path string,
link string,
compressionLevel string,
getFileBuffer func() (afero.File, func() error, error),
name string,
info os.FileInfo,
@@ -71,7 +75,9 @@ func NewFile(
path: path,
link: link,
compressionLevel: compressionLevel,
getFileBuffer: getFileBuffer,
name: name,
info: info,
@@ -148,9 +154,14 @@ func (f *File) syncWithoutLocking() error {
}
done = true
stat, err := f.writeBuf.Stat()
if err != nil {
return config.FileConfig{}, err
}
f.info = &FileInfo{
name: f.info.Name(),
size: int64(f.writeBuf.Buff.Len()),
size: stat.Size(),
mode: f.info.Mode(),
modTime: f.info.ModTime(),
isDir: f.info.IsDir(),
@@ -197,12 +208,16 @@ func (f *File) Truncate(size int64) error {
defer f.ioLock.Unlock()
if f.writeBuf == nil {
f.writeBuf = filebuffer.New([]byte{})
writeBuf, cleanWriteBuf, err := f.getFileBuffer()
if err != nil {
return err
}
f.writeBuf.Buff.Truncate(f.writeBuf.Buff.Len())
f.writeBuf = writeBuf
f.cleanWriteBuf = cleanWriteBuf
}
return nil
return f.writeBuf.Truncate(0)
}
func (f *File) WriteString(s string) (ret int, err error) {
@@ -235,6 +250,10 @@ func (f *File) closeWithoutLocking() error {
if err := f.syncWithoutLocking(); err != nil {
return err
}
// if err := f.writeBuf.Del os.Remove(f.writeBuf.Name()); err != nil {
// return err
// }
}
f.readOpReader = nil
@@ -409,10 +428,23 @@ func (f *File) Write(p []byte) (n int, err error) {
defer f.ioLock.Unlock()
if f.writeBuf == nil {
f.writeBuf = filebuffer.New([]byte{})
writeBuf, cleanWriteBuf, err := f.getFileBuffer()
if err != nil {
return -1, err
}
return f.writeBuf.Write(p)
f.writeBuf = writeBuf
f.cleanWriteBuf = cleanWriteBuf
}
n, err = f.writeBuf.Write(p)
if err != nil {
log.Fatal(err)
return -1, err
}
return n, f.writeBuf.Sync()
}
func (f *File) WriteAt(p []byte, off int64) (n int, err error) {

5
internal/fs/filesystem.go
View File

@@ -31,6 +31,7 @@ type FileSystem struct {
metadata config.MetadataConfig
compressionLevel string
getFileBuffer func() (afero.File, func() error, error)
onHeader func(hdr *models.Header)
}
@@ -42,6 +43,7 @@ func NewFileSystem(
metadata config.MetadataConfig,
compressionLevel string,
getFileBuffer func() (afero.File, func() error, error),
onHeader func(hdr *models.Header),
) afero.Fs {
@@ -52,6 +54,7 @@ func NewFileSystem(
metadata: metadata,
compressionLevel: compressionLevel,
getFileBuffer: getFileBuffer,
onHeader: onHeader,
}
@@ -194,7 +197,9 @@ func (f *FileSystem) Open(name string) (afero.File, error) {
hdr.Name,
hdr.Linkname,
f.compressionLevel,
f.getFileBuffer,
path.Base(hdr.Name),
NewFileInfo(hdr),