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           2 : func (p *provider) vfsPath(fileType base.FileType, fileNum base.DiskFileNum) string {
      17           2 :         return base.MakeFilepath(p.st.FS, p.st.FSDirName, fileType, fileNum)
      18           2 : }
      19             : 
      20             : func (p *provider) vfsOpenForReading(
      21             :         ctx context.Context,
      22             :         fileType base.FileType,
      23             :         fileNum base.DiskFileNum,
      24             :         opts objstorage.OpenOptions,
      25           2 : ) (objstorage.Readable, error) {
      26           2 :         filename := p.vfsPath(fileType, fileNum)
      27           2 :         file, err := p.st.FS.Open(filename, vfs.RandomReadsOption)
      28           2 :         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           2 :         return newFileReadable(file, p.st.FS, filename)
      35             : }
      36             : 
      37             : func (p *provider) vfsCreate(
      38             :         _ context.Context, fileType base.FileType, fileNum base.DiskFileNum,
      39           2 : ) (objstorage.Writable, objstorage.ObjectMetadata, error) {
      40           2 :         filename := p.vfsPath(fileType, fileNum)
      41           2 :         file, err := p.st.FS.Create(filename)
      42           2 :         if err != nil {
      43           1 :                 return nil, objstorage.ObjectMetadata{}, err
      44           1 :         }
      45           2 :         file = vfs.NewSyncingFile(file, vfs.SyncingFileOptions{
      46           2 :                 NoSyncOnClose: p.st.NoSyncOnClose,
      47           2 :                 BytesPerSync:  p.st.BytesPerSync,
      48           2 :         })
      49           2 :         meta := objstorage.ObjectMetadata{
      50           2 :                 DiskFileNum: fileNum,
      51           2 :                 FileType:    fileType,
      52           2 :         }
      53           2 :         return newFileBufferedWritable(file), meta, nil
      54             : }
      55             : 
      56           2 : func (p *provider) vfsRemove(fileType base.FileType, fileNum base.DiskFileNum) error {
      57           2 :         return p.st.FSCleaner.Clean(p.st.FS, fileType, p.vfsPath(fileType, fileNum))
      58           2 : }
      59             : 
      60             : // vfsInit finds any local FS objects.
      61           2 : func (p *provider) vfsInit() error {
      62           2 :         listing := p.st.FSDirInitialListing
      63           2 :         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           2 :         for _, filename := range listing {
      72           2 :                 fileType, fileNum, ok := base.ParseFilename(p.st.FS, filename)
      73           2 :                 if ok && fileType == base.FileTypeTable {
      74           2 :                         o := objstorage.ObjectMetadata{
      75           2 :                                 FileType:    fileType,
      76           2 :                                 DiskFileNum: fileNum,
      77           2 :                         }
      78           2 :                         p.mu.knownObjects[o.DiskFileNum] = o
      79           2 :                 }
      80             :         }
      81           2 :         return nil
      82             : }
      83             : 
      84           2 : func (p *provider) vfsSync() error {
      85           2 :         p.mu.Lock()
      86           2 :         shouldSync := p.mu.localObjectsChanged
      87           2 :         p.mu.localObjectsChanged = false
      88           2 :         p.mu.Unlock()
      89           2 : 
      90           2 :         if !shouldSync {
      91           2 :                 return nil
      92           2 :         }
      93           2 :         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           2 :         return nil
     100             : }
     101             : 
     102           2 : func (p *provider) vfsSize(fileType base.FileType, fileNum base.DiskFileNum) (int64, error) {
     103           2 :         filename := p.vfsPath(fileType, fileNum)
     104           2 :         stat, err := p.st.FS.Stat(filename)
     105           2 :         if err != nil {
     106           0 :                 return 0, err
     107           0 :         }
     108           2 :         return stat.Size(), nil
     109             : }