FDBCORE-3674 adding delta file count before resnapshot to minimize impact of small… (apple#355)

Author: sfc-gh-jslocum

fdbserver/BlobWorker.actor.cpp

@@ -1847,6 +1847,7 @@ Version doGranuleRollback(Reference<GranuleMetadata> metadata,
 		int toPop = 0;
 		// keep bytes in delta files pending here, then add back already durable delta files at end
 		metadata->bytesInNewDeltaFiles = 0;
+		metadata->newDeltaFileCount = 0;
 
 		for (auto& f : inFlightFiles) {
 			if (f.snapshot) {
@@ -1863,6 +1864,7 @@ Version doGranuleRollback(Reference<GranuleMetadata> metadata,
 				} else {
 					metadata->pendingSnapshotVersion = f.version;
 					metadata->bytesInNewDeltaFiles = 0;
+					metadata->newDeltaFileCount = 0;
 				}
 			} else {
 				if (f.version > rollbackVersion) {
@@ -1879,6 +1881,7 @@ Version doGranuleRollback(Reference<GranuleMetadata> metadata,
 					ASSERT(f.version > cfRollbackVersion);
 					cfRollbackVersion = f.version;
 					metadata->bytesInNewDeltaFiles += f.bytes;
+					metadata->newDeltaFileCount++;
 				}
 			}
 		}
@@ -1911,6 +1914,7 @@ Version doGranuleRollback(Reference<GranuleMetadata> metadata,
 		     i >= 0 && metadata->files.deltaFiles[i].version > metadata->pendingSnapshotVersion;
 		     i--) {
 			metadata->bytesInNewDeltaFiles += metadata->files.deltaFiles[i].logicalSize;
+			metadata->newDeltaFileCount++;
 		}
 
 		// Track that this rollback happened, since we have to re-read mutations up to the rollback
@@ -2196,6 +2200,7 @@ ACTOR Future<Void> blobGranuleUpdateFiles(Reference<BlobWorkerData> bwData,
 				for (int i = files.deltaFiles.size() - 1; i >= 0; i--) {
 					if (files.deltaFiles[i].version > snapshotVersion) {
 						metadata->bytesInNewDeltaFiles += files.deltaFiles[i].logicalSize;
+						metadata->newDeltaFileCount++;
 					}
 				}
 			}
@@ -2803,6 +2808,7 @@ ACTOR Future<Void> blobGranuleUpdateFiles(Reference<BlobWorkerData> bwData,
 				ASSERT(metadata->bufferedDeltaVersion <= lastDeltaVersion);
 				metadata->bufferedDeltaVersion = lastDeltaVersion; // In case flush was forced at non-mutation version
 				metadata->bytesInNewDeltaFiles += metadata->bufferedDeltaBytes;
+				metadata->newDeltaFileCount++;
 
 				bwData->stats.mutationBytesBuffered -= metadata->bufferedDeltaBytes;
 
@@ -2838,11 +2844,11 @@ ACTOR Future<Void> blobGranuleUpdateFiles(Reference<BlobWorkerData> bwData,
 			// making a bunch of extra delta files at some point, even if we don't consider it for a split
 			// yet
 
-			// If we have enough delta files, try to re-snapshot
-			// FIXME: have max file count in addition to bytes count
+			// If we have enough delta file data, try to re-snapshot
 			if (snapshotEligible &&
 			    (metadata->doEarlyReSnapshot() ||
-			     metadata->bytesInNewDeltaFiles >= metadata->writeAmpTarget.getBytesBeforeCompact())) {
+			     metadata->bytesInNewDeltaFiles >= metadata->writeAmpTarget.getBytesBeforeCompact() ||
+			     metadata->newDeltaFileCount >= 20)) {
 				if (BW_DEBUG && !inFlightFiles.empty()) {
 					fmt::print("Granule [{0} - {1}) ready to re-snapshot at {2} after {3} > {4} bytes, "
 					           "waiting for outstanding {5} files to finish\n",
@@ -2854,6 +2860,9 @@ ACTOR Future<Void> blobGranuleUpdateFiles(Reference<BlobWorkerData> bwData,
 					           inFlightFiles.size());
 				}
 
+				CODE_PROBE(metadata->doEarlyReSnapshot(), "granule snapshotting early");
+				CODE_PROBE(metadata->newDeltaFileCount >= 20, "granule snapshotting due to many small delta files");
+
 				// cancel previous candidate checker
 				checkMergeCandidate.cancel();
 
@@ -2892,6 +2901,7 @@ ACTOR Future<Void> blobGranuleUpdateFiles(Reference<BlobWorkerData> bwData,
 
 				// reset metadata
 				metadata->bytesInNewDeltaFiles = 0;
+				metadata->newDeltaFileCount = 0;
 				metadata->resetReadStats();
 
 				// If we have more than one snapshot file and that file is unblocked (committedVersion >=

fdbserver/include/fdbserver/BlobWorker.h

@@ -88,6 +88,7 @@ struct GranuleMetadata : NonCopyable, ReferenceCounted<GranuleMetadata> {
 
 	uint64_t bytesInNewDeltaFiles = 0;
 	uint64_t bufferedDeltaBytes = 0;
+	uint16_t newDeltaFileCount = 0;
 
 	// for client to know when it is safe to read a certain version and from where (check waitForVersion)
 	Version bufferedDeltaVersion; // largest delta version in currentDeltas (including empty versions)