Line data Source code
1 : // Copyright 2021 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 atomicfs
6 :
7 : import (
8 : "fmt"
9 : "strconv"
10 : "strings"
11 :
12 : "github.com/cockroachdb/errors"
13 : "github.com/cockroachdb/errors/oserror"
14 : "github.com/cockroachdb/pebble/vfs"
15 : )
16 :
17 : // ReadMarker looks up the current state of a marker returning just the
18 : // current value of the marker. Callers that may need to move the marker
19 : // to a new value should use LocateMarker.
20 1 : func ReadMarker(fs vfs.FS, dir, markerName string) (string, error) {
21 1 : state, err := scanForMarker(fs, dir, markerName)
22 1 : if err != nil {
23 0 : return "", err
24 0 : }
25 1 : return state.value, nil
26 : }
27 :
28 : // LocateMarker loads the current state of a marker. It returns a handle
29 : // to the Marker that may be used to move the marker and the
30 : // current value of the marker.
31 2 : func LocateMarker(fs vfs.FS, dir, markerName string) (*Marker, string, error) {
32 2 : state, err := scanForMarker(fs, dir, markerName)
33 2 : if err != nil {
34 1 : return nil, "", err
35 1 : }
36 2 : dirFD, err := fs.OpenDir(dir)
37 2 : if err != nil {
38 1 : return nil, "", err
39 1 : }
40 2 : return &Marker{
41 2 : fs: fs,
42 2 : dir: dir,
43 2 : dirFD: dirFD,
44 2 : name: markerName,
45 2 : filename: state.filename,
46 2 : iter: state.iter,
47 2 : obsoleteFiles: state.obsolete,
48 2 : }, state.value, nil
49 : }
50 :
51 : type scannedState struct {
52 : // filename is the latest marker file found (the one with the highest iter value).
53 : filename string
54 : iter uint64
55 : value string
56 : // obsolete is a list of earlier markers that were found.
57 : obsolete []string
58 : }
59 :
60 2 : func scanForMarker(fs vfs.FS, dir, markerName string) (scannedState, error) {
61 2 : ls, err := fs.List(dir)
62 2 : if err != nil {
63 1 : return scannedState{}, err
64 1 : }
65 2 : var state scannedState
66 2 : for _, filename := range ls {
67 2 : if !strings.HasPrefix(filename, `marker.`) {
68 2 : continue
69 : }
70 : // Any filenames with the `marker.` prefix are required to be
71 : // well-formed and parse as markers.
72 2 : name, iter, value, err := parseMarkerFilename(filename)
73 2 : if err != nil {
74 0 : return scannedState{}, err
75 0 : }
76 2 : if name != markerName {
77 2 : continue
78 : }
79 :
80 2 : if state.filename == "" || state.iter < iter {
81 2 : if state.filename != "" {
82 1 : state.obsolete = append(state.obsolete, state.filename)
83 1 : }
84 2 : state.filename = filename
85 2 : state.iter = iter
86 2 : state.value = value
87 1 : } else {
88 1 : state.obsolete = append(state.obsolete, filename)
89 1 : }
90 : }
91 2 : return state, nil
92 : }
93 :
94 : // A Marker provides an interface for maintaining a single string value on the
95 : // filesystem. The marker may be atomically moved from value to value.
96 : //
97 : // The implementation creates a new marker file for each new value, embedding
98 : // the value in the marker filename.
99 : //
100 : // Marker is not safe for concurrent use. Multiple processes may not read or
101 : // move the same marker simultaneously. A Marker may only be constructed through
102 : // LocateMarker.
103 : //
104 : // Marker names must be unique within the directory.
105 : type Marker struct {
106 : fs vfs.FS
107 : dir string
108 : dirFD vfs.File
109 : // name identifies the marker.
110 : name string
111 : // filename contains the entire filename of the current marker. It
112 : // has a format of `marker.<name>.<iter>.<value>`. It's not
113 : // necessarily in sync with iter, since filename is only updated
114 : // when the marker is successfully moved.
115 : filename string
116 : // iter holds the current iteration value. It matches the iteration
117 : // value encoded within filename, if filename is non-empty. Iter is
118 : // monotonically increasing over the lifetime of a marker. Actual
119 : // marker files will always have a positive iter value.
120 : iter uint64
121 : // obsoleteFiles holds a list of marker files discovered by LocateMarker that
122 : // are old values for this marker. These files may exist in certain error
123 : // cases or crashes (e.g. if the deletion of the previous marker file failed
124 : // during Move).
125 : obsoleteFiles []string
126 : }
127 :
128 2 : func markerFilename(name string, iter uint64, value string) string {
129 2 : return fmt.Sprintf("marker.%s.%06d.%s", name, iter, value)
130 2 : }
131 :
132 2 : func parseMarkerFilename(s string) (name string, iter uint64, value string, err error) {
133 2 : // Check for and remove the `marker.` prefix.
134 2 : if !strings.HasPrefix(s, `marker.`) {
135 1 : return "", 0, "", errors.Newf("invalid marker filename: %q", s)
136 1 : }
137 2 : s = s[len(`marker.`):]
138 2 :
139 2 : // Extract the marker's name.
140 2 : i := strings.IndexByte(s, '.')
141 2 : if i == -1 {
142 0 : return "", 0, "", errors.Newf("invalid marker filename: %q", s)
143 0 : }
144 2 : name = s[:i]
145 2 : s = s[i+1:]
146 2 :
147 2 : // Extract the marker's iteration number.
148 2 : i = strings.IndexByte(s, '.')
149 2 : if i == -1 {
150 0 : return "", 0, "", errors.Newf("invalid marker filename: %q", s)
151 0 : }
152 2 : iter, err = strconv.ParseUint(s[:i], 10, 64)
153 2 : if err != nil {
154 1 : return "", 0, "", errors.Newf("invalid marker filename: %q", s)
155 1 : }
156 :
157 : // Everything after the iteration's `.` delimiter is the value.
158 2 : s = s[i+1:]
159 2 :
160 2 : return name, iter, s, nil
161 : }
162 :
163 : // Close releases all resources in use by the marker.
164 2 : func (a *Marker) Close() error {
165 2 : return a.dirFD.Close()
166 2 : }
167 :
168 : // Move atomically moves the marker to a new value.
169 : //
170 : // If Move returns a nil error, the new marker value is guaranteed to be
171 : // persisted to stable storage. If Move returns an error, the current
172 : // value of the marker may be the old value or the new value. Callers
173 : // may retry a Move error.
174 : //
175 : // If an error occurs while syncing the directory, Move panics.
176 2 : func (a *Marker) Move(newValue string) error {
177 2 : a.iter++
178 2 : dstFilename := markerFilename(a.name, a.iter, newValue)
179 2 : dstPath := a.fs.PathJoin(a.dir, dstFilename)
180 2 : oldFilename := a.filename
181 2 :
182 2 : // Create the new marker.
183 2 : f, err := a.fs.Create(dstPath)
184 2 : if err != nil {
185 1 : // On a distributed filesystem, an error doesn't guarantee that
186 1 : // the file wasn't created. A retry of the same Move call will
187 1 : // use a new iteration value, and try to a create a new file. If
188 1 : // the errored invocation was actually successful in creating
189 1 : // the file, we'll leak a file. That's okay, because the next
190 1 : // time the marker is located we'll add it to the obsolete files
191 1 : // list.
192 1 : //
193 1 : // Note that the unconditional increment of `a.iter` means that
194 1 : // `a.iter` and `a.filename` are not necessarily in sync,
195 1 : // because `a.filename` is only updated on success.
196 1 : return err
197 1 : }
198 2 : a.filename = dstFilename
199 2 : if err := f.Close(); err != nil {
200 0 : return err
201 0 : }
202 :
203 : // Remove the now defunct file. If an error is surfaced, we record
204 : // the file as an obsolete file. The file's presence does not
205 : // affect correctness, and it will be cleaned up the next time
206 : // RemoveObsolete is called, either by this process or the next.
207 2 : if oldFilename != "" {
208 2 : if err := a.fs.Remove(a.fs.PathJoin(a.dir, oldFilename)); err != nil && !oserror.IsNotExist(err) {
209 1 : a.obsoleteFiles = append(a.obsoleteFiles, oldFilename)
210 1 : }
211 : }
212 :
213 : // Sync the directory to ensure marker movement is synced.
214 2 : if err := a.dirFD.Sync(); err != nil {
215 0 : // Fsync errors are unrecoverable.
216 0 : // See https://wiki.postgresql.org/wiki/Fsync_Errors and
217 0 : // https://danluu.com/fsyncgate.
218 0 : panic(errors.WithStack(err))
219 : }
220 2 : return nil
221 : }
222 :
223 : // NextIter returns the next iteration number that the marker will use.
224 : // Clients may use this number for formulating new values that are
225 : // unused.
226 2 : func (a *Marker) NextIter() uint64 {
227 2 : return a.iter + 1
228 2 : }
229 :
230 : // RemoveObsolete removes any obsolete files discovered while locating
231 : // the marker or files unable to be removed during Move.
232 2 : func (a *Marker) RemoveObsolete() error {
233 2 : for i, filename := range a.obsoleteFiles {
234 1 : if err := a.fs.Remove(a.fs.PathJoin(a.dir, filename)); err != nil && !oserror.IsNotExist(err) {
235 0 : a.obsoleteFiles = a.obsoleteFiles[i:]
236 0 : return err
237 0 : }
238 : }
239 2 : a.obsoleteFiles = nil
240 2 : return nil
241 : }
|
