From 4ab05592633ae75de5a39d16495f57eb6ea263a2 Mon Sep 17 00:00:00 2001 From: jakka Date: Sat, 12 Apr 2025 19:30:05 +0300 Subject: reimplemented database in badger so it actually works --- files/database.go | 250 ++++++++++++++++++++++++++++++++++------------------ files/index.go | 56 ------------ files/operations.go | 143 ++++++++++++++++++++++++++++++ files/types.go | 6 +- 4 files changed, 309 insertions(+), 146 deletions(-) delete mode 100644 files/index.go create mode 100644 files/operations.go (limited to 'files') diff --git a/files/database.go b/files/database.go index 24bd5af..1ad40aa 100644 --- a/files/database.go +++ b/files/database.go @@ -2,125 +2,201 @@ package files import ( "context" - "crypto/sha256" "encoding/json" - "fmt" - "io" - "io/fs" - "os" - "os/exec" - "path/filepath" - "regexp" - "strings" - - "github.com/tidwall/buntdb" -) + "time" -func evaluateFile(database *buntdb.DB, hashsum string, filedata FileInfo) error { - return database.View(func(tx *buntdb.Tx) error { - data, err := tx.Get(hashsum) - if err != nil { - return err - } + badger "github.com/dgraph-io/badger/v4" + "github.com/dgraph-io/badger/v4/options" + "golang.org/x/sync/errgroup" +) - var info FileInfo - if err := json.Unmarshal([]byte(data), &info); err != nil { +func decodeDbInfo(item *badger.Item) (*FileInfo, error) { + var info FileInfo + if err := item.Value(func(val []byte) error { + if err := json.Unmarshal(val, &info); err != nil { return err } - if info == filedata { - return UpToDate - } - switch info.Modtime == filedata.Modtime { - case true: - if info.AbsPath == filedata.AbsPath { - if info.Process { - return NeedsReencode - } - return UpToDate - } - if info.Process { - return NeedsReencode - } - return MovedFile - default: - return nil - } - }) -} - -func getEncoderVer(path string) (string, error) { - out, err := exec.Command("metaflac", "--show-vendor-tag", path).Output() - if err != nil { - return "", err - } - - r := regexp.MustCompile("libFLAC \\d\\.\\d\\.\\d") - encoder := r.FindString(string(out)) - switch encoder { - case "": - return "", nil - default: - return strings.Split(encoder, " ")[1], nil + return nil + }); err != nil { + return nil, err } + return &info, nil } -func updateFile(database *buntdb.DB, hashsum string, filedata FileInfo) error { - return database.Update(func(tx *buntdb.Tx) error { - data, err := json.Marshal(filedata) +func getInfoFromDb(database *badger.DB, hashsum []byte) (*FileInfo, error) { + var info *FileInfo + if err := database.View(func(txn *badger.Txn) error { + item, err := txn.Get(hashsum) if err != nil { return err } - if _, _, err := tx.Set(hashsum, string(data), nil); err != nil { + + info, err = decodeDbInfo(item) + if err != nil { return err } return nil - }) + }); err != nil { + return nil, err + } + return info, nil } -func ProcessFile(ctx context.Context, path string, info fs.DirEntry) error { - tmp, err := info.Info() +func (filedata *FileInfo) evaluateFile(database *badger.DB, hashsum []byte, encoder string) error { + info, err := getInfoFromDb(database, hashsum) if err != nil { return err } - modtime := tmp.ModTime() - - database := ctx.Value("database").(*buntdb.DB) - - file, err := os.Open(path) - if err != nil { - return err + if info.Process { + return ReencodeNeeded } - defer file.Close() - - hash := sha256.New() - if _, err := io.Copy(hash, file); err != nil { - return err + if filedata.Encoder != encoder { + return ReencodeNeeded } - hashsum := fmt.Sprintf("%x", hash.Sum(nil)) - - abspath, err := filepath.Abs(path) - if err != nil { - return err + if info.AbsPath != filedata.AbsPath { + return FileMoved } + return ReencodeNotNeeded +} - encoder, err := getEncoderVer(path) +func (filedata *FileInfo) updateFile(database *badger.DB, hashsum []byte) error { + return database.Update(func(txn *badger.Txn) error { + encoded, err := json.Marshal(filedata) + if err != nil { + return err + } + return txn.Set(hashsum, encoded) + }) +} - filedata := FileInfo{AbsPath: abspath, Modtime: modtime, Encoder: encoder, Process: true} +func (filedata *FileInfo) IndexFile(ctx context.Context, hashsum []byte) error { + database := ctx.Value("database").(*badger.DB) + encoder := ctx.Value("encoder").(string) - err = evaluateFile(database, hashsum, filedata) + err := filedata.evaluateFile(database, hashsum, encoder) switch err { - case MovedFile, UpToDate: - if filedata.Encoder == ctx.Value("encoder").(string) { - return nil - } - case buntdb.ErrNotFound, NeedsReencode: + case FileMoved, ReencodeNotNeeded: + filedata.Process = false + case ReencodeNeeded, badger.ErrKeyNotFound: default: return err } - if err := updateFile(database, hashsum, filedata); err != nil { + if err := filedata.updateFile(database, hashsum); err != nil { return err } return nil } + +func ReencodeFlacs(ctx context.Context) error { + db := ctx.Value("database").(*badger.DB) + wg := new(errgroup.Group) + wg.SetLimit(4) + err := db.View(func(txn *badger.Txn) error { + opts := badger.DefaultIteratorOptions + opts.PrefetchSize = 10 + it := txn.NewIterator(opts) + defer it.Close() + + for it.Rewind(); it.Valid(); it.Next() { + /* item := it.Item() + key := item.Key() + info, err := decodeDbInfo(item) + if err != nil { + return err + } + if err := info.reencodeFile(ctx); err != nil { + return err + } + if err := info.updateFile(db, key); err != nil { + return err + } */ + select { + case <-ctx.Done(): + return nil + default: + item := it.Item() + key := item.Key() + info, err := decodeDbInfo(item) + if err != nil { + return err + } + wg.Go(func() error { + select { + case <-ctx.Done(): + return nil + default: + if err := info.reencodeFile(ctx); err != nil { + return err + } + if err := info.updateFile(db, key); err != nil { + return err + } + return nil + } + }) + } + } + return nil + }) + wg.Wait() + return err +} + +func Options(path string) badger.Options { + return badger.Options{ + Dir: path, + ValueDir: path, + + MemTableSize: 64 << 20, + BaseTableSize: 2 << 20, + BaseLevelSize: 10 << 20, + TableSizeMultiplier: 2, + LevelSizeMultiplier: 10, + MaxLevels: 7, + NumGoroutines: 8, + MetricsEnabled: true, + + NumCompactors: 4, // Run at least 2 compactors. Zero-th compactor prioritizes L0. + NumLevelZeroTables: 5, + NumLevelZeroTablesStall: 15, + NumMemtables: 5, + BloomFalsePositive: 0.01, + BlockSize: 4 * 1024, + SyncWrites: false, + NumVersionsToKeep: 1, + CompactL0OnClose: false, + VerifyValueChecksum: false, + Compression: options.Snappy, + BlockCacheSize: 256 << 20, + IndexCacheSize: 0, + + // The following benchmarks were done on a 4 KB block size (default block size). The + // compression is ratio supposed to increase with increasing compression level but since the + // input for compression algorithm is small (4 KB), we don't get significant benefit at + // level 3. + // NOTE: The benchmarks are with DataDog ZSTD that requires CGO. Hence, no longer valid. + // no_compression-16 10 502848865 ns/op 165.46 MB/s - + // zstd_compression/level_1-16 7 739037966 ns/op 112.58 MB/s 2.93 + // zstd_compression/level_3-16 7 756950250 ns/op 109.91 MB/s 2.72 + // zstd_compression/level_15-16 1 11135686219 ns/op 7.47 MB/s 4.38 + // Benchmark code can be found in table/builder_test.go file + ZSTDCompressionLevel: 1, + + // (2^30 - 1)*2 when mmapping < 2^31 - 1, max int32. + // -1 so 2*ValueLogFileSize won't overflow on 32-bit systems. + ValueLogFileSize: 1<<30 - 1, + + ValueLogMaxEntries: 1000000, + + VLogPercentile: 0.0, + ValueThreshold: (1 << 20), + + Logger: nil, + EncryptionKey: []byte{}, + EncryptionKeyRotationDuration: 10 * 24 * time.Hour, // Default 10 days. + DetectConflicts: true, + NamespaceOffset: -1, + } +} diff --git a/files/index.go b/files/index.go deleted file mode 100644 index ea05dc4..0000000 --- a/files/index.go +++ /dev/null @@ -1,56 +0,0 @@ -package files - -import ( - "context" - "io/fs" - "path/filepath" - "time" - - "github.com/briandowns/spinner" - "golang.org/x/sync/errgroup" -) - -func IndexFlacs(ctx context.Context) error { - spin := spinner.New(spinner.CharSets[9], 100*time.Millisecond) - spin.Suffix = " Indexing flacs..." - - spin.Start() - - wg := new(errgroup.Group) - wg.SetLimit(100) - - if err := filepath.WalkDir(ctx.Value("path").(string), func(path string, info fs.DirEntry, err error) error { - select { - case <-ctx.Done(): - spin.FinalMSG = "Stopping...\n" - spin.Stop() - return nil - default: - if err != nil { - return err - } - if !info.IsDir() { - if filepath.Ext(path) == ".flac" { - wg.Go(func() error { - select { - case <-ctx.Done(): - return nil - default: - return ProcessFile(ctx, path, info) - } - }) - } - - } - return nil - } - - }); err != nil { - return err - } - - wg.Wait() - spin.FinalMSG = "Done indexing flacs" - spin.Stop() - return nil -} diff --git a/files/operations.go b/files/operations.go new file mode 100644 index 0000000..3da6586 --- /dev/null +++ b/files/operations.go @@ -0,0 +1,143 @@ +package files + +import ( + "context" + "crypto/sha256" + "fmt" + "io" + "io/fs" + "os" + "os/exec" + "path/filepath" + "regexp" + "strings" + "time" + + "github.com/briandowns/spinner" + "golang.org/x/sync/errgroup" +) + +func (file *FileInfo) reencodeFile(ctx context.Context) error { + /* cmd := exec.Command("flac", "-8", "-f", file.AbsPath) + if err := cmd.Run(); err != nil { + log.Printf("%s\n", err.Error()) + return err + } + file.Encoder = ctx.Value("encoder").(string) */ + fmt.Println(file) + return nil +} + +func getEncoderVer(path string) (string, error) { + out, err := exec.Command("metaflac", "--show-vendor-tag", path).Output() + if err != nil { + return "", err + } + + r := regexp.MustCompile("libFLAC \\d\\.\\d\\.\\d") + encoder := r.FindString(string(out)) + switch encoder { + case "": + return "", nil + default: + return strings.Split(encoder, " ")[1], nil + } +} + +func getInfoFromFile(path string, info fs.DirEntry) (*FileInfo, error) { + var filedata FileInfo + + filedata.Process = true + + tmp, err := info.Info() + if err != nil { + return nil, err + } + + filedata.Modtime = tmp.ModTime() + + abspath, err := filepath.Abs(path) + if err != nil { + return nil, err + } + + filedata.AbsPath = abspath + + encoder, err := getEncoderVer(path) + if err != nil { + return nil, err + } + + filedata.Encoder = encoder + + return &filedata, nil +} + +func getSha256(path string) ([]byte, error) { + file, err := os.Open(path) + if err != nil { + return nil, err + } + defer file.Close() + + hash := sha256.New() + if _, err := io.Copy(hash, file); err != nil { + return nil, err + } + return hash.Sum(nil), nil +} + +func IndexFlacs(ctx context.Context) error { + spin := spinner.New(spinner.CharSets[9], 100*time.Millisecond) + spin.Suffix = " Indexing flacs..." + + spin.Start() + + wg := new(errgroup.Group) + wg.SetLimit(100) + + if err := filepath.WalkDir(ctx.Value("path").(string), func(path string, info fs.DirEntry, err error) error { + select { + case <-ctx.Done(): + spin.FinalMSG = "Stopping...\n" + spin.Stop() + return filepath.SkipAll + default: + if err != nil { + return err + } + if !info.IsDir() { + if filepath.Ext(path) == ".flac" { + wg.Go(func() error { + select { + case <-ctx.Done(): + return nil + default: + data, err := getInfoFromFile(path, info) + if err != nil { + return err + } + + hashsum, err := getSha256(path) + if err != nil { + return err + } + + return data.IndexFile(ctx, hashsum) + } + }) + } + + } + return nil + } + + }); err != nil { + return err + } + + wg.Wait() + spin.FinalMSG = "Done indexing flacs\n" + spin.Stop() + return nil +} diff --git a/files/types.go b/files/types.go index 54d8563..87ae6e1 100644 --- a/files/types.go +++ b/files/types.go @@ -6,9 +6,9 @@ import ( ) var ( - UpToDate = errors.New("file is up to date") - MovedFile = errors.New("file was moved") - NeedsReencode = errors.New("file needs to be reencoded") + ReencodeNotNeeded = errors.New("file is up to date") + FileMoved = errors.New("file was moved") + ReencodeNeeded = errors.New("file needs to be reencoded") ) type FileInfo struct { -- cgit v1.3.1