From 507101e453c074a7f46fda6ceb0fb3e62739b188 Mon Sep 17 00:00:00 2001
From: Andrea Cavalli <andrea@cavallium.it>
Date: Sun, 27 Jun 2021 15:06:48 +0200
Subject: [PATCH] Add verifyChecksum method

---
 .../dbengine/client/CompositeDatabase.java    |  2 ++
 .../dbengine/database/LLKeyValueDatabase.java |  2 ++
 .../database/disk/LLLocalDictionary.java      |  2 +-
 .../disk/LLLocalKeyValueDatabase.java         | 20 ++++++++++++++++---
 4 files changed, 22 insertions(+), 4 deletions(-)

diff --git a/src/main/java/it/cavallium/dbengine/client/CompositeDatabase.java b/src/main/java/it/cavallium/dbengine/client/CompositeDatabase.java
index 9368217..3e97b1c 100644
--- a/src/main/java/it/cavallium/dbengine/client/CompositeDatabase.java
+++ b/src/main/java/it/cavallium/dbengine/client/CompositeDatabase.java
@@ -24,4 +24,6 @@ public interface CompositeDatabase {
 	 * Find corrupted items
 	 */
 	Flux<BadBlock> badBlocks();
+
+	Mono<Void> verifyChecksum();
 }
diff --git a/src/main/java/it/cavallium/dbengine/database/LLKeyValueDatabase.java b/src/main/java/it/cavallium/dbengine/database/LLKeyValueDatabase.java
index 20ad4e7..568f5c6 100644
--- a/src/main/java/it/cavallium/dbengine/database/LLKeyValueDatabase.java
+++ b/src/main/java/it/cavallium/dbengine/database/LLKeyValueDatabase.java
@@ -44,6 +44,8 @@ public interface LLKeyValueDatabase extends LLSnapshottable, LLKeyValueDatabaseS
 
 	Mono<Long> getProperty(String propertyName);
 
+	Mono<Void> verifyChecksum();
+
 	ByteBufAllocator getAllocator();
 
 	Mono<Void> close();
diff --git a/src/main/java/it/cavallium/dbengine/database/disk/LLLocalDictionary.java b/src/main/java/it/cavallium/dbengine/database/disk/LLLocalDictionary.java
index f57e8a1..543d382 100644
--- a/src/main/java/it/cavallium/dbengine/database/disk/LLLocalDictionary.java
+++ b/src/main/java/it/cavallium/dbengine/database/disk/LLLocalDictionary.java
@@ -1263,7 +1263,7 @@ public class LLLocalDictionary implements LLDictionary {
 							try (var rocksIterator = rocksIteratorTuple.getT1()) {
 								rocksIterator.seekToFirst();
 								rocksIterator.status();
-								while (rocksIterator.isValid()) {
+								while (rocksIterator.isValid() && !sink.isCancelled()) {
 									try {
 										rocksIterator.status();
 										rocksIterator.key(DUMMY_WRITE_ONLY_BYTE_BUFFER);
diff --git a/src/main/java/it/cavallium/dbengine/database/disk/LLLocalKeyValueDatabase.java b/src/main/java/it/cavallium/dbengine/database/disk/LLLocalKeyValueDatabase.java
index 5102653..f8d412d 100644
--- a/src/main/java/it/cavallium/dbengine/database/disk/LLLocalKeyValueDatabase.java
+++ b/src/main/java/it/cavallium/dbengine/database/disk/LLLocalKeyValueDatabase.java
@@ -132,7 +132,9 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
 							logger.warn(ex.getLocalizedMessage());
 							options
 									.setUseDirectReads(false)
-									.setUseDirectIoForFlushAndCompaction(false);
+									.setUseDirectIoForFlushAndCompaction(false)
+									.setAllowMmapReads(true)
+									.setAllowMmapWrites(true);
 						}
 						default -> throw ex;
 					}
@@ -270,8 +272,6 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
 			// LOW MEMORY
 			options
 					.setLevelCompactionDynamicLevelBytes(false)
-					.setAllowMmapReads(false)
-					.setAllowMmapWrites(false)
 					.setBytesPerSync(0) // default
 					.setWalBytesPerSync(0) // default
 					.setIncreaseParallelism(1)
@@ -317,6 +317,8 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
 
 			if (USE_DIRECT_IO) {
 				options
+						.setAllowMmapReads(false)
+						.setAllowMmapWrites(false)
 						.setUseDirectIoForFlushAndCompaction(true)
 						.setUseDirectReads(true)
 						// Option to enable readahead in compaction
@@ -486,6 +488,18 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
 				.subscribeOn(dbScheduler);
 	}
 
+	@Override
+	public Mono<Void> verifyChecksum() {
+		return Mono
+				.<Void>fromCallable(() -> {
+					db.verifyChecksum();
+					return null;
+				})
+				.onErrorMap(cause -> new IOException("Failed to verify checksum of database \""
+						+ getDatabaseName() + "\"", cause))
+				.subscribeOn(dbScheduler);
+	}
+
 	@Override
 	public ByteBufAllocator getAllocator() {
 		return allocator;