db: don't hold DB.mu when annotating

While we expect annotations to be very cheap for the most part, they
can be expensive after startup when many of the tables don't have the
properties or stats populated.

Now that the annotation infrastructure is thread-safe, we can
annotate without holding the DB mutex.

Author: RaduBerinde

db.go

@@ -506,9 +506,10 @@ type DB struct {
 			// validating is set to true when validation is running.
 			validating bool
 		}
-		fileSizeAnnotator manifest.TableAnnotator[fileSizeByBacking]
 	}
 
+	fileSizeAnnotator manifest.TableAnnotator[fileSizeByBacking]
+
 	// problemSpans keeps track of spans of keys within LSM levels where
 	// compactions have failed; used to avoid retrying these compactions too
 	// quickly.
@@ -1566,17 +1567,14 @@ func (d *DB) Close() error {
 	// Wait for all cleaning jobs to finish.
 	d.cleanupManager.Close()
 
-	// Sanity check metrics.
+	d.mu.Lock()
+	// Sanity check compaction metrics.
 	if invariants.Enabled {
-		m := d.Metrics()
-		if m.Compact.NumInProgress > 0 || m.Compact.InProgressBytes > 0 {
-			d.mu.Lock()
-			panic(fmt.Sprintf("invalid metrics on close:\n%s", m))
+		if d.mu.compact.compactingCount > 0 || d.mu.compact.downloadingCount > 0 || d.mu.versions.atomicInProgressBytes.Load() > 0 {
+			panic("compacting counts not 0 on close")
 		}
 	}
 
-	d.mu.Lock()
-
 	// As a sanity check, ensure that there are no zombie tables or blob files.
 	// A non-zero count hints at a reference count leak.
 	if ztbls := d.mu.versions.zombieTables.Count(); ztbls > 0 {
@@ -1854,6 +1852,9 @@ func (d *DB) Metrics() *Metrics {
 
 	d.mu.Lock()
 	vers := d.mu.versions.currentVersion()
+	vers.Ref()
+	defer vers.Unref()
+
 	*metrics = d.mu.versions.metrics
 	metrics.Compact.EstimatedDebt = d.mu.versions.picker.estimatedCompactionDebt()
 	metrics.Compact.InProgressBytes = d.mu.versions.atomicInProgressBytes.Load()
@@ -1936,15 +1937,6 @@ func (d *DB) Metrics() *Metrics {
 	metrics.BlobFiles.ObsoleteSize += pendingObsoleteFileStats.blobFilesAll.size
 	metrics.private.optionsFileSize = d.optionsFileSize
 
-	// TODO(jackson): Consider making these metrics optional.
-	aggProps := tablePropsAnnotator.MultiLevelAnnotation(vers.Levels[:])
-	metrics.Keys.RangeKeySetsCount = aggProps.NumRangeKeySets
-	metrics.Keys.TombstoneCount = aggProps.NumDeletions
-
-	delBytes := deletionBytesAnnotator.MultiLevelAnnotation(vers.Levels[:])
-	metrics.Table.Garbage.PointDeletionsBytesEstimate = delBytes.PointDels
-	metrics.Table.Garbage.RangeDeletionsBytesEstimate = delBytes.RangeDels
-
 	d.mu.versions.logLock()
 	metrics.private.manifestFileSize = uint64(d.mu.versions.manifest.Size())
 	backingCount, backingTotalSize := d.mu.versions.latest.virtualBackings.Stats()
@@ -1969,6 +1961,17 @@ func (d *DB) Metrics() *Metrics {
 	metrics.Table.PendingStatsCollectionCount = int64(len(d.mu.tableStats.pending))
 	metrics.Table.InitialStatsCollectionComplete = d.mu.tableStats.loadedInitial
 
+	d.mu.Unlock()
+
+	// TODO(jackson): Consider making these metrics optional.
+	aggProps := tablePropsAnnotator.MultiLevelAnnotation(vers.Levels[:])
+	metrics.Keys.RangeKeySetsCount = aggProps.NumRangeKeySets
+	metrics.Keys.TombstoneCount = aggProps.NumDeletions
+
+	delBytes := deletionBytesAnnotator.MultiLevelAnnotation(vers.Levels[:])
+	metrics.Table.Garbage.PointDeletionsBytesEstimate = delBytes.PointDels
+	metrics.Table.Garbage.RangeDeletionsBytesEstimate = delBytes.RangeDels
+
 	for i := 0; i < numLevels; i++ {
 		aggProps := tablePropsAnnotator.LevelAnnotation(vers.Levels[i])
 		metrics.Levels[i].Additional.ValueBlocksSize = aggProps.ValueBlocksSize
@@ -1978,8 +1981,6 @@ func (d *DB) Metrics() *Metrics {
 	blobCompressionMetrics := blobCompressionStatsAnnotator.Annotation(&vers.BlobFiles)
 	metrics.BlobFiles.Compression.MergeWith(&blobCompressionMetrics)
 
-	d.mu.Unlock()
-
 	metrics.BlockCache = d.opts.Cache.Metrics()
 	metrics.FileCache, metrics.Filter = d.fileCache.Metrics()
 	metrics.TableIters = d.fileCache.IterCount()

disk_usage.go

@@ -48,7 +48,7 @@ func (d *DB) EstimateDiskUsageByBackingType(
 	readState := d.loadReadState()
 	defer readState.unref()
 
-	sizes := d.mu.fileSizeAnnotator.VersionRangeAnnotation(readState.current, bounds)
+	sizes := d.fileSizeAnnotator.VersionRangeAnnotation(readState.current, bounds)
 	return sizes.totalSize, sizes.remoteSize, sizes.externalSize, nil
 }
 

open.go

@@ -442,7 +442,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 	d.newIters = d.fileCache.newIters
 	d.tableNewRangeKeyIter = tableNewRangeKeyIter(d.newIters)
 
-	d.mu.fileSizeAnnotator = d.makeFileSizeAnnotator()
+	d.fileSizeAnnotator = d.makeFileSizeAnnotator()
 
 	var previousOptionsFileNum base.DiskFileNum
 	var previousOptionsFilename string