Line data Source code
1 : // Copyright 2024 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 wal
6 :
7 : import (
8 : "os"
9 : "sync"
10 : "time"
11 :
12 : "github.com/cockroachdb/pebble/internal/base"
13 : "github.com/cockroachdb/pebble/vfs"
14 : "golang.org/x/exp/rand"
15 : )
16 :
17 : // dirProber probes the primary dir, until it is confirmed to be healthy. If
18 : // it doesn't have enough samples, it is deemed to be unhealthy. It is only
19 : // used for failback to the primary.
20 : type dirProber struct {
21 : fs vfs.FS
22 : // The full path of the file to use for the probe. The probe is destructive
23 : // in that it deletes and (re)creates the file.
24 : filename string
25 : // The probing interval, when enabled.
26 : interval time.Duration
27 : timeSource
28 : // buf holds the random bytes that are written during the probe.
29 : buf [100 << 10]byte
30 : // enabled is signaled to enable and disable probing. The initial state of
31 : // the prober is disabled.
32 : enabled chan bool
33 : mu struct {
34 : sync.Mutex
35 : // Circular buffer of history samples.
36 : history [probeHistoryLength]time.Duration
37 : // The history is in [firstProbeIndex, nextProbeIndex).
38 : firstProbeIndex int
39 : nextProbeIndex int
40 : }
41 : iterationForTesting chan<- struct{}
42 : }
43 :
44 : const probeHistoryLength = 128
45 :
46 : // Large value.
47 : const failedProbeDuration = 24 * 60 * 60 * time.Second
48 :
49 : // There is no close/stop method on the dirProber -- use stopper.
50 : func (p *dirProber) init(
51 : fs vfs.FS,
52 : filename string,
53 : interval time.Duration,
54 : stopper *stopper,
55 : ts timeSource,
56 : notifyIterationForTesting chan<- struct{},
57 1 : ) {
58 1 : *p = dirProber{
59 1 : fs: fs,
60 1 : filename: filename,
61 1 : interval: interval,
62 1 : timeSource: ts,
63 1 : enabled: make(chan bool),
64 1 : iterationForTesting: notifyIterationForTesting,
65 1 : }
66 1 : // Random bytes for writing, to defeat any FS compression optimization.
67 1 : _, err := rand.Read(p.buf[:])
68 1 : if err != nil {
69 0 : panic(err)
70 : }
71 1 : stopper.runAsync(func() {
72 1 : p.probeLoop(stopper.shouldQuiesce())
73 1 : })
74 : }
75 :
76 1 : func (p *dirProber) probeLoop(shouldQuiesce <-chan struct{}) {
77 1 : ticker := p.timeSource.newTicker(p.interval)
78 1 : ticker.stop()
79 1 : tickerCh := ticker.ch()
80 1 : done := false
81 1 : var enabled bool
82 1 : for !done {
83 1 : select {
84 1 : case <-tickerCh:
85 1 : if !enabled {
86 0 : // Could have a tick waiting before we disabled it. Ignore.
87 0 : continue
88 : }
89 1 : probeDur := func() time.Duration {
90 1 : // Delete, create, write, sync.
91 1 : start := p.timeSource.now()
92 1 : _ = p.fs.Remove(p.filename)
93 1 : f, err := p.fs.Create(p.filename)
94 1 : if err != nil {
95 1 : return failedProbeDuration
96 1 : }
97 1 : defer f.Close()
98 1 : n, err := f.Write(p.buf[:])
99 1 : if err != nil {
100 0 : return failedProbeDuration
101 0 : }
102 1 : if n != len(p.buf) {
103 0 : panic("invariant violation")
104 : }
105 1 : err = f.Sync()
106 1 : if err != nil {
107 0 : return failedProbeDuration
108 0 : }
109 1 : return p.timeSource.now().Sub(start)
110 : }()
111 1 : p.mu.Lock()
112 1 : nextIndex := p.mu.nextProbeIndex % probeHistoryLength
113 1 : p.mu.history[nextIndex] = probeDur
114 1 : p.mu.nextProbeIndex++
115 1 : numSamples := p.mu.nextProbeIndex - p.mu.firstProbeIndex
116 1 : if numSamples > probeHistoryLength {
117 0 : // Length has exceeded capacity, i.e., overwritten the first probe.
118 0 : p.mu.firstProbeIndex++
119 0 : }
120 1 : p.mu.Unlock()
121 :
122 1 : case <-shouldQuiesce:
123 1 : done = true
124 :
125 1 : case enabled = <-p.enabled:
126 1 : if enabled {
127 1 : ticker.reset(p.interval)
128 1 : } else {
129 1 : ticker.stop()
130 1 : p.mu.Lock()
131 1 : p.mu.firstProbeIndex = 0
132 1 : p.mu.nextProbeIndex = 0
133 1 : p.mu.Unlock()
134 1 : }
135 : }
136 1 : if p.iterationForTesting != nil {
137 1 : p.iterationForTesting <- struct{}{}
138 1 : }
139 : }
140 : }
141 :
142 1 : func (p *dirProber) enableProbing() {
143 1 : p.enabled <- true
144 1 : }
145 :
146 1 : func (p *dirProber) disableProbing() {
147 1 : p.enabled <- false
148 1 : }
149 :
150 1 : func (p *dirProber) getMeanMax(interval time.Duration) (time.Duration, time.Duration) {
151 1 : p.mu.Lock()
152 1 : defer p.mu.Unlock()
153 1 : numSamples := p.mu.nextProbeIndex - p.mu.firstProbeIndex
154 1 : samplesNeeded := int((interval + p.interval - 1) / p.interval)
155 1 : if samplesNeeded == 0 {
156 0 : panic("interval is too short")
157 1 : } else if samplesNeeded > probeHistoryLength {
158 0 : panic("interval is too long")
159 : }
160 1 : if samplesNeeded > numSamples {
161 1 : // Not enough samples, so assume not yet healthy.
162 1 : return failedProbeDuration, failedProbeDuration
163 1 : }
164 1 : offset := numSamples - samplesNeeded
165 1 : var sum, max time.Duration
166 1 : for i := p.mu.firstProbeIndex + offset; i < p.mu.nextProbeIndex; i++ {
167 1 : sampleDur := p.mu.history[i%probeHistoryLength]
168 1 : sum += sampleDur
169 1 : if max < sampleDur {
170 1 : max = sampleDur
171 1 : }
172 : }
173 1 : mean := sum / time.Duration(samplesNeeded)
174 1 : return mean, max
175 : }
176 :
177 : type dirIndex int
178 :
179 : const (
180 : primaryDirIndex dirIndex = iota
181 : secondaryDirIndex
182 : numDirIndices
183 : )
184 :
185 : type dirAndFileHandle struct {
186 : Dir
187 : vfs.File
188 : }
189 :
190 : // switchableWriter is a subset of failoverWriter needed by failoverMonitor.
191 : type switchableWriter interface {
192 : switchToNewDir(dirAndFileHandle) error
193 : ongoingLatencyOrErrorForCurDir() (time.Duration, error)
194 : }
195 :
196 : type failoverMonitorOptions struct {
197 : // The primary and secondary dir.
198 : dirs [numDirIndices]dirAndFileHandle
199 :
200 : FailoverOptions
201 : stopper *stopper
202 : }
203 :
204 : // failoverMonitor monitors the latency and error observed by the
205 : // switchableWriter, and does failover by switching the dir. It also monitors
206 : // the primary dir for failback.
207 : type failoverMonitor struct {
208 : opts failoverMonitorOptions
209 : prober dirProber
210 : mu struct {
211 : sync.Mutex
212 : // dirIndex and lastFailbackTime are only modified by monitorLoop. They
213 : // are protected by the mutex for concurrent reads.
214 :
215 : // dirIndex is the directory to use by writer (if non-nil), or when the
216 : // writer becomes non-nil.
217 : dirIndex
218 : // The time at which the monitor last failed back to the primary. This is
219 : // only relevant when dirIndex == primaryDirIndex.
220 : lastFailBackTime time.Time
221 : // The current failoverWriter, exposed via a narrower interface.
222 : writer switchableWriter
223 :
224 : // Stats.
225 : dirSwitchCount int64
226 : lastAccumulateIntoDurations time.Time
227 : primaryWriteDuration time.Duration
228 : secondaryWriteDuration time.Duration
229 : }
230 : }
231 :
232 1 : func newFailoverMonitor(opts failoverMonitorOptions) *failoverMonitor {
233 1 : m := &failoverMonitor{
234 1 : opts: opts,
235 1 : }
236 1 : m.mu.lastAccumulateIntoDurations = opts.timeSource.now()
237 1 : m.prober.init(opts.dirs[primaryDirIndex].FS,
238 1 : opts.dirs[primaryDirIndex].FS.PathJoin(opts.dirs[primaryDirIndex].Dirname, "probe-file"),
239 1 : opts.PrimaryDirProbeInterval, opts.stopper, opts.timeSource, opts.proberIterationForTesting)
240 1 : opts.stopper.runAsync(func() {
241 1 : m.monitorLoop(opts.stopper.shouldQuiesce())
242 1 : })
243 1 : return m
244 : }
245 :
246 : // Called when previous writer is closed
247 1 : func (m *failoverMonitor) noWriter() {
248 1 : now := m.opts.timeSource.now()
249 1 : m.mu.Lock()
250 1 : defer m.mu.Unlock()
251 1 : m.mu.writer = nil
252 1 : m.accumulateDurationLocked(now)
253 1 : }
254 :
255 : // writerCreateFunc is allowed to return nil, if there is an error. It is not
256 : // the responsibility of failoverMonitor to handle that error. So this should
257 : // not be due to an IO error (which failoverMonitor is interested in).
258 1 : func (m *failoverMonitor) newWriter(writerCreateFunc func(dir dirAndFileHandle) switchableWriter) {
259 1 : m.mu.Lock()
260 1 : defer m.mu.Unlock()
261 1 : if m.mu.writer != nil {
262 0 : panic("previous writer not closed")
263 : }
264 1 : m.mu.writer = writerCreateFunc(m.opts.dirs[m.mu.dirIndex])
265 : }
266 :
267 1 : func (m *failoverMonitor) elevateWriteStallThresholdForFailover() bool {
268 1 : m.mu.Lock()
269 1 : defer m.mu.Unlock()
270 1 : if m.mu.dirIndex == secondaryDirIndex {
271 1 : return true
272 1 : }
273 1 : intervalSinceFailedback := m.opts.timeSource.now().Sub(m.mu.lastFailBackTime)
274 1 : return intervalSinceFailedback < m.opts.ElevatedWriteStallThresholdLag
275 : }
276 :
277 1 : func (m *failoverMonitor) accumulateDurationLocked(now time.Time) {
278 1 : dur := now.Sub(m.mu.lastAccumulateIntoDurations)
279 1 : m.mu.lastAccumulateIntoDurations = now
280 1 : if m.mu.dirIndex == primaryDirIndex {
281 1 : m.mu.primaryWriteDuration += dur
282 1 : return
283 1 : }
284 1 : m.mu.secondaryWriteDuration += dur
285 : }
286 :
287 1 : func (m *failoverMonitor) stats() FailoverStats {
288 1 : now := m.opts.timeSource.now()
289 1 : m.mu.Lock()
290 1 : defer m.mu.Unlock()
291 1 : m.accumulateDurationLocked(now)
292 1 : return FailoverStats{
293 1 : DirSwitchCount: m.mu.dirSwitchCount,
294 1 : PrimaryWriteDuration: m.mu.primaryWriteDuration,
295 1 : SecondaryWriteDuration: m.mu.secondaryWriteDuration,
296 1 : }
297 1 : }
298 :
299 : // lastWriterInfo is state maintained in the monitorLoop for the latest
300 : // switchable writer. It is mainly used to dampen the switching.
301 : type lastWriterInfo struct {
302 : writer switchableWriter
303 : numSwitches int
304 : ongoingLatencyAtSwitch time.Duration
305 : errorCounts [numDirIndices]int
306 : }
307 :
308 1 : func (m *failoverMonitor) monitorLoop(shouldQuiesce <-chan struct{}) {
309 1 : ticker := m.opts.timeSource.newTicker(m.opts.UnhealthySamplingInterval)
310 1 : if m.opts.monitorIterationForTesting != nil {
311 1 : m.opts.monitorIterationForTesting <- struct{}{}
312 1 : }
313 1 : tickerCh := ticker.ch()
314 1 : dirIndex := primaryDirIndex
315 1 : var lastWriter lastWriterInfo
316 1 : for {
317 1 : select {
318 1 : case <-shouldQuiesce:
319 1 : ticker.stop()
320 1 : return
321 1 : case <-tickerCh:
322 1 : writerOngoingLatency, writerErr := func() (time.Duration, error) {
323 1 : m.mu.Lock()
324 1 : defer m.mu.Unlock()
325 1 : if m.mu.writer != lastWriter.writer {
326 1 : lastWriter = lastWriterInfo{writer: m.mu.writer}
327 1 : }
328 1 : if lastWriter.writer == nil {
329 1 : return 0, nil
330 1 : }
331 1 : return lastWriter.writer.ongoingLatencyOrErrorForCurDir()
332 : }()
333 1 : switchDir := false
334 1 : // Arbitrary value.
335 1 : const highSecondaryErrorCountThreshold = 2
336 1 : // We don't consider a switch if currently using the primary dir and the
337 1 : // secondary dir has high enough errors. It is more likely that someone
338 1 : // has misconfigured a secondary e.g. wrong permissions or not enough
339 1 : // disk space. We only remember the error history in the context of the
340 1 : // lastWriter since an operator can fix the underlying misconfiguration.
341 1 : if !(lastWriter.errorCounts[secondaryDirIndex] >= highSecondaryErrorCountThreshold &&
342 1 : dirIndex == primaryDirIndex) {
343 1 : // Switching heuristics. Subject to change based on real world experience.
344 1 : if writerErr != nil {
345 1 : // An error causes an immediate switch, since a LogWriter with an
346 1 : // error is useless.
347 1 : lastWriter.errorCounts[dirIndex]++
348 1 : switchDir = true
349 1 : } else if writerOngoingLatency > m.opts.UnhealthyOperationLatencyThreshold() {
350 1 : // Arbitrary value.
351 1 : const switchImmediatelyCountThreshold = 2
352 1 : // High latency. Switch immediately if the number of switches that
353 1 : // have been done is below the threshold, since that gives us an
354 1 : // observation of both dirs. Once above that threshold, decay the
355 1 : // switch rate by increasing the observed latency needed for a
356 1 : // switch.
357 1 : if lastWriter.numSwitches < switchImmediatelyCountThreshold ||
358 1 : writerOngoingLatency > 2*lastWriter.ongoingLatencyAtSwitch {
359 1 : switchDir = true
360 1 : lastWriter.ongoingLatencyAtSwitch = writerOngoingLatency
361 1 : }
362 : // Else high latency, but not high enough yet to motivate switch.
363 1 : } else if dirIndex == secondaryDirIndex {
364 1 : // The writer looks healthy. We can still switch if the writer is using the
365 1 : // secondary dir and the primary is healthy again.
366 1 : primaryMean, primaryMax := m.prober.getMeanMax(m.opts.HealthyInterval)
367 1 : if primaryMean < m.opts.HealthyProbeLatencyThreshold &&
368 1 : primaryMax < m.opts.HealthyProbeLatencyThreshold {
369 1 : switchDir = true
370 1 : }
371 : }
372 : }
373 1 : if switchDir {
374 1 : lastWriter.numSwitches++
375 1 : if dirIndex == secondaryDirIndex {
376 1 : // Switching back to primary, so don't need to probe to see if
377 1 : // primary is healthy.
378 1 : m.prober.disableProbing()
379 1 : dirIndex = primaryDirIndex
380 1 : } else {
381 1 : m.prober.enableProbing()
382 1 : dirIndex = secondaryDirIndex
383 1 : }
384 1 : dir := m.opts.dirs[dirIndex]
385 1 : now := m.opts.timeSource.now()
386 1 : m.mu.Lock()
387 1 : m.accumulateDurationLocked(now)
388 1 : m.mu.dirIndex = dirIndex
389 1 : m.mu.dirSwitchCount++
390 1 : if dirIndex == primaryDirIndex {
391 1 : m.mu.lastFailBackTime = now
392 1 : }
393 1 : if m.mu.writer != nil {
394 1 : m.mu.writer.switchToNewDir(dir)
395 1 : }
396 1 : m.mu.Unlock()
397 : }
398 : }
399 1 : if m.opts.monitorStateForTesting != nil {
400 1 : m.opts.monitorStateForTesting(lastWriter.numSwitches, lastWriter.ongoingLatencyAtSwitch)
401 1 : }
402 1 : if m.opts.monitorIterationForTesting != nil {
403 1 : m.opts.monitorIterationForTesting <- struct{}{}
404 1 : }
405 : }
406 : }
407 :
408 : type logicalLogWithSizesEtc struct {
409 : num NumWAL
410 : segments []segmentWithSizeEtc
411 : }
412 :
413 : type segmentWithSizeEtc struct {
414 : segment
415 : approxFileSize uint64
416 : synchronouslyClosed bool
417 : }
418 :
419 : type failoverManager struct {
420 : opts Options
421 : // initialObsolete holds the set of DeletableLogs that formed the logs
422 : // passed into Init. The initialObsolete logs are all obsolete. Once
423 : // returned via Manager.Obsolete, initialObsolete is cleared. The
424 : // initialObsolete logs are stored separately from mu.queue because they may
425 : // include logs that were NOT created by the standalone manager, and
426 : // multiple physical log files may form one logical WAL.
427 : initialObsolete []DeletableLog
428 :
429 : // TODO(jackson/sumeer): read-path etc.
430 :
431 : dirHandles [numDirIndices]vfs.File
432 : stopper *stopper
433 : monitor *failoverMonitor
434 : mu struct {
435 : sync.Mutex
436 : closedWALs []logicalLogWithSizesEtc
437 : ww *failoverWriter
438 : }
439 : recycler LogRecycler
440 : // Due to async creation of files in failoverWriter, multiple goroutines can
441 : // concurrently try to get a file from the recycler. This mutex protects the
442 : // logRecycler.{Peek,Pop} pair.
443 : recyclerPeekPopMu sync.Mutex
444 : }
445 :
446 : var _ Manager = &failoverManager{}
447 :
448 : // TODO(sumeer):
449 : // - log deletion: if record.LogWriter did not close yet, the cleaner may
450 : // get an error when deleting or renaming (only under windows?).
451 :
452 : // init implements Manager.
453 1 : func (wm *failoverManager) init(o Options, initial Logs) error {
454 1 : if o.timeSource == nil {
455 1 : o.timeSource = defaultTime{}
456 1 : }
457 1 : o.FailoverOptions.EnsureDefaults()
458 1 : stopper := newStopper()
459 1 : var dirs [numDirIndices]dirAndFileHandle
460 1 : for i, dir := range []Dir{o.Primary, o.Secondary} {
461 1 : dirs[i].Dir = dir
462 1 : f, err := dir.FS.OpenDir(dir.Dirname)
463 1 : if err != nil {
464 0 : return err
465 0 : }
466 1 : dirs[i].File = f
467 : }
468 1 : fmOpts := failoverMonitorOptions{
469 1 : dirs: dirs,
470 1 : FailoverOptions: o.FailoverOptions,
471 1 : stopper: stopper,
472 1 : }
473 1 : monitor := newFailoverMonitor(fmOpts)
474 1 : *wm = failoverManager{
475 1 : opts: o,
476 1 : dirHandles: [numDirIndices]vfs.File{dirs[primaryDirIndex].File, dirs[secondaryDirIndex].File},
477 1 : stopper: stopper,
478 1 : monitor: monitor,
479 1 : }
480 1 : wm.recycler.Init(o.MaxNumRecyclableLogs)
481 1 : for _, ll := range initial {
482 1 : if wm.recycler.MinRecycleLogNum() <= ll.Num {
483 1 : wm.recycler.SetMinRecycleLogNum(ll.Num + 1)
484 1 : }
485 1 : var err error
486 1 : wm.initialObsolete, err = appendDeletableLogs(wm.initialObsolete, ll)
487 1 : if err != nil {
488 0 : return err
489 0 : }
490 : }
491 1 : return nil
492 : }
493 :
494 : // List implements Manager.
495 1 : func (wm *failoverManager) List() (Logs, error) {
496 1 : wm.mu.Lock()
497 1 : defer wm.mu.Unlock()
498 1 : n := len(wm.mu.closedWALs)
499 1 : if wm.mu.ww != nil {
500 1 : n++
501 1 : }
502 1 : wals := make(Logs, n)
503 1 : setLogicalLog := func(index int, llse logicalLogWithSizesEtc) {
504 1 : segments := make([]segment, len(llse.segments))
505 1 : for j := range llse.segments {
506 1 : segments[j] = llse.segments[j].segment
507 1 : }
508 1 : wals[index] = LogicalLog{
509 1 : Num: llse.num,
510 1 : segments: segments,
511 1 : }
512 : }
513 1 : for i, llse := range wm.mu.closedWALs {
514 1 : setLogicalLog(i, llse)
515 1 : }
516 1 : if wm.mu.ww != nil {
517 1 : setLogicalLog(n-1, wm.mu.ww.getLog())
518 1 : }
519 1 : return wals, nil
520 : }
521 :
522 : // Obsolete implements Manager.
523 : func (wm *failoverManager) Obsolete(
524 : minUnflushedNum NumWAL, noRecycle bool,
525 1 : ) (toDelete []DeletableLog, err error) {
526 1 : wm.mu.Lock()
527 1 : defer wm.mu.Unlock()
528 1 :
529 1 : // If this is the first call to Obsolete after Open, we may have deletable
530 1 : // logs outside the queue.
531 1 : toDelete, wm.initialObsolete = wm.initialObsolete, nil
532 1 :
533 1 : i := 0
534 1 : for ; i < len(wm.mu.closedWALs); i++ {
535 1 : ll := wm.mu.closedWALs[i]
536 1 : if ll.num >= minUnflushedNum {
537 1 : break
538 : }
539 : // Recycle only the primary at logNameIndex=0, if there was no failover,
540 : // and synchronously closed. It may not be safe to recycle a file that is
541 : // still being written to. And recycling when there was a failover may
542 : // fill up the recycler with smaller log files. The restriction regarding
543 : // logNameIndex=0 is because logRecycler.Peek only exposes the
544 : // DiskFileNum, and we need to use that to construct the path -- we could
545 : // remove this restriction by changing the logRecycler interface, but we
546 : // don't bother.
547 1 : canRecycle := !noRecycle && len(ll.segments) == 1 && ll.segments[0].synchronouslyClosed &&
548 1 : ll.segments[0].logNameIndex == 0 &&
549 1 : ll.segments[0].dir == wm.opts.Primary
550 1 : if !canRecycle || !wm.recycler.Add(base.FileInfo{
551 1 : FileNum: base.DiskFileNum(ll.num),
552 1 : FileSize: ll.segments[0].approxFileSize,
553 1 : }) {
554 1 : for _, s := range ll.segments {
555 1 : toDelete = append(toDelete, DeletableLog{
556 1 : FS: s.dir.FS,
557 1 : Path: s.dir.FS.PathJoin(s.dir.Dirname, makeLogFilename(ll.num, s.logNameIndex)),
558 1 : NumWAL: ll.num,
559 1 : ApproxFileSize: s.approxFileSize,
560 1 : })
561 1 : }
562 : }
563 : }
564 1 : wm.mu.closedWALs = wm.mu.closedWALs[i:]
565 1 : return toDelete, nil
566 : }
567 :
568 : // Create implements Manager.
569 1 : func (wm *failoverManager) Create(wn NumWAL, jobID int) (Writer, error) {
570 1 : func() {
571 1 : wm.mu.Lock()
572 1 : defer wm.mu.Unlock()
573 1 : if wm.mu.ww != nil {
574 0 : panic("previous wal.Writer not closed")
575 : }
576 : }()
577 1 : fwOpts := failoverWriterOpts{
578 1 : wn: wn,
579 1 : logger: wm.opts.Logger,
580 1 : timeSource: wm.opts.timeSource,
581 1 : jobID: jobID,
582 1 : logCreator: wm.logCreator,
583 1 : noSyncOnClose: wm.opts.NoSyncOnClose,
584 1 : bytesPerSync: wm.opts.BytesPerSync,
585 1 : preallocateSize: wm.opts.PreallocateSize,
586 1 : minSyncInterval: wm.opts.MinSyncInterval,
587 1 : fsyncLatency: wm.opts.FsyncLatency,
588 1 : queueSemChan: wm.opts.QueueSemChan,
589 1 : stopper: wm.stopper,
590 1 : writerClosed: wm.writerClosed,
591 1 : writerCreatedForTest: wm.opts.logWriterCreatedForTesting,
592 1 : }
593 1 : var err error
594 1 : var ww *failoverWriter
595 1 : writerCreateFunc := func(dir dirAndFileHandle) switchableWriter {
596 1 : ww, err = newFailoverWriter(fwOpts, dir)
597 1 : if err != nil {
598 0 : return nil
599 0 : }
600 1 : return ww
601 : }
602 1 : wm.monitor.newWriter(writerCreateFunc)
603 1 : if ww != nil {
604 1 : wm.mu.Lock()
605 1 : defer wm.mu.Unlock()
606 1 : wm.mu.ww = ww
607 1 : }
608 1 : return ww, err
609 : }
610 :
611 : // ElevateWriteStallThresholdForFailover implements Manager.
612 1 : func (wm *failoverManager) ElevateWriteStallThresholdForFailover() bool {
613 1 : return wm.monitor.elevateWriteStallThresholdForFailover()
614 1 : }
615 :
616 1 : func (wm *failoverManager) writerClosed(llse logicalLogWithSizesEtc) {
617 1 : wm.monitor.noWriter()
618 1 : wm.mu.Lock()
619 1 : defer wm.mu.Unlock()
620 1 : wm.mu.closedWALs = append(wm.mu.closedWALs, llse)
621 1 : wm.mu.ww = nil
622 1 : }
623 :
624 : // Stats implements Manager.
625 1 : func (wm *failoverManager) Stats() Stats {
626 1 : obsoleteLogsCount, obsoleteLogSize := wm.recycler.Stats()
627 1 : failoverStats := wm.monitor.stats()
628 1 : wm.mu.Lock()
629 1 : defer wm.mu.Unlock()
630 1 : var liveFileCount int
631 1 : var liveFileSize uint64
632 1 : updateStats := func(segments []segmentWithSizeEtc) {
633 1 : for _, s := range segments {
634 1 : liveFileCount++
635 1 : liveFileSize += s.approxFileSize
636 1 : }
637 : }
638 1 : for _, llse := range wm.mu.closedWALs {
639 1 : updateStats(llse.segments)
640 1 : }
641 1 : if wm.mu.ww != nil {
642 1 : updateStats(wm.mu.ww.getLog().segments)
643 1 : }
644 1 : for i := range wm.initialObsolete {
645 1 : if i == 0 || wm.initialObsolete[i].NumWAL != wm.initialObsolete[i-1].NumWAL {
646 1 : obsoleteLogsCount++
647 1 : }
648 1 : obsoleteLogSize += wm.initialObsolete[i].ApproxFileSize
649 : }
650 1 : return Stats{
651 1 : ObsoleteFileCount: obsoleteLogsCount,
652 1 : ObsoleteFileSize: obsoleteLogSize,
653 1 : LiveFileCount: liveFileCount,
654 1 : LiveFileSize: liveFileSize,
655 1 : Failover: failoverStats,
656 1 : }
657 : }
658 :
659 : // Close implements Manager.
660 1 : func (wm *failoverManager) Close() error {
661 1 : wm.stopper.stop()
662 1 : // Since all goroutines are stopped, can close the dirs.
663 1 : var err error
664 1 : for _, f := range wm.dirHandles {
665 1 : err = firstError(err, f.Close())
666 1 : }
667 1 : return err
668 : }
669 :
670 : // RecyclerForTesting implements Manager.
671 0 : func (wm *failoverManager) RecyclerForTesting() *LogRecycler {
672 0 : return nil
673 0 : }
674 :
675 : // logCreator implements the logCreator func type.
676 : func (wm *failoverManager) logCreator(
677 : dir Dir, wn NumWAL, li LogNameIndex, r *latencyAndErrorRecorder, jobID int,
678 1 : ) (logFile vfs.File, initialFileSize uint64, err error) {
679 1 : logFilename := dir.FS.PathJoin(dir.Dirname, makeLogFilename(wn, li))
680 1 : isPrimary := dir == wm.opts.Primary
681 1 : // Only recycling when logNameIndex is 0 is a somewhat arbitrary choice.
682 1 : considerRecycle := li == 0 && isPrimary
683 1 : createInfo := CreateInfo{
684 1 : JobID: jobID,
685 1 : Path: logFilename,
686 1 : IsSecondary: !isPrimary,
687 1 : Num: wn,
688 1 : Err: nil,
689 1 : }
690 1 : defer func() {
691 1 : createInfo.Err = err
692 1 : if wm.opts.EventListener != nil {
693 1 : wm.opts.EventListener.LogCreated(createInfo)
694 1 : }
695 : }()
696 1 : if considerRecycle {
697 1 : // Try to use a recycled log file. Recycling log files is an important
698 1 : // performance optimization as it is faster to sync a file that has
699 1 : // already been written, than one which is being written for the first
700 1 : // time. This is due to the need to sync file metadata when a file is
701 1 : // being written for the first time. Note this is true even if file
702 1 : // preallocation is performed (e.g. fallocate).
703 1 : var recycleLog base.FileInfo
704 1 : var recycleOK bool
705 1 : func() {
706 1 : wm.recyclerPeekPopMu.Lock()
707 1 : defer wm.recyclerPeekPopMu.Unlock()
708 1 : recycleLog, recycleOK = wm.recycler.Peek()
709 1 : if recycleOK {
710 1 : if err = wm.recycler.Pop(recycleLog.FileNum); err != nil {
711 0 : panic(err)
712 : }
713 : }
714 : }()
715 1 : if recycleOK {
716 1 : createInfo.RecycledFileNum = recycleLog.FileNum
717 1 : recycleLogName := dir.FS.PathJoin(dir.Dirname, makeLogFilename(NumWAL(recycleLog.FileNum), 0))
718 1 : r.writeStart()
719 1 : logFile, err = dir.FS.ReuseForWrite(recycleLogName, logFilename)
720 1 : r.writeEnd(err)
721 1 : // TODO(sumeer): should we fatal since primary dir? At some point it is
722 1 : // better to fatal instead of continuing to failover.
723 1 : // base.MustExist(dir.FS, logFilename, wm.opts.Logger, err)
724 1 : if err != nil {
725 0 : // TODO(sumeer): we have popped from the logRecycler, which is
726 0 : // arguably correct, since we don't want to keep trying to reuse a log
727 0 : // that causes some error. But the original or new file may exist, and
728 0 : // no one will clean it up unless the process restarts.
729 0 : return nil, 0, err
730 0 : }
731 : // Figure out the recycled WAL size. This Stat is necessary because
732 : // ReuseForWrite's contract allows for removing the old file and
733 : // creating a new one. We don't know whether the WAL was actually
734 : // recycled.
735 : //
736 : // TODO(jackson): Adding a boolean to the ReuseForWrite return value
737 : // indicating whether or not the file was actually reused would allow us
738 : // to skip the stat and use recycleLog.FileSize.
739 1 : var finfo os.FileInfo
740 1 : finfo, err = logFile.Stat()
741 1 : if err != nil {
742 0 : logFile.Close()
743 0 : return nil, 0, err
744 0 : }
745 1 : initialFileSize = uint64(finfo.Size())
746 1 : return logFile, initialFileSize, nil
747 : }
748 : }
749 : // Did not recycle.
750 : //
751 : // Create file.
752 1 : r.writeStart()
753 1 : logFile, err = dir.FS.Create(logFilename)
754 1 : r.writeEnd(err)
755 1 : return logFile, 0, err
756 : }
757 :
758 : type stopper struct {
759 : quiescer chan struct{} // Closed when quiescing
760 : wg sync.WaitGroup
761 : }
762 :
763 1 : func newStopper() *stopper {
764 1 : return &stopper{
765 1 : quiescer: make(chan struct{}),
766 1 : }
767 1 : }
768 :
769 1 : func (s *stopper) runAsync(f func()) {
770 1 : s.wg.Add(1)
771 1 : go func() {
772 1 : f()
773 1 : s.wg.Done()
774 1 : }()
775 : }
776 :
777 : // shouldQuiesce returns a channel which will be closed when stop() has been
778 : // invoked and outstanding goroutines should begin to quiesce.
779 1 : func (s *stopper) shouldQuiesce() <-chan struct{} {
780 1 : return s.quiescer
781 1 : }
782 :
783 1 : func (s *stopper) stop() {
784 1 : close(s.quiescer)
785 1 : s.wg.Wait()
786 1 : }
787 :
788 : // timeSource and tickerI are extracted from CockroachDB's timeutil, with
789 : // removal of support for Timer, and added support in the manual
790 : // implementation for reset.
791 :
792 : // timeSource is used to interact with the clock and tickers. Abstracts
793 : // time.Now and time.NewTicker for testing.
794 : type timeSource interface {
795 : now() time.Time
796 : newTicker(duration time.Duration) tickerI
797 : }
798 :
799 : // tickerI is an interface wrapping time.Ticker.
800 : type tickerI interface {
801 : reset(duration time.Duration)
802 : stop()
803 : ch() <-chan time.Time
804 : }
805 :
806 : // defaultTime is a timeSource using the time package.
807 : type defaultTime struct{}
808 :
809 : var _ timeSource = defaultTime{}
810 :
811 1 : func (defaultTime) now() time.Time {
812 1 : return time.Now()
813 1 : }
814 :
815 1 : func (defaultTime) newTicker(duration time.Duration) tickerI {
816 1 : return (*defaultTicker)(time.NewTicker(duration))
817 1 : }
818 :
819 : // defaultTicker uses time.Ticker.
820 : type defaultTicker time.Ticker
821 :
822 : var _ tickerI = &defaultTicker{}
823 :
824 0 : func (t *defaultTicker) reset(duration time.Duration) {
825 0 : (*time.Ticker)(t).Reset(duration)
826 0 : }
827 :
828 1 : func (t *defaultTicker) stop() {
829 1 : (*time.Ticker)(t).Stop()
830 1 : }
831 :
832 1 : func (t *defaultTicker) ch() <-chan time.Time {
833 1 : return (*time.Ticker)(t).C
834 1 : }
835 :
836 : // Make lint happy.
837 : var _ = (*failoverMonitor).noWriter
838 : var _ = (*failoverManager).writerClosed
839 : var _ = (&stopper{}).shouldQuiesce
|
