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, p.st.Local.ReadaheadConfig, filename)
35 : }
36 :
37 : func (p *provider) vfsCreate(
38 : _ context.Context,
39 : fileType base.FileType,
40 : fileNum base.DiskFileNum,
41 : category vfs.DiskWriteCategory,
42 1 : ) (objstorage.Writable, objstorage.ObjectMetadata, error) {
43 1 : filename := p.vfsPath(fileType, fileNum)
44 1 : file, err := p.st.FS.Create(filename, category)
45 1 : if err != nil {
46 1 : return nil, objstorage.ObjectMetadata{}, err
47 1 : }
48 1 : file = vfs.NewSyncingFile(file, vfs.SyncingFileOptions{
49 1 : NoSyncOnClose: p.st.NoSyncOnClose,
50 1 : BytesPerSync: p.st.BytesPerSync,
51 1 : })
52 1 : meta := objstorage.ObjectMetadata{
53 1 : DiskFileNum: fileNum,
54 1 : FileType: fileType,
55 1 : }
56 1 : return newFileBufferedWritable(file), meta, nil
57 : }
58 :
59 1 : func (p *provider) vfsRemove(fileType base.FileType, fileNum base.DiskFileNum) error {
60 1 : return p.st.FSCleaner.Clean(p.st.FS, fileType, p.vfsPath(fileType, fileNum))
61 1 : }
62 :
63 : // vfsInit finds any local FS objects.
64 1 : func (p *provider) vfsInit() error {
65 1 : listing := p.st.FSDirInitialListing
66 1 : if listing == nil {
67 1 : var err error
68 1 : listing, err = p.st.FS.List(p.st.FSDirName)
69 1 : if err != nil {
70 0 : return errors.Wrapf(err, "pebble: could not list store directory")
71 0 : }
72 : }
73 :
74 1 : for _, filename := range listing {
75 1 : fileType, fileNum, ok := base.ParseFilename(p.st.FS, filename)
76 1 : if ok && fileType == base.FileTypeTable {
77 1 : o := objstorage.ObjectMetadata{
78 1 : FileType: fileType,
79 1 : DiskFileNum: fileNum,
80 1 : }
81 1 : p.mu.knownObjects[o.DiskFileNum] = o
82 1 : }
83 : }
84 1 : return nil
85 : }
86 :
87 1 : func (p *provider) vfsSync() error {
88 1 : p.mu.Lock()
89 1 : counterVal := p.mu.localObjectsChangeCounter
90 1 : lastSynced := p.mu.localObjectsChangeCounterSynced
91 1 : p.mu.Unlock()
92 1 :
93 1 : if lastSynced >= counterVal {
94 1 : return nil
95 1 : }
96 1 : if err := p.fsDir.Sync(); err != nil {
97 1 : return err
98 1 : }
99 :
100 1 : p.mu.Lock()
101 1 : if p.mu.localObjectsChangeCounterSynced < counterVal {
102 1 : p.mu.localObjectsChangeCounterSynced = counterVal
103 1 : }
104 1 : p.mu.Unlock()
105 1 :
106 1 : return nil
107 : }
108 :
109 1 : func (p *provider) vfsSize(fileType base.FileType, fileNum base.DiskFileNum) (int64, error) {
110 1 : filename := p.vfsPath(fileType, fileNum)
111 1 : stat, err := p.st.FS.Stat(filename)
112 1 : if err != nil {
113 0 : return 0, err
114 0 : }
115 1 : return stat.Size(), nil
116 : }
|