Line data Source code
1 : // Copyright 2019 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 pebble
6 :
7 : import (
8 : "cmp"
9 : "context"
10 : "runtime/pprof"
11 : "slices"
12 : "sync"
13 : "time"
14 :
15 : "github.com/cockroachdb/crlib/crtime"
16 : "github.com/cockroachdb/errors/oserror"
17 : "github.com/cockroachdb/pebble/internal/base"
18 : "github.com/cockroachdb/pebble/objstorage"
19 : "github.com/cockroachdb/pebble/vfs"
20 : "github.com/cockroachdb/pebble/wal"
21 : "github.com/cockroachdb/tokenbucket"
22 : )
23 :
24 : // Cleaner exports the base.Cleaner type.
25 : type Cleaner = base.Cleaner
26 :
27 : // DeleteCleaner exports the base.DeleteCleaner type.
28 : type DeleteCleaner = base.DeleteCleaner
29 :
30 : // ArchiveCleaner exports the base.ArchiveCleaner type.
31 : type ArchiveCleaner = base.ArchiveCleaner
32 :
33 : type cleanupManager struct {
34 : opts *Options
35 : objProvider objstorage.Provider
36 : onTableDeleteFn func(fileSize uint64, isLocal bool)
37 : deletePacer *deletionPacer
38 :
39 : // jobsCh is used as the cleanup job queue.
40 : jobsCh chan *cleanupJob
41 : // waitGroup is used to wait for the background goroutine to exit.
42 : waitGroup sync.WaitGroup
43 :
44 : mu struct {
45 : sync.Mutex
46 : // totalJobs is the total number of enqueued jobs (completed or in progress).
47 : totalJobs int
48 : completedJobs int
49 : completedJobsCond sync.Cond
50 : jobsQueueWarningIssued bool
51 : }
52 : }
53 :
54 : // We can queue this many jobs before we have to block EnqueueJob.
55 : const jobsQueueDepth = 1000
56 :
57 : // deletableFile is used for non log files.
58 : type deletableFile struct {
59 : dir string
60 : fileNum base.DiskFileNum
61 : fileSize uint64
62 : isLocal bool
63 : }
64 :
65 : // obsoleteFile holds information about a file that needs to be deleted soon.
66 : type obsoleteFile struct {
67 : fileType fileType
68 : // nonLogFile is populated when fileType != fileTypeLog.
69 : nonLogFile deletableFile
70 : // logFile is populated when fileType == fileTypeLog.
71 : logFile wal.DeletableLog
72 : }
73 :
74 : type cleanupJob struct {
75 : jobID JobID
76 : obsoleteFiles []obsoleteFile
77 : }
78 :
79 : // openCleanupManager creates a cleanupManager and starts its background goroutine.
80 : // The cleanupManager must be Close()d.
81 : func openCleanupManager(
82 : opts *Options,
83 : objProvider objstorage.Provider,
84 : onTableDeleteFn func(fileSize uint64, isLocal bool),
85 : getDeletePacerInfo func() deletionPacerInfo,
86 2 : ) *cleanupManager {
87 2 : cm := &cleanupManager{
88 2 : opts: opts,
89 2 : objProvider: objProvider,
90 2 : onTableDeleteFn: onTableDeleteFn,
91 2 : deletePacer: newDeletionPacer(crtime.NowMono(), int64(opts.TargetByteDeletionRate), getDeletePacerInfo),
92 2 : jobsCh: make(chan *cleanupJob, jobsQueueDepth),
93 2 : }
94 2 : cm.mu.completedJobsCond.L = &cm.mu.Mutex
95 2 : cm.waitGroup.Add(1)
96 2 :
97 2 : go func() {
98 2 : pprof.Do(context.Background(), gcLabels, func(context.Context) {
99 2 : cm.mainLoop()
100 2 : })
101 : }()
102 :
103 2 : return cm
104 : }
105 :
106 : // Close stops the background goroutine, waiting until all queued jobs are completed.
107 : // Delete pacing is disabled for the remaining jobs.
108 2 : func (cm *cleanupManager) Close() {
109 2 : close(cm.jobsCh)
110 2 : cm.waitGroup.Wait()
111 2 : }
112 :
113 : // EnqueueJob adds a cleanup job to the manager's queue.
114 2 : func (cm *cleanupManager) EnqueueJob(jobID JobID, obsoleteFiles []obsoleteFile) {
115 2 : job := &cleanupJob{
116 2 : jobID: jobID,
117 2 : obsoleteFiles: obsoleteFiles,
118 2 : }
119 2 :
120 2 : // Report deleted bytes to the pacer, which can use this data to potentially
121 2 : // increase the deletion rate to keep up. We want to do this at enqueue time
122 2 : // rather than when we get to the job, otherwise the reported bytes will be
123 2 : // subject to the throttling rate which defeats the purpose.
124 2 : var pacingBytes uint64
125 2 : for _, of := range obsoleteFiles {
126 2 : if cm.needsPacing(of.fileType, of.nonLogFile.fileNum) {
127 2 : pacingBytes += of.nonLogFile.fileSize
128 2 : }
129 : }
130 2 : if pacingBytes > 0 {
131 2 : cm.deletePacer.ReportDeletion(crtime.NowMono(), pacingBytes)
132 2 : }
133 :
134 2 : cm.mu.Lock()
135 2 : cm.mu.totalJobs++
136 2 : cm.maybeLogLocked()
137 2 : cm.mu.Unlock()
138 2 :
139 2 : cm.jobsCh <- job
140 : }
141 :
142 : // Wait until the completion of all jobs that were already queued.
143 : //
144 : // Does not wait for jobs that are enqueued during the call.
145 : //
146 : // Note that DB.mu should not be held while calling this method; the background
147 : // goroutine needs to acquire DB.mu to update deleted table metrics.
148 2 : func (cm *cleanupManager) Wait() {
149 2 : cm.mu.Lock()
150 2 : defer cm.mu.Unlock()
151 2 : n := cm.mu.totalJobs
152 2 : for cm.mu.completedJobs < n {
153 2 : cm.mu.completedJobsCond.Wait()
154 2 : }
155 : }
156 :
157 : // mainLoop runs the manager's background goroutine.
158 2 : func (cm *cleanupManager) mainLoop() {
159 2 : defer cm.waitGroup.Done()
160 2 :
161 2 : var tb tokenbucket.TokenBucket
162 2 : // Use a token bucket with 1 token / second refill rate and 1 token burst.
163 2 : tb.Init(1.0, 1.0)
164 2 : for job := range cm.jobsCh {
165 2 : for _, of := range job.obsoleteFiles {
166 2 : switch of.fileType {
167 2 : case fileTypeTable:
168 2 : cm.maybePace(&tb, of.fileType, of.nonLogFile.fileNum, of.nonLogFile.fileSize)
169 2 : cm.onTableDeleteFn(of.nonLogFile.fileSize, of.nonLogFile.isLocal)
170 2 : cm.deleteObsoleteObject(fileTypeTable, job.jobID, of.nonLogFile.fileNum)
171 2 : case fileTypeLog:
172 2 : cm.deleteObsoleteFile(of.logFile.FS, fileTypeLog, job.jobID, of.logFile.Path,
173 2 : base.DiskFileNum(of.logFile.NumWAL), of.logFile.ApproxFileSize)
174 2 : default:
175 2 : path := base.MakeFilepath(cm.opts.FS, of.nonLogFile.dir, of.fileType, of.nonLogFile.fileNum)
176 2 : cm.deleteObsoleteFile(
177 2 : cm.opts.FS, of.fileType, job.jobID, path, of.nonLogFile.fileNum, of.nonLogFile.fileSize)
178 : }
179 : }
180 2 : cm.mu.Lock()
181 2 : cm.mu.completedJobs++
182 2 : cm.mu.completedJobsCond.Broadcast()
183 2 : cm.maybeLogLocked()
184 2 : cm.mu.Unlock()
185 : }
186 : }
187 :
188 : // fileNumIfSST is read iff fileType is fileTypeTable.
189 2 : func (cm *cleanupManager) needsPacing(fileType base.FileType, fileNumIfSST base.DiskFileNum) bool {
190 2 : if fileType != fileTypeTable {
191 2 : return false
192 2 : }
193 2 : meta, err := cm.objProvider.Lookup(fileType, fileNumIfSST)
194 2 : if err != nil {
195 2 : // The object was already removed from the provider; we won't actually
196 2 : // delete anything, so we don't need to pace.
197 2 : return false
198 2 : }
199 : // Don't throttle deletion of remote objects.
200 2 : return !meta.IsRemote()
201 : }
202 :
203 : // maybePace sleeps before deleting an object if appropriate. It is always
204 : // called from the background goroutine.
205 : func (cm *cleanupManager) maybePace(
206 : tb *tokenbucket.TokenBucket, fileType base.FileType, fileNum base.DiskFileNum, fileSize uint64,
207 2 : ) {
208 2 : if !cm.needsPacing(fileType, fileNum) {
209 2 : return
210 2 : }
211 :
212 2 : tokens := cm.deletePacer.PacingDelay(crtime.NowMono(), fileSize)
213 2 : if tokens == 0.0 {
214 2 : // The token bucket might be in debt; it could make us wait even for 0
215 2 : // tokens. We don't want that if the pacer decided throttling should be
216 2 : // disabled.
217 2 : return
218 2 : }
219 : // Wait for tokens. We use a token bucket instead of sleeping outright because
220 : // the token bucket accumulates up to one second of unused tokens.
221 2 : for {
222 2 : ok, d := tb.TryToFulfill(tokenbucket.Tokens(tokens))
223 2 : if ok {
224 2 : break
225 : }
226 1 : time.Sleep(d)
227 : }
228 : }
229 :
230 : // deleteObsoleteFile deletes a (non-object) file that is no longer needed.
231 : func (cm *cleanupManager) deleteObsoleteFile(
232 : fs vfs.FS, fileType fileType, jobID JobID, path string, fileNum base.DiskFileNum, fileSize uint64,
233 2 : ) {
234 2 : // TODO(peter): need to handle this error, probably by re-adding the
235 2 : // file that couldn't be deleted to one of the obsolete slices map.
236 2 : err := cm.opts.Cleaner.Clean(fs, fileType, path)
237 2 : if oserror.IsNotExist(err) {
238 1 : return
239 1 : }
240 :
241 2 : switch fileType {
242 2 : case fileTypeLog:
243 2 : cm.opts.EventListener.WALDeleted(WALDeleteInfo{
244 2 : JobID: int(jobID),
245 2 : Path: path,
246 2 : FileNum: fileNum,
247 2 : Err: err,
248 2 : })
249 2 : case fileTypeManifest:
250 2 : cm.opts.EventListener.ManifestDeleted(ManifestDeleteInfo{
251 2 : JobID: int(jobID),
252 2 : Path: path,
253 2 : FileNum: fileNum,
254 2 : Err: err,
255 2 : })
256 0 : case fileTypeTable:
257 0 : panic("invalid deletion of object file")
258 : }
259 : }
260 :
261 : func (cm *cleanupManager) deleteObsoleteObject(
262 : fileType fileType, jobID JobID, fileNum base.DiskFileNum,
263 2 : ) {
264 2 : if fileType != fileTypeTable {
265 0 : panic("not an object")
266 : }
267 :
268 2 : var path string
269 2 : meta, err := cm.objProvider.Lookup(fileType, fileNum)
270 2 : if err != nil {
271 2 : path = "<nil>"
272 2 : } else {
273 2 : path = cm.objProvider.Path(meta)
274 2 : err = cm.objProvider.Remove(fileType, fileNum)
275 2 : }
276 2 : if cm.objProvider.IsNotExistError(err) {
277 2 : return
278 2 : }
279 :
280 2 : switch fileType {
281 2 : case fileTypeTable:
282 2 : cm.opts.EventListener.TableDeleted(TableDeleteInfo{
283 2 : JobID: int(jobID),
284 2 : Path: path,
285 2 : FileNum: fileNum,
286 2 : Err: err,
287 2 : })
288 : }
289 : }
290 :
291 : // maybeLogLocked issues a log if the job queue gets 75% full and issues a log
292 : // when the job queue gets back to less than 10% full.
293 : //
294 : // Must be called with cm.mu locked.
295 2 : func (cm *cleanupManager) maybeLogLocked() {
296 2 : const highThreshold = jobsQueueDepth * 3 / 4
297 2 : const lowThreshold = jobsQueueDepth / 10
298 2 :
299 2 : jobsInQueue := cm.mu.totalJobs - cm.mu.completedJobs
300 2 :
301 2 : if !cm.mu.jobsQueueWarningIssued && jobsInQueue > highThreshold {
302 0 : cm.mu.jobsQueueWarningIssued = true
303 0 : cm.opts.Logger.Infof("cleanup falling behind; job queue has over %d jobs", highThreshold)
304 0 : }
305 :
306 2 : if cm.mu.jobsQueueWarningIssued && jobsInQueue < lowThreshold {
307 0 : cm.mu.jobsQueueWarningIssued = false
308 0 : cm.opts.Logger.Infof("cleanup back to normal; job queue has under %d jobs", lowThreshold)
309 0 : }
310 : }
311 :
312 2 : func (d *DB) getDeletionPacerInfo() deletionPacerInfo {
313 2 : var pacerInfo deletionPacerInfo
314 2 : // Call GetDiskUsage after every file deletion. This may seem inefficient,
315 2 : // but in practice this was observed to take constant time, regardless of
316 2 : // volume size used, at least on linux with ext4 and zfs. All invocations
317 2 : // take 10 microseconds or less.
318 2 : pacerInfo.freeBytes = d.calculateDiskAvailableBytes()
319 2 : d.mu.Lock()
320 2 : pacerInfo.obsoleteBytes = d.mu.versions.metrics.Table.ObsoleteSize
321 2 : pacerInfo.liveBytes = uint64(d.mu.versions.metrics.Total().Size)
322 2 : d.mu.Unlock()
323 2 : return pacerInfo
324 2 : }
325 :
326 : // onObsoleteTableDelete is called to update metrics when an sstable is deleted.
327 2 : func (d *DB) onObsoleteTableDelete(fileSize uint64, isLocal bool) {
328 2 : d.mu.Lock()
329 2 : d.mu.versions.metrics.Table.ObsoleteCount--
330 2 : d.mu.versions.metrics.Table.ObsoleteSize -= fileSize
331 2 : if isLocal {
332 2 : d.mu.versions.metrics.Table.Local.ObsoleteSize -= fileSize
333 2 : }
334 2 : d.mu.Unlock()
335 : }
336 :
337 : // scanObsoleteFiles scans the filesystem for files that are no longer needed
338 : // and adds those to the internal lists of obsolete files. Note that the files
339 : // are not actually deleted by this method. A subsequent call to
340 : // deleteObsoleteFiles must be performed. Must be not be called concurrently
341 : // with compactions and flushes. db.mu must be held when calling this function.
342 2 : func (d *DB) scanObsoleteFiles(list []string, flushableIngests []*ingestedFlushable) {
343 2 : // Disable automatic compactions temporarily to avoid concurrent compactions /
344 2 : // flushes from interfering. The original value is restored on completion.
345 2 : disabledPrev := d.opts.DisableAutomaticCompactions
346 2 : defer func() {
347 2 : d.opts.DisableAutomaticCompactions = disabledPrev
348 2 : }()
349 2 : d.opts.DisableAutomaticCompactions = true
350 2 :
351 2 : // Wait for any ongoing compaction to complete before continuing.
352 2 : for d.mu.compact.compactingCount > 0 || d.mu.compact.downloadingCount > 0 || d.mu.compact.flushing {
353 2 : d.mu.compact.cond.Wait()
354 2 : }
355 :
356 2 : liveFileNums := make(map[base.DiskFileNum]struct{})
357 2 : d.mu.versions.addLiveFileNums(liveFileNums)
358 2 : // Protect against files which are only referred to by the ingestedFlushable
359 2 : // from being deleted. These are added to the flushable queue on WAL replay
360 2 : // and handle their own obsoletion/deletion. We exclude them from this obsolete
361 2 : // file scan to avoid double-deleting these files.
362 2 : for _, f := range flushableIngests {
363 2 : for _, file := range f.files {
364 2 : liveFileNums[file.FileBacking.DiskFileNum] = struct{}{}
365 2 : }
366 : }
367 :
368 2 : manifestFileNum := d.mu.versions.manifestFileNum
369 2 :
370 2 : var obsoleteTables []tableInfo
371 2 : var obsoleteManifests []fileInfo
372 2 : var obsoleteOptions []fileInfo
373 2 :
374 2 : for _, filename := range list {
375 2 : fileType, diskFileNum, ok := base.ParseFilename(d.opts.FS, filename)
376 2 : if !ok {
377 2 : continue
378 : }
379 2 : switch fileType {
380 2 : case fileTypeManifest:
381 2 : if diskFileNum >= manifestFileNum {
382 2 : continue
383 : }
384 2 : fi := fileInfo{FileNum: diskFileNum}
385 2 : if stat, err := d.opts.FS.Stat(filename); err == nil {
386 1 : fi.FileSize = uint64(stat.Size())
387 1 : }
388 2 : obsoleteManifests = append(obsoleteManifests, fi)
389 2 : case fileTypeOptions:
390 2 : if diskFileNum >= d.optionsFileNum {
391 2 : continue
392 : }
393 2 : fi := fileInfo{FileNum: diskFileNum}
394 2 : if stat, err := d.opts.FS.Stat(filename); err == nil {
395 1 : fi.FileSize = uint64(stat.Size())
396 1 : }
397 2 : obsoleteOptions = append(obsoleteOptions, fi)
398 2 : case fileTypeTable:
399 : // Objects are handled through the objstorage provider below.
400 2 : default:
401 : // Don't delete files we don't know about.
402 : }
403 : }
404 :
405 2 : objects := d.objProvider.List()
406 2 : for _, obj := range objects {
407 2 : switch obj.FileType {
408 2 : case fileTypeTable:
409 2 : if _, ok := liveFileNums[obj.DiskFileNum]; ok {
410 2 : continue
411 : }
412 2 : fileInfo := fileInfo{
413 2 : FileNum: obj.DiskFileNum,
414 2 : }
415 2 : if size, err := d.objProvider.Size(obj); err == nil {
416 2 : fileInfo.FileSize = uint64(size)
417 2 : }
418 2 : obsoleteTables = append(obsoleteTables, tableInfo{
419 2 : fileInfo: fileInfo,
420 2 : isLocal: !obj.IsRemote(),
421 2 : })
422 :
423 0 : default:
424 : // Ignore object types we don't know about.
425 : }
426 : }
427 :
428 2 : d.mu.versions.obsoleteTables = mergeTableInfos(d.mu.versions.obsoleteTables, obsoleteTables)
429 2 : d.mu.versions.updateObsoleteTableMetricsLocked()
430 2 : d.mu.versions.obsoleteManifests = merge(d.mu.versions.obsoleteManifests, obsoleteManifests)
431 2 : d.mu.versions.obsoleteOptions = merge(d.mu.versions.obsoleteOptions, obsoleteOptions)
432 : }
433 :
434 : // disableFileDeletions disables file deletions and then waits for any
435 : // in-progress deletion to finish. The caller is required to call
436 : // enableFileDeletions in order to enable file deletions again. It is ok for
437 : // multiple callers to disable file deletions simultaneously, though they must
438 : // all invoke enableFileDeletions in order for file deletions to be re-enabled
439 : // (there is an internal reference count on file deletion disablement).
440 : //
441 : // d.mu must be held when calling this method.
442 2 : func (d *DB) disableFileDeletions() {
443 2 : d.mu.disableFileDeletions++
444 2 : d.mu.Unlock()
445 2 : defer d.mu.Lock()
446 2 : d.cleanupManager.Wait()
447 2 : }
448 :
449 : // enableFileDeletions enables previously disabled file deletions. A cleanup job
450 : // is queued if necessary.
451 : //
452 : // d.mu must be held when calling this method.
453 2 : func (d *DB) enableFileDeletions() {
454 2 : if d.mu.disableFileDeletions <= 0 {
455 1 : panic("pebble: file deletion disablement invariant violated")
456 : }
457 2 : d.mu.disableFileDeletions--
458 2 : if d.mu.disableFileDeletions > 0 {
459 0 : return
460 0 : }
461 2 : d.deleteObsoleteFiles(d.newJobIDLocked())
462 : }
463 :
464 : type fileInfo = base.FileInfo
465 :
466 : // deleteObsoleteFiles enqueues a cleanup job to the cleanup manager, if necessary.
467 : //
468 : // d.mu must be held when calling this. The function will release and re-aquire the mutex.
469 : //
470 : // Does nothing if file deletions are disabled (see disableFileDeletions). A
471 : // cleanup job will be scheduled when file deletions are re-enabled.
472 2 : func (d *DB) deleteObsoleteFiles(jobID JobID) {
473 2 : if d.mu.disableFileDeletions > 0 {
474 2 : return
475 2 : }
476 2 : _, noRecycle := d.opts.Cleaner.(base.NeedsFileContents)
477 2 :
478 2 : // NB: d.mu.versions.minUnflushedLogNum is the log number of the earliest
479 2 : // log that has not had its contents flushed to an sstable.
480 2 : obsoleteLogs, err := d.mu.log.manager.Obsolete(wal.NumWAL(d.mu.versions.minUnflushedLogNum), noRecycle)
481 2 : if err != nil {
482 0 : panic(err)
483 : }
484 :
485 2 : obsoleteTables := append([]tableInfo(nil), d.mu.versions.obsoleteTables...)
486 2 : d.mu.versions.obsoleteTables = nil
487 2 :
488 2 : for _, tbl := range obsoleteTables {
489 2 : delete(d.mu.versions.zombieTables, tbl.FileNum)
490 2 : }
491 :
492 : // Sort the manifests cause we want to delete some contiguous prefix
493 : // of the older manifests.
494 2 : slices.SortFunc(d.mu.versions.obsoleteManifests, func(a, b fileInfo) int {
495 2 : return cmp.Compare(a.FileNum, b.FileNum)
496 2 : })
497 :
498 2 : var obsoleteManifests []fileInfo
499 2 : manifestsToDelete := len(d.mu.versions.obsoleteManifests) - d.opts.NumPrevManifest
500 2 : if manifestsToDelete > 0 {
501 2 : obsoleteManifests = d.mu.versions.obsoleteManifests[:manifestsToDelete]
502 2 : d.mu.versions.obsoleteManifests = d.mu.versions.obsoleteManifests[manifestsToDelete:]
503 2 : if len(d.mu.versions.obsoleteManifests) == 0 {
504 0 : d.mu.versions.obsoleteManifests = nil
505 0 : }
506 : }
507 :
508 2 : obsoleteOptions := d.mu.versions.obsoleteOptions
509 2 : d.mu.versions.obsoleteOptions = nil
510 2 :
511 2 : // Release d.mu while preparing the cleanup job and possibly waiting.
512 2 : // Note the unusual order: Unlock and then Lock.
513 2 : d.mu.Unlock()
514 2 : defer d.mu.Lock()
515 2 :
516 2 : filesToDelete := make([]obsoleteFile, 0, len(obsoleteLogs)+len(obsoleteTables)+len(obsoleteManifests)+len(obsoleteOptions))
517 2 : for _, f := range obsoleteLogs {
518 2 : filesToDelete = append(filesToDelete, obsoleteFile{fileType: fileTypeLog, logFile: f})
519 2 : }
520 : // We sort to make the order of deletions deterministic, which is nice for
521 : // tests.
522 2 : slices.SortFunc(obsoleteTables, func(a, b tableInfo) int {
523 2 : return cmp.Compare(a.FileNum, b.FileNum)
524 2 : })
525 2 : for _, f := range obsoleteTables {
526 2 : d.fileCache.evict(f.FileNum)
527 2 : filesToDelete = append(filesToDelete, obsoleteFile{
528 2 : fileType: fileTypeTable,
529 2 : nonLogFile: deletableFile{
530 2 : dir: d.dirname,
531 2 : fileNum: f.FileNum,
532 2 : fileSize: f.FileSize,
533 2 : isLocal: f.isLocal,
534 2 : },
535 2 : })
536 2 : }
537 2 : files := [2]struct {
538 2 : fileType fileType
539 2 : obsolete []fileInfo
540 2 : }{
541 2 : {fileTypeManifest, obsoleteManifests},
542 2 : {fileTypeOptions, obsoleteOptions},
543 2 : }
544 2 : for _, f := range files {
545 2 : // We sort to make the order of deletions deterministic, which is nice for
546 2 : // tests.
547 2 : slices.SortFunc(f.obsolete, func(a, b fileInfo) int {
548 2 : return cmp.Compare(a.FileNum, b.FileNum)
549 2 : })
550 2 : for _, fi := range f.obsolete {
551 2 : dir := d.dirname
552 2 : filesToDelete = append(filesToDelete, obsoleteFile{
553 2 : fileType: f.fileType,
554 2 : nonLogFile: deletableFile{
555 2 : dir: dir,
556 2 : fileNum: fi.FileNum,
557 2 : fileSize: fi.FileSize,
558 2 : isLocal: true,
559 2 : },
560 2 : })
561 2 : }
562 : }
563 2 : if len(filesToDelete) > 0 {
564 2 : d.cleanupManager.EnqueueJob(jobID, filesToDelete)
565 2 : }
566 2 : if d.opts.private.testingAlwaysWaitForCleanup {
567 1 : d.cleanupManager.Wait()
568 1 : }
569 : }
570 :
571 2 : func (d *DB) maybeScheduleObsoleteTableDeletion() {
572 2 : d.mu.Lock()
573 2 : defer d.mu.Unlock()
574 2 : d.maybeScheduleObsoleteTableDeletionLocked()
575 2 : }
576 :
577 2 : func (d *DB) maybeScheduleObsoleteTableDeletionLocked() {
578 2 : if len(d.mu.versions.obsoleteTables) > 0 {
579 2 : d.deleteObsoleteFiles(d.newJobIDLocked())
580 2 : }
581 : }
582 :
583 2 : func merge(a, b []fileInfo) []fileInfo {
584 2 : if len(b) == 0 {
585 2 : return a
586 2 : }
587 :
588 2 : a = append(a, b...)
589 2 : slices.SortFunc(a, func(a, b fileInfo) int {
590 2 : return cmp.Compare(a.FileNum, b.FileNum)
591 2 : })
592 2 : return slices.CompactFunc(a, func(a, b fileInfo) bool {
593 2 : return a.FileNum == b.FileNum
594 2 : })
595 : }
596 :
597 2 : func mergeTableInfos(a, b []tableInfo) []tableInfo {
598 2 : if len(b) == 0 {
599 2 : return a
600 2 : }
601 :
602 2 : a = append(a, b...)
603 2 : slices.SortFunc(a, func(a, b tableInfo) int {
604 2 : return cmp.Compare(a.FileNum, b.FileNum)
605 2 : })
606 2 : return slices.CompactFunc(a, func(a, b tableInfo) bool {
607 2 : return a.FileNum == b.FileNum
608 2 : })
609 : }
|
