Line data Source code
1 : // Copyright 2020 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 vfs
6 :
7 : import (
8 : "fmt"
9 : "io"
10 : "os"
11 : "path/filepath"
12 : "sync"
13 : "sync/atomic"
14 : "time"
15 :
16 : "github.com/cockroachdb/redact"
17 : )
18 :
19 : const (
20 : // preallocatedSlotCount is the default number of slots available for
21 : // concurrent filesystem operations. The slot count may be exceeded, but
22 : // each additional slot will incur an additional allocation. We choose 16
23 : // here with the expectation that it is significantly more than required in
24 : // practice. See the comment above the diskHealthCheckingFS type definition.
25 : preallocatedSlotCount = 16
26 : // deltaBits is the number of bits in the packed 64-bit integer used for
27 : // identifying a delta from the file creation time in milliseconds.
28 : deltaBits = 40
29 : // writeSizeBits is the number of bits in the packed 64-bit integer used for
30 : // identifying the size of the write operation, if the operation is sized. See
31 : // writeSizePrecision below for precision of size.
32 : writeSizeBits = 20
33 : // Track size of writes at kilobyte precision. See comment above lastWritePacked for more.
34 : writeSizePrecision = 1024
35 : )
36 :
37 : // Variables to enable testing.
38 : var (
39 : // defaultTickInterval is the default interval between two ticks of each
40 : // diskHealthCheckingFile loop iteration.
41 : defaultTickInterval = 2 * time.Second
42 : )
43 :
44 : // OpType is the type of IO operation being monitored by a
45 : // diskHealthCheckingFile.
46 : type OpType uint8
47 :
48 : // The following OpTypes is limited to the subset of file system operations that
49 : // a diskHealthCheckingFile supports (namely writes and syncs).
50 : const (
51 : OpTypeUnknown OpType = iota
52 : OpTypeWrite
53 : OpTypeSync
54 : OpTypeSyncData
55 : OpTypeSyncTo
56 : OpTypeCreate
57 : OpTypeLink
58 : OpTypeMkdirAll
59 : OpTypePreallocate
60 : OpTypeRemove
61 : OpTypeRemoveAll
62 : OpTypeRename
63 : OpTypeReuseForWrite
64 : // Note: opTypeMax is just used in tests. It must appear last in the list
65 : // of OpTypes.
66 : opTypeMax
67 : )
68 :
69 : // String implements fmt.Stringer.
70 0 : func (o OpType) String() string {
71 0 : switch o {
72 0 : case OpTypeWrite:
73 0 : return "write"
74 0 : case OpTypeSync:
75 0 : return "sync"
76 0 : case OpTypeSyncData:
77 0 : return "syncdata"
78 0 : case OpTypeSyncTo:
79 0 : return "syncto"
80 0 : case OpTypeCreate:
81 0 : return "create"
82 0 : case OpTypeLink:
83 0 : return "link"
84 0 : case OpTypeMkdirAll:
85 0 : return "mkdirall"
86 0 : case OpTypePreallocate:
87 0 : return "preallocate"
88 0 : case OpTypeRemove:
89 0 : return "remove"
90 0 : case OpTypeRemoveAll:
91 0 : return "removall"
92 0 : case OpTypeRename:
93 0 : return "rename"
94 0 : case OpTypeReuseForWrite:
95 0 : return "reuseforwrite"
96 0 : case OpTypeUnknown:
97 0 : return "unknown"
98 0 : default:
99 0 : panic(fmt.Sprintf("vfs: unknown op type: %d", o))
100 : }
101 : }
102 :
103 : // diskHealthCheckingFile is a File wrapper to detect slow disk operations, and
104 : // call onSlowDisk if a disk operation is seen to exceed diskSlowThreshold.
105 : //
106 : // This struct creates a goroutine (in startTicker()) that, at every tick
107 : // interval, sees if there's a disk operation taking longer than the specified
108 : // duration. This setup is preferable to creating a new timer at every disk
109 : // operation, as it reduces overhead per disk operation.
110 : type diskHealthCheckingFile struct {
111 : file File
112 : onSlowDisk func(opType OpType, writeSizeInBytes int, duration time.Duration)
113 : diskSlowThreshold time.Duration
114 : tickInterval time.Duration
115 :
116 : stopper chan struct{}
117 : // lastWritePacked is a 64-bit unsigned int. The most significant
118 : // 40 bits represent an delta (in milliseconds) from the creation
119 : // time of the diskHealthCheckingFile. The next most significant 20 bits
120 : // represent the size of the write in KBs, if the write has a size. (If
121 : // it doesn't, the 20 bits are zeroed). The least significant four bits
122 : // contains the OpType.
123 : //
124 : // The use of 40 bits for an delta provides ~34 years of effective
125 : // monitoring time before the uint wraps around, at millisecond precision.
126 : // ~34 years of process uptime "ought to be enough for anybody". Millisecond
127 : // writeSizePrecision is sufficient, given that we are monitoring for writes that take
128 : // longer than one millisecond.
129 : //
130 : // The use of 20 bits for the size in KBs allows representing sizes up
131 : // to nearly one GB. If the write is larger than that, we round down to ~one GB.
132 : //
133 : // The use of four bits for OpType allows for 16 operation types.
134 : //
135 : // NB: this packing scheme is not persisted, and is therefore safe to adjust
136 : // across process boundaries.
137 : lastWritePacked atomic.Uint64
138 : createTimeNanos int64
139 : }
140 :
141 : // newDiskHealthCheckingFile instantiates a new diskHealthCheckingFile, with the
142 : // specified time threshold and event listener.
143 : func newDiskHealthCheckingFile(
144 : file File,
145 : diskSlowThreshold time.Duration,
146 : onSlowDisk func(OpType OpType, writeSizeInBytes int, duration time.Duration),
147 1 : ) *diskHealthCheckingFile {
148 1 : return &diskHealthCheckingFile{
149 1 : file: file,
150 1 : onSlowDisk: onSlowDisk,
151 1 : diskSlowThreshold: diskSlowThreshold,
152 1 : tickInterval: defaultTickInterval,
153 1 :
154 1 : stopper: make(chan struct{}),
155 1 : createTimeNanos: time.Now().UnixNano(),
156 1 : }
157 1 : }
158 :
159 : // startTicker starts a new goroutine with a ticker to monitor disk operations.
160 : // Can only be called if the ticker goroutine isn't running already.
161 1 : func (d *diskHealthCheckingFile) startTicker() {
162 1 : if d.diskSlowThreshold == 0 {
163 0 : return
164 0 : }
165 :
166 1 : go func() {
167 1 : ticker := time.NewTicker(d.tickInterval)
168 1 : defer ticker.Stop()
169 1 :
170 1 : for {
171 1 : select {
172 1 : case <-d.stopper:
173 1 : return
174 :
175 1 : case <-ticker.C:
176 1 : packed := d.lastWritePacked.Load()
177 1 : if packed == 0 {
178 1 : continue
179 : }
180 0 : delta, writeSize, op := unpack(packed)
181 0 : lastWrite := time.Unix(0, d.createTimeNanos+delta.Nanoseconds())
182 0 : now := time.Now()
183 0 : if lastWrite.Add(d.diskSlowThreshold).Before(now) {
184 0 : // diskSlowThreshold was exceeded. Call the passed-in
185 0 : // listener.
186 0 : d.onSlowDisk(op, writeSize, now.Sub(lastWrite))
187 0 : }
188 : }
189 : }
190 : }()
191 : }
192 :
193 : // stopTicker stops the goroutine started in startTicker.
194 1 : func (d *diskHealthCheckingFile) stopTicker() {
195 1 : close(d.stopper)
196 1 : }
197 :
198 : // Fd implements (vfs.File).Fd.
199 1 : func (d *diskHealthCheckingFile) Fd() uintptr {
200 1 : return d.file.Fd()
201 1 : }
202 :
203 : // Read implements (vfs.File).Read
204 0 : func (d *diskHealthCheckingFile) Read(p []byte) (int, error) {
205 0 : return d.file.Read(p)
206 0 : }
207 :
208 : // ReadAt implements (vfs.File).ReadAt
209 0 : func (d *diskHealthCheckingFile) ReadAt(p []byte, off int64) (int, error) {
210 0 : return d.file.ReadAt(p, off)
211 0 : }
212 :
213 : // Write implements the io.Writer interface.
214 1 : func (d *diskHealthCheckingFile) Write(p []byte) (n int, err error) {
215 1 : d.timeDiskOp(OpTypeWrite, int64(len(p)), func() {
216 1 : n, err = d.file.Write(p)
217 1 : }, time.Now().UnixNano())
218 1 : return n, err
219 : }
220 :
221 : // Write implements the io.WriterAt interface.
222 0 : func (d *diskHealthCheckingFile) WriteAt(p []byte, ofs int64) (n int, err error) {
223 0 : d.timeDiskOp(OpTypeWrite, int64(len(p)), func() {
224 0 : n, err = d.file.WriteAt(p, ofs)
225 0 : }, time.Now().UnixNano())
226 0 : return n, err
227 : }
228 :
229 : // Close implements the io.Closer interface.
230 1 : func (d *diskHealthCheckingFile) Close() error {
231 1 : d.stopTicker()
232 1 : return d.file.Close()
233 1 : }
234 :
235 : // Prefetch implements (vfs.File).Prefetch.
236 0 : func (d *diskHealthCheckingFile) Prefetch(offset, length int64) error {
237 0 : return d.file.Prefetch(offset, length)
238 0 : }
239 :
240 : // Preallocate implements (vfs.File).Preallocate.
241 0 : func (d *diskHealthCheckingFile) Preallocate(off, n int64) (err error) {
242 0 : d.timeDiskOp(OpTypePreallocate, n, func() {
243 0 : err = d.file.Preallocate(off, n)
244 0 : }, time.Now().UnixNano())
245 0 : return err
246 : }
247 :
248 : // Stat implements (vfs.File).Stat.
249 0 : func (d *diskHealthCheckingFile) Stat() (os.FileInfo, error) {
250 0 : return d.file.Stat()
251 0 : }
252 :
253 : // Sync implements the io.Syncer interface.
254 1 : func (d *diskHealthCheckingFile) Sync() (err error) {
255 1 : d.timeDiskOp(OpTypeSync, 0, func() {
256 1 : err = d.file.Sync()
257 1 : }, time.Now().UnixNano())
258 1 : return err
259 : }
260 :
261 : // SyncData implements (vfs.File).SyncData.
262 1 : func (d *diskHealthCheckingFile) SyncData() (err error) {
263 1 : d.timeDiskOp(OpTypeSyncData, 0, func() {
264 1 : err = d.file.SyncData()
265 1 : }, time.Now().UnixNano())
266 1 : return err
267 : }
268 :
269 : // SyncTo implements (vfs.File).SyncTo.
270 0 : func (d *diskHealthCheckingFile) SyncTo(length int64) (fullSync bool, err error) {
271 0 : d.timeDiskOp(OpTypeSyncTo, length, func() {
272 0 : fullSync, err = d.file.SyncTo(length)
273 0 : }, time.Now().UnixNano())
274 0 : return fullSync, err
275 : }
276 :
277 : // timeDiskOp runs the specified closure and makes its timing visible to the
278 : // monitoring goroutine, in case it exceeds one of the slow disk durations.
279 : // opType should always be set. writeSizeInBytes should be set if the write
280 : // operation is sized. If not, it should be set to zero.
281 : //
282 : // The start time is taken as a parameter in the form of nanoseconds since the
283 : // unix epoch so that it appears in stack traces during crashes (if GOTRACEBACK
284 : // is set appropriately), aiding postmortem debugging.
285 : func (d *diskHealthCheckingFile) timeDiskOp(
286 : opType OpType, writeSizeInBytes int64, op func(), startNanos int64,
287 1 : ) {
288 1 : if d == nil {
289 0 : op()
290 0 : return
291 0 : }
292 :
293 1 : delta := time.Duration(startNanos - d.createTimeNanos)
294 1 : packed := pack(delta, writeSizeInBytes, opType)
295 1 : if d.lastWritePacked.Swap(packed) != 0 {
296 0 : panic("concurrent write operations detected on file")
297 : }
298 1 : defer func() {
299 1 : if d.lastWritePacked.Swap(0) != packed {
300 0 : panic("concurrent write operations detected on file")
301 : }
302 : }()
303 1 : op()
304 : }
305 :
306 : // Note the slight lack of symmetry between pack & unpack. pack takes an int64 for writeSizeInBytes, since
307 : // callers of pack use an int64. This is dictated by the vfs interface. unpack OTOH returns an int. This is
308 : // safe because the packing scheme implies we only actually need 32 bits.
309 1 : func pack(delta time.Duration, writeSizeInBytes int64, opType OpType) uint64 {
310 1 : // We have no guarantee of clock monotonicity. If we have a small regression
311 1 : // in the clock, we set deltaMillis to zero, so we can still catch the operation
312 1 : // if happens to be slow.
313 1 : deltaMillis := delta.Milliseconds()
314 1 : if deltaMillis < 0 {
315 0 : deltaMillis = 0
316 0 : }
317 : // As of 3/7/2023, the use of 40 bits for an delta provides ~34 years
318 : // of effective monitoring time before the uint wraps around, at millisecond
319 : // precision.
320 1 : if deltaMillis > 1<<deltaBits-1 {
321 0 : panic("vfs: last write delta would result in integer wraparound")
322 : }
323 :
324 : // See writeSizePrecision to get the unit of writeSize. As of 1/26/2023, the unit is KBs.
325 1 : writeSize := writeSizeInBytes / writeSizePrecision
326 1 : // If the size of the write is larger than we can store in the packed int, store the max
327 1 : // value we can store in the packed int.
328 1 : const writeSizeCeiling = 1<<writeSizeBits - 1
329 1 : if writeSize > writeSizeCeiling {
330 0 : writeSize = writeSizeCeiling
331 0 : }
332 :
333 1 : return uint64(deltaMillis)<<(64-deltaBits) | uint64(writeSize)<<(64-deltaBits-writeSizeBits) | uint64(opType)
334 : }
335 :
336 0 : func unpack(packed uint64) (delta time.Duration, writeSizeInBytes int, opType OpType) {
337 0 : delta = time.Duration(packed>>(64-deltaBits)) * time.Millisecond
338 0 : wz := int64(packed>>(64-deltaBits-writeSizeBits)) & ((1 << writeSizeBits) - 1) * writeSizePrecision
339 0 : // Given the packing scheme, converting wz to an int will not truncate anything.
340 0 : writeSizeInBytes = int(wz)
341 0 : opType = OpType(packed & 0xf)
342 0 : return delta, writeSizeInBytes, opType
343 0 : }
344 :
345 : // diskHealthCheckingDir implements disk-health checking for directories. Unlike
346 : // other files, we allow directories to receive concurrent write operations
347 : // (Syncs are the only write operations supported by a directory.) Since the
348 : // diskHealthCheckingFile's timeDiskOp can only track a single in-flight
349 : // operation at a time, we time the operation using the filesystem-level
350 : // timeFilesystemOp function instead.
351 : type diskHealthCheckingDir struct {
352 : File
353 : name string
354 : fs *diskHealthCheckingFS
355 : }
356 :
357 : // Sync implements the io.Syncer interface.
358 1 : func (d *diskHealthCheckingDir) Sync() (err error) {
359 1 : d.fs.timeFilesystemOp(d.name, OpTypeSync, func() {
360 1 : err = d.File.Sync()
361 1 : }, time.Now().UnixNano())
362 1 : return err
363 : }
364 :
365 : // DiskSlowInfo captures info about detected slow operations on the vfs.
366 : type DiskSlowInfo struct {
367 : // Path of file being written to.
368 : Path string
369 : // Operation being performed on the file.
370 : OpType OpType
371 : // Size of write in bytes, if the write is sized.
372 : WriteSize int
373 : // Duration that has elapsed since this disk operation started.
374 : Duration time.Duration
375 : }
376 :
377 0 : func (i DiskSlowInfo) String() string {
378 0 : return redact.StringWithoutMarkers(i)
379 0 : }
380 :
381 : // SafeFormat implements redact.SafeFormatter.
382 0 : func (i DiskSlowInfo) SafeFormat(w redact.SafePrinter, _ rune) {
383 0 : switch i.OpType {
384 : // Operations for which i.WriteSize is meaningful.
385 0 : case OpTypeWrite, OpTypeSyncTo, OpTypePreallocate:
386 0 : w.Printf("disk slowness detected: %s on file %s (%d bytes) has been ongoing for %0.1fs",
387 0 : redact.Safe(i.OpType.String()), redact.Safe(filepath.Base(i.Path)),
388 0 : redact.Safe(i.WriteSize), redact.Safe(i.Duration.Seconds()))
389 0 : default:
390 0 : w.Printf("disk slowness detected: %s on file %s has been ongoing for %0.1fs",
391 0 : redact.Safe(i.OpType.String()), redact.Safe(filepath.Base(i.Path)),
392 0 : redact.Safe(i.Duration.Seconds()))
393 : }
394 : }
395 :
396 : // diskHealthCheckingFS adds disk-health checking facilities to a VFS.
397 : // It times disk write operations in two ways:
398 : //
399 : // 1. Wrapping vfs.Files.
400 : //
401 : // The bulk of write I/O activity is file writing and syncing, invoked through
402 : // the `vfs.File` interface. This VFS wraps all files open for writing with a
403 : // special diskHealthCheckingFile implementation of the vfs.File interface. See
404 : // above for the implementation.
405 : //
406 : // 2. Monitoring filesystem metadata operations.
407 : //
408 : // Filesystem metadata operations (create, link, remove, rename, etc) are also
409 : // sources of disk writes. Unlike a vfs.File which requires Write and Sync calls
410 : // to be sequential, a vfs.FS may receive these filesystem metadata operations
411 : // in parallel. To accommodate this parallelism, the diskHealthCheckingFS's
412 : // write-oriented filesystem operations record their start times into a 'slot'
413 : // on the filesystem. A single long-running goroutine periodically scans the
414 : // slots looking for slow operations.
415 : //
416 : // The number of slots on a diskHealthCheckingFS grows to a working set of the
417 : // maximum concurrent filesystem operations. This is expected to be very few
418 : // for these reasons:
419 : // 1. Pebble has limited write concurrency. Flushes, compactions and WAL
420 : // rotations are the primary sources of filesystem metadata operations. With
421 : // the default max-compaction concurrency, these operations require at most 5
422 : // concurrent slots if all 5 perform a filesystem metadata operation
423 : // simultaneously.
424 : // 2. Pebble's limited concurrent I/O writers spend most of their time
425 : // performing file I/O, not performing the filesystem metadata operations that
426 : // require recording a slot on the diskHealthCheckingFS.
427 : // 3. In CockroachDB, each additional store/Pebble instance has its own vfs.FS
428 : // which provides a separate goroutine and set of slots.
429 : // 4. In CockroachDB, many of the additional sources of filesystem metadata
430 : // operations (like encryption-at-rest) are sequential with respect to Pebble's
431 : // threads.
432 : type diskHealthCheckingFS struct {
433 : tickInterval time.Duration
434 : diskSlowThreshold time.Duration
435 : onSlowDisk func(DiskSlowInfo)
436 : fs FS
437 : mu struct {
438 : sync.Mutex
439 : tickerRunning bool
440 : stopper chan struct{}
441 : inflight []*slot
442 : }
443 : // prealloc preallocates the memory for mu.inflight slots and the slice
444 : // itself. The contained fields are not accessed directly except by
445 : // WithDiskHealthChecks when initializing mu.inflight. The number of slots
446 : // in d.mu.inflight will grow to the maximum number of concurrent file
447 : // metadata operations (create, remove, link, etc). If the number of
448 : // concurrent operations never exceeds preallocatedSlotCount, we'll never
449 : // incur an additional allocation.
450 : prealloc struct {
451 : slots [preallocatedSlotCount]slot
452 : slotPtrSlice [preallocatedSlotCount]*slot
453 : }
454 : }
455 :
456 : type slot struct {
457 : name string
458 : opType OpType
459 : startNanos atomic.Int64
460 : }
461 :
462 : // diskHealthCheckingFS implements FS.
463 : var _ FS = (*diskHealthCheckingFS)(nil)
464 :
465 : // WithDiskHealthChecks wraps an FS and ensures that all write-oriented
466 : // operations on the FS are wrapped with disk health detection checks. Disk
467 : // operations that are observed to take longer than diskSlowThreshold trigger an
468 : // onSlowDisk call.
469 : //
470 : // A threshold of zero disables disk-health checking.
471 : func WithDiskHealthChecks(
472 : innerFS FS, diskSlowThreshold time.Duration, onSlowDisk func(info DiskSlowInfo),
473 1 : ) (FS, io.Closer) {
474 1 : if diskSlowThreshold == 0 {
475 0 : return innerFS, noopCloser{}
476 0 : }
477 :
478 1 : fs := &diskHealthCheckingFS{
479 1 : fs: innerFS,
480 1 : tickInterval: defaultTickInterval,
481 1 : diskSlowThreshold: diskSlowThreshold,
482 1 : onSlowDisk: onSlowDisk,
483 1 : }
484 1 : fs.mu.stopper = make(chan struct{})
485 1 : // The fs holds preallocated slots and a preallocated array of slot pointers
486 1 : // with equal length. Initialize the inflight slice to use a slice backed by
487 1 : // the preallocated array with each slot initialized to a preallocated slot.
488 1 : fs.mu.inflight = fs.prealloc.slotPtrSlice[:]
489 1 : for i := range fs.mu.inflight {
490 1 : fs.mu.inflight[i] = &fs.prealloc.slots[i]
491 1 : }
492 1 : return fs, fs
493 : }
494 :
495 : // timeFilesystemOp executes the provided closure, which should perform a
496 : // singular filesystem operation of a type matching opType on the named file. It
497 : // records the provided start time such that the long-lived disk-health checking
498 : // goroutine can observe if the operation is blocked for an inordinate time.
499 : //
500 : // The start time is taken as a parameter in the form of nanoseconds since the
501 : // unix epoch so that it appears in stack traces during crashes (if GOTRACEBACK
502 : // is set appropriately), aiding postmortem debugging.
503 : func (d *diskHealthCheckingFS) timeFilesystemOp(
504 : name string, opType OpType, op func(), startNanos int64,
505 1 : ) {
506 1 : if d == nil {
507 0 : op()
508 0 : return
509 0 : }
510 :
511 : // Record this operation's start time on the FS, so that the long-running
512 : // goroutine can monitor the filesystem operation.
513 : //
514 : // The diskHealthCheckingFile implementation uses a single field that is
515 : // atomically updated, taking advantage of the fact that writes to a single
516 : // vfs.File handle are not performed in parallel. The vfs.FS however may
517 : // receive write filesystem operations in parallel. To accommodate this
518 : // parallelism, writing goroutines append their start time to a
519 : // mutex-protected vector. On ticks, the long-running goroutine scans the
520 : // vector searching for start times older than the slow-disk threshold. When
521 : // a writing goroutine completes its operation, it atomically overwrites its
522 : // slot to signal completion.
523 1 : var s *slot
524 1 : func() {
525 1 : d.mu.Lock()
526 1 : defer d.mu.Unlock()
527 1 :
528 1 : // If there's no long-running goroutine to monitor this filesystem
529 1 : // operation, start one.
530 1 : if !d.mu.tickerRunning {
531 1 : d.startTickerLocked()
532 1 : }
533 :
534 1 : for i := 0; i < len(d.mu.inflight); i++ {
535 1 : if d.mu.inflight[i].startNanos.Load() == 0 {
536 1 : // This slot is not in use. Claim it.
537 1 : s = d.mu.inflight[i]
538 1 : s.name = name
539 1 : s.opType = opType
540 1 : s.startNanos.Store(startNanos)
541 1 : break
542 : }
543 : }
544 : // If we didn't find any unused slots, create a new slot and append it.
545 : // This slot will exist forever. The number of slots will grow to the
546 : // maximum number of concurrent filesystem operations over the lifetime
547 : // of the process. Only operations that grow the number of slots must
548 : // incur an allocation.
549 1 : if s == nil {
550 0 : s = &slot{
551 0 : name: name,
552 0 : opType: opType,
553 0 : }
554 0 : s.startNanos.Store(startNanos)
555 0 : d.mu.inflight = append(d.mu.inflight, s)
556 0 : }
557 : }()
558 :
559 1 : op()
560 1 :
561 1 : // Signal completion by zeroing the start time.
562 1 : s.startNanos.Store(0)
563 : }
564 :
565 : // startTickerLocked starts a new goroutine with a ticker to monitor disk
566 : // filesystem operations. Requires d.mu and !d.mu.tickerRunning.
567 1 : func (d *diskHealthCheckingFS) startTickerLocked() {
568 1 : d.mu.tickerRunning = true
569 1 : stopper := d.mu.stopper
570 1 : go func() {
571 1 : ticker := time.NewTicker(d.tickInterval)
572 1 : defer ticker.Stop()
573 1 : type exceededSlot struct {
574 1 : name string
575 1 : opType OpType
576 1 : startNanos int64
577 1 : }
578 1 : var exceededSlots []exceededSlot
579 1 :
580 1 : for {
581 1 : select {
582 1 : case <-ticker.C:
583 1 : // Scan the inflight slots for any slots recording a start
584 1 : // time older than the diskSlowThreshold.
585 1 : exceededSlots = exceededSlots[:0]
586 1 : d.mu.Lock()
587 1 : now := time.Now()
588 1 : for i := range d.mu.inflight {
589 1 : nanos := d.mu.inflight[i].startNanos.Load()
590 1 : if nanos != 0 && time.Unix(0, nanos).Add(d.diskSlowThreshold).Before(now) {
591 0 : // diskSlowThreshold was exceeded. Copy this inflightOp into
592 0 : // exceededSlots and call d.onSlowDisk after dropping the mutex.
593 0 : inflightOp := exceededSlot{
594 0 : name: d.mu.inflight[i].name,
595 0 : opType: d.mu.inflight[i].opType,
596 0 : startNanos: nanos,
597 0 : }
598 0 : exceededSlots = append(exceededSlots, inflightOp)
599 0 : }
600 : }
601 1 : d.mu.Unlock()
602 1 : for i := range exceededSlots {
603 0 : d.onSlowDisk(
604 0 : DiskSlowInfo{
605 0 : Path: exceededSlots[i].name,
606 0 : OpType: exceededSlots[i].opType,
607 0 : WriteSize: 0, // writes at the fs level are not sized
608 0 : Duration: now.Sub(time.Unix(0, exceededSlots[i].startNanos)),
609 0 : })
610 0 : }
611 1 : case <-stopper:
612 1 : return
613 : }
614 : }
615 : }()
616 : }
617 :
618 : // Close implements io.Closer. Close stops the long-running goroutine that
619 : // monitors for slow filesystem metadata operations. Close may be called
620 : // multiple times. If the filesystem is used after Close has been called, a new
621 : // long-running goroutine will be created.
622 1 : func (d *diskHealthCheckingFS) Close() error {
623 1 : d.mu.Lock()
624 1 : if !d.mu.tickerRunning {
625 1 : // Nothing to stop.
626 1 : d.mu.Unlock()
627 1 : return nil
628 1 : }
629 :
630 : // Grab the stopper so we can request the long-running goroutine to stop.
631 : // Replace the stopper in case this FS is reused. It's possible to Close and
632 : // reuse a disk-health checking FS. This is to accommodate the on-by-default
633 : // behavior in Pebble, and the possibility that users may continue to use
634 : // the Pebble default FS beyond the lifetime of a single DB.
635 1 : stopper := d.mu.stopper
636 1 : d.mu.stopper = make(chan struct{})
637 1 : d.mu.tickerRunning = false
638 1 : d.mu.Unlock()
639 1 :
640 1 : // Ask the long-running goroutine to stop. This is a synchronous channel
641 1 : // send.
642 1 : stopper <- struct{}{}
643 1 : close(stopper)
644 1 : return nil
645 : }
646 :
647 : // Create implements the FS interface.
648 1 : func (d *diskHealthCheckingFS) Create(name string) (File, error) {
649 1 : var f File
650 1 : var err error
651 1 : d.timeFilesystemOp(name, OpTypeCreate, func() {
652 1 : f, err = d.fs.Create(name)
653 1 : }, time.Now().UnixNano())
654 1 : if err != nil {
655 0 : return f, err
656 0 : }
657 1 : if d.diskSlowThreshold == 0 {
658 0 : return f, nil
659 0 : }
660 1 : checkingFile := newDiskHealthCheckingFile(f, d.diskSlowThreshold, func(opType OpType, writeSizeInBytes int, duration time.Duration) {
661 0 : d.onSlowDisk(
662 0 : DiskSlowInfo{
663 0 : Path: name,
664 0 : OpType: opType,
665 0 : WriteSize: writeSizeInBytes,
666 0 : Duration: duration,
667 0 : })
668 0 : })
669 1 : checkingFile.startTicker()
670 1 : return checkingFile, nil
671 : }
672 :
673 : // GetDiskUsage implements the FS interface.
674 1 : func (d *diskHealthCheckingFS) GetDiskUsage(path string) (DiskUsage, error) {
675 1 : return d.fs.GetDiskUsage(path)
676 1 : }
677 :
678 : // Link implements the FS interface.
679 1 : func (d *diskHealthCheckingFS) Link(oldname, newname string) error {
680 1 : var err error
681 1 : d.timeFilesystemOp(newname, OpTypeLink, func() {
682 1 : err = d.fs.Link(oldname, newname)
683 1 : }, time.Now().UnixNano())
684 1 : return err
685 : }
686 :
687 : // List implements the FS interface.
688 1 : func (d *diskHealthCheckingFS) List(dir string) ([]string, error) {
689 1 : return d.fs.List(dir)
690 1 : }
691 :
692 : // Lock implements the FS interface.
693 1 : func (d *diskHealthCheckingFS) Lock(name string) (io.Closer, error) {
694 1 : return d.fs.Lock(name)
695 1 : }
696 :
697 : // MkdirAll implements the FS interface.
698 1 : func (d *diskHealthCheckingFS) MkdirAll(dir string, perm os.FileMode) error {
699 1 : var err error
700 1 : d.timeFilesystemOp(dir, OpTypeMkdirAll, func() {
701 1 : err = d.fs.MkdirAll(dir, perm)
702 1 : }, time.Now().UnixNano())
703 1 : return err
704 : }
705 :
706 : // Open implements the FS interface.
707 1 : func (d *diskHealthCheckingFS) Open(name string, opts ...OpenOption) (File, error) {
708 1 : return d.fs.Open(name, opts...)
709 1 : }
710 :
711 : // OpenReadWrite implements the FS interface.
712 1 : func (d *diskHealthCheckingFS) OpenReadWrite(name string, opts ...OpenOption) (File, error) {
713 1 : return d.fs.OpenReadWrite(name, opts...)
714 1 : }
715 :
716 : // OpenDir implements the FS interface.
717 1 : func (d *diskHealthCheckingFS) OpenDir(name string) (File, error) {
718 1 : f, err := d.fs.OpenDir(name)
719 1 : if err != nil {
720 0 : return f, err
721 0 : }
722 : // Directories opened with OpenDir must be opened with health checking,
723 : // because they may be explicitly synced.
724 1 : return &diskHealthCheckingDir{
725 1 : File: f,
726 1 : name: name,
727 1 : fs: d,
728 1 : }, nil
729 : }
730 :
731 : // PathBase implements the FS interface.
732 1 : func (d *diskHealthCheckingFS) PathBase(path string) string {
733 1 : return d.fs.PathBase(path)
734 1 : }
735 :
736 : // PathJoin implements the FS interface.
737 1 : func (d *diskHealthCheckingFS) PathJoin(elem ...string) string {
738 1 : return d.fs.PathJoin(elem...)
739 1 : }
740 :
741 : // PathDir implements the FS interface.
742 1 : func (d *diskHealthCheckingFS) PathDir(path string) string {
743 1 : return d.fs.PathDir(path)
744 1 : }
745 :
746 : // Remove implements the FS interface.
747 1 : func (d *diskHealthCheckingFS) Remove(name string) error {
748 1 : var err error
749 1 : d.timeFilesystemOp(name, OpTypeRemove, func() {
750 1 : err = d.fs.Remove(name)
751 1 : }, time.Now().UnixNano())
752 1 : return err
753 : }
754 :
755 : // RemoveAll implements the FS interface.
756 0 : func (d *diskHealthCheckingFS) RemoveAll(name string) error {
757 0 : var err error
758 0 : d.timeFilesystemOp(name, OpTypeRemoveAll, func() {
759 0 : err = d.fs.RemoveAll(name)
760 0 : }, time.Now().UnixNano())
761 0 : return err
762 : }
763 :
764 : // Rename implements the FS interface.
765 1 : func (d *diskHealthCheckingFS) Rename(oldname, newname string) error {
766 1 : var err error
767 1 : d.timeFilesystemOp(newname, OpTypeRename, func() {
768 1 : err = d.fs.Rename(oldname, newname)
769 1 : }, time.Now().UnixNano())
770 1 : return err
771 : }
772 :
773 : // ReuseForWrite implements the FS interface.
774 0 : func (d *diskHealthCheckingFS) ReuseForWrite(oldname, newname string) (File, error) {
775 0 : var f File
776 0 : var err error
777 0 : d.timeFilesystemOp(newname, OpTypeReuseForWrite, func() {
778 0 : f, err = d.fs.ReuseForWrite(oldname, newname)
779 0 : }, time.Now().UnixNano())
780 0 : if err != nil {
781 0 : return f, err
782 0 : }
783 0 : if d.diskSlowThreshold == 0 {
784 0 : return f, nil
785 0 : }
786 0 : checkingFile := newDiskHealthCheckingFile(f, d.diskSlowThreshold, func(opType OpType, writeSizeInBytes int, duration time.Duration) {
787 0 : d.onSlowDisk(
788 0 : DiskSlowInfo{
789 0 : Path: newname,
790 0 : OpType: opType,
791 0 : WriteSize: writeSizeInBytes,
792 0 : Duration: duration,
793 0 : })
794 0 : })
795 0 : checkingFile.startTicker()
796 0 : return checkingFile, nil
797 : }
798 :
799 : // Stat implements the FS interface.
800 1 : func (d *diskHealthCheckingFS) Stat(name string) (os.FileInfo, error) {
801 1 : return d.fs.Stat(name)
802 1 : }
803 :
804 : type noopCloser struct{}
805 :
806 0 : func (noopCloser) Close() error { return nil }
|
