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, 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 2 : ) (objstorage.Writable, objstorage.ObjectMetadata, error) {
43 2 : filename := p.vfsPath(fileType, fileNum)
44 2 : file, err := p.st.FS.Create(filename, category)
45 2 : if err != nil {
46 1 : return nil, objstorage.ObjectMetadata{}, err
47 1 : }
48 2 : file = vfs.NewSyncingFile(file, vfs.SyncingFileOptions{
49 2 : NoSyncOnClose: p.st.NoSyncOnClose,
50 2 : BytesPerSync: p.st.BytesPerSync,
51 2 : })
52 2 : meta := objstorage.ObjectMetadata{
53 2 : DiskFileNum: fileNum,
54 2 : FileType: fileType,
55 2 : }
56 2 : return newFileBufferedWritable(file), meta, nil
57 : }
58 :
59 2 : func (p *provider) vfsRemove(fileType base.FileType, fileNum base.DiskFileNum) error {
60 2 : return p.st.FSCleaner.Clean(p.st.FS, fileType, p.vfsPath(fileType, fileNum))
61 2 : }
62 :
63 : // vfsInit finds any local FS objects.
64 2 : func (p *provider) vfsInit() error {
65 2 : listing := p.st.FSDirInitialListing
66 2 : 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 2 : for _, filename := range listing {
75 2 : fileType, fileNum, ok := base.ParseFilename(p.st.FS, filename)
76 2 : if ok && fileType == base.FileTypeTable {
77 2 : o := objstorage.ObjectMetadata{
78 2 : FileType: fileType,
79 2 : DiskFileNum: fileNum,
80 2 : }
81 2 : p.mu.knownObjects[o.DiskFileNum] = o
82 2 : }
83 : }
84 2 : return nil
85 : }
86 :
87 2 : func (p *provider) vfsSync() error {
88 2 : p.mu.Lock()
89 2 : shouldSync := p.mu.localObjectsChanged
90 2 : p.mu.localObjectsChanged = false
91 2 : p.mu.Unlock()
92 2 :
93 2 : if !shouldSync {
94 2 : return nil
95 2 : }
96 2 : if err := p.fsDir.Sync(); err != nil {
97 1 : p.mu.Lock()
98 1 : defer p.mu.Unlock()
99 1 : p.mu.localObjectsChanged = true
100 1 : return err
101 1 : }
102 2 : return nil
103 : }
104 :
105 2 : func (p *provider) vfsSize(fileType base.FileType, fileNum base.DiskFileNum) (int64, error) {
106 2 : filename := p.vfsPath(fileType, fileNum)
107 2 : stat, err := p.st.FS.Stat(filename)
108 2 : if err != nil {
109 0 : return 0, err
110 0 : }
111 2 : return stat.Size(), nil
112 : }
|