Line data    Source code

       1             : // Copyright 2023 The LevelDB-Go and Pebble Authors. All rights reserved. Use
       2             : // of this source code is governed by a BSD-style license that can be found in
       3             : // the LICENSE file.
       4             : 
       5             : package objstorageprovider
       6             : 
       7             : import (
       8             :         "context"
       9             : 
      10             :         "github.com/cockroachdb/errors"
      11             :         "github.com/cockroachdb/pebble/internal/base"
      12             :         "github.com/cockroachdb/pebble/objstorage"
      13             :         "github.com/cockroachdb/pebble/vfs"
      14             : )
      15             : 
      16           1 : func (p *provider) vfsPath(fileType base.FileType, fileNum base.DiskFileNum) string {
      17           1 :         return base.MakeFilepath(p.st.FS, p.st.FSDirName, fileType, fileNum)
      18           1 : }
      19             : 
      20             : func (p *provider) vfsOpenForReading(
      21             :         ctx context.Context,
      22             :         fileType base.FileType,
      23             :         fileNum base.DiskFileNum,
      24             :         opts objstorage.OpenOptions,
      25           1 : ) (objstorage.Readable, error) {
      26           1 :         filename := p.vfsPath(fileType, fileNum)
      27           1 :         file, err := p.st.FS.Open(filename, vfs.RandomReadsOption)
      28           1 :         if err != nil {
      29           1 :                 if opts.MustExist {
      30           1 :                         base.MustExist(p.st.FS, filename, p.st.Logger, err)
      31           1 :                 }
      32           1 :                 return nil, err
      33             :         }
      34           1 :         return newFileReadable(file, p.st.FS, filename)
      35             : }
      36             : 
      37             : func (p *provider) vfsCreate(
      38             :         _ context.Context, fileType base.FileType, fileNum base.DiskFileNum,
      39           1 : ) (objstorage.Writable, objstorage.ObjectMetadata, error) {
      40           1 :         filename := p.vfsPath(fileType, fileNum)
      41           1 :         file, err := p.st.FS.Create(filename)
      42           1 :         if err != nil {
      43           1 :                 return nil, objstorage.ObjectMetadata{}, err
      44           1 :         }
      45           1 :         file = vfs.NewSyncingFile(file, vfs.SyncingFileOptions{
      46           1 :                 NoSyncOnClose: p.st.NoSyncOnClose,
      47           1 :                 BytesPerSync:  p.st.BytesPerSync,
      48           1 :         })
      49           1 :         meta := objstorage.ObjectMetadata{
      50           1 :                 DiskFileNum: fileNum,
      51           1 :                 FileType:    fileType,
      52           1 :         }
      53           1 :         return newFileBufferedWritable(file), meta, nil
      54             : }
      55             : 
      56           1 : func (p *provider) vfsRemove(fileType base.FileType, fileNum base.DiskFileNum) error {
      57           1 :         return p.st.FSCleaner.Clean(p.st.FS, fileType, p.vfsPath(fileType, fileNum))
      58           1 : }
      59             : 
      60             : // vfsInit finds any local FS objects.
      61           1 : func (p *provider) vfsInit() error {
      62           1 :         listing := p.st.FSDirInitialListing
      63           1 :         if listing == nil {
      64           1 :                 var err error
      65           1 :                 listing, err = p.st.FS.List(p.st.FSDirName)
      66           1 :                 if err != nil {
      67           0 :                         return errors.Wrapf(err, "pebble: could not list store directory")
      68           0 :                 }
      69             :         }
      70             : 
      71           1 :         for _, filename := range listing {
      72           1 :                 fileType, fileNum, ok := base.ParseFilename(p.st.FS, filename)
      73           1 :                 if ok && fileType == base.FileTypeTable {
      74           1 :                         o := objstorage.ObjectMetadata{
      75           1 :                                 FileType:    fileType,
      76           1 :                                 DiskFileNum: fileNum,
      77           1 :                         }
      78           1 :                         p.mu.knownObjects[o.DiskFileNum] = o
      79           1 :                 }
      80             :         }
      81           1 :         return nil
      82             : }
      83             : 
      84           1 : func (p *provider) vfsSync() error {
      85           1 :         p.mu.Lock()
      86           1 :         shouldSync := p.mu.localObjectsChanged
      87           1 :         p.mu.localObjectsChanged = false
      88           1 :         p.mu.Unlock()
      89           1 : 
      90           1 :         if !shouldSync {
      91           1 :                 return nil
      92           1 :         }
      93           1 :         if err := p.fsDir.Sync(); err != nil {
      94           1 :                 p.mu.Lock()
      95           1 :                 defer p.mu.Unlock()
      96           1 :                 p.mu.localObjectsChanged = true
      97           1 :                 return err
      98           1 :         }
      99           1 :         return nil
     100             : }
     101             : 
     102           1 : func (p *provider) vfsSize(fileType base.FileType, fileNum base.DiskFileNum) (int64, error) {
     103           1 :         filename := p.vfsPath(fileType, fileNum)
     104           1 :         stat, err := p.st.FS.Stat(filename)
     105           1 :         if err != nil {
     106           0 :                 return 0, err
     107           0 :         }
     108           1 :         return stat.Size(), nil
     109             : }