From 7f3cca96d21addfb448e18b8c85cc9d12699c44f Mon Sep 17 00:00:00 2001 From: leventov Date: Sun, 26 Feb 2017 21:02:34 -0600 Subject: [PATCH] Add ChronicleHashCorruption listening interface, and Beta annotation --- pom.xml | 1 + .../java/net/openhft/chronicle/hash/Beta.java | 42 ++++++ .../chronicle/hash/ChronicleHashBuilder.java | 112 ++++++++++++++- .../hash/ChronicleHashCorruption.java | 86 ++++++++++++ .../hash/impl/VanillaChronicleHash.java | 62 ++++++--- .../impl/stage/iter/SegmentsRecovery.java | 92 ++++++++----- .../hash/impl/stage/iter/TierRecovery.java | 129 ++++++++++++------ .../map/ChronicleHashCorruptionImpl.java | 77 +++++++++++ .../chronicle/map/ChronicleMapBuilder.java | 61 +++++++-- .../chronicle/map/VanillaChronicleMap.java | 9 +- .../chronicle/map/impl/IterationContext.java | 6 +- .../stage/iter/ReplicatedTierRecovery.java | 26 ++-- .../chronicle/set/ChronicleSetBuilder.java | 52 +++++-- .../openhft/chronicle/map/RecoverTest.java | 53 ++++++- 14 files changed, 669 insertions(+), 139 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/hash/Beta.java create mode 100644 src/main/java/net/openhft/chronicle/hash/ChronicleHashCorruption.java create mode 100644 src/main/java/net/openhft/chronicle/map/ChronicleHashCorruptionImpl.java diff --git a/pom.xml b/pom.xml index 903ee79d1..0cc9359b2 100755 --- a/pom.xml +++ b/pom.xml @@ -279,6 +279,7 @@ net/openhft/chronicle/map/Replica.java net/openhft/chronicle/map/ReplicatedChronicleMap.java net/openhft/chronicle/map/ReplicatedGlobalMutableState*Native.java + net/openhft/chronicle/map/ChronicleHashCorruptionImpl.java net/openhft/chronicle/set/DummyValueData.java **/Compiled*.java diff --git a/src/main/java/net/openhft/chronicle/hash/Beta.java b/src/main/java/net/openhft/chronicle/hash/Beta.java new file mode 100644 index 000000000..f0bb11d81 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/Beta.java @@ -0,0 +1,42 @@ +/* + * Copyright (C) 2016 Roman Leventov + * Copyright (C) 2010 The Guava Authors + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.hash; + +import java.lang.annotation.Documented; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +import static java.lang.annotation.ElementType.*; + +/** + * Signifies that a public API (public class, method or field) is subject to incompatible changes, + * or even removal, in a future release. An API bearing this annotation is exempt from any + * compatibility guarantees. Note that the presence of this annotation implies nothing about the + * quality or performance of the API in question, only the fact that it is not "API-frozen." + * + *

This annotation is a copy of Guava's + * {@code Beta}. + */ +@Retention(value = RetentionPolicy.CLASS) +@Target(value = {ANNOTATION_TYPE, CONSTRUCTOR, FIELD,METHOD, TYPE}) +@Documented +@Beta +public @interface Beta { +} diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java index b8c46098c..13ddf9980 100755 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java @@ -528,6 +528,7 @@ public interface ChronicleHashBuilder, * @param file the file with existing hash container or a desired location of a new off-heap * persisted hash container * @return a hash container mapped to the given file + * @throws NullPointerException if the given file is null * @throws IOException if any IO error, related to off-heap memory allocation or file mapping, * or establishing replication connections, occurs * @see ChronicleHash#file() @@ -552,6 +553,7 @@ public interface ChronicleHashBuilder, * * @param file the persistence file for existing of future hash container * @return a {@code ChronicleHash} instance, mapped to the given instance + * @throws NullPointerException if the given file is null * @throws IOException if any IO error occurs on reading data from the file, or related to * off-heap memory allocation or file mapping, or establishing replication connections. Probably * the file is corrupted on OS level, and should be recovered on that level first, before @@ -560,6 +562,7 @@ public interface ChronicleHashBuilder, * @see #createPersistedTo(File) * @see #createOrRecoverPersistedTo(File, boolean) * @see #recoverPersistedTo(File, boolean) + * @see #createOrRecoverPersistedTo(File, boolean, ChronicleHashCorruption.Listener) */ H createOrRecoverPersistedTo(File file) throws IOException; @@ -570,6 +573,10 @@ public interface ChronicleHashBuilder, * file doesn't exist, and {@link #recoverPersistedTo(File, boolean) * recoverPersistedTo(file, sameLibraryVersion)}, if the file exists. * + *

The difference between this method and {@link #createOrRecoverPersistedTo(File, boolean, + * ChronicleHashCorruption.Listener)} is that this method just logs the encountered corruptions, + * instead of passing them to the specified corruption listener. + * * @param file the persistence file for existing of future hash container * @param sameLibraryVersion if this builder is configured with the same configurations, as the * builder, which created the Chronicle Map in the given file initially, and the same version @@ -577,6 +584,7 @@ public interface ChronicleHashBuilder, * (with presumably the same configurations), protecting from {@link * ChronicleHashRecoveryFailedException}, if the header is corrupted. * @return a {@code ChronicleHash} instance, mapped to the given instance + * @throws NullPointerException if the given file is null * @throws IOException if any IO error occurs on reading data from the file, or related to * off-heap memory allocation or file mapping, or establishing replication connections. Probably * the file is corrupted on OS level, and should be recovered on that level first, before @@ -584,9 +592,42 @@ public interface ChronicleHashBuilder, * @throws ChronicleHashRecoveryFailedException if recovery is impossible * @see #createPersistedTo(File) * @see #recoverPersistedTo(File, boolean) + * @see #createOrRecoverPersistedTo(File, boolean, ChronicleHashCorruption.Listener) */ H createOrRecoverPersistedTo(File file, boolean sameLibraryVersion) throws IOException; + /** + * Recovers and opens the hash container, persisted to the specified file, or creates a new one + * from this builder, if the file doesn't exist yet, and maps its off-heap memory to the file. + * In other words, this methods behaves like {@link #createPersistedTo(File)}, if the given + * file doesn't exist, and {@link + * #recoverPersistedTo(File, boolean, ChronicleHashCorruption.Listener) + * recoverPersistedTo(file, sameLibraryVersion, corruptionListener)}, if the file exists. + * + *

If this procedure encounters corruptions, it fixes them (recovers from them) and + * notifies the provided corruption listener with the details about the corruption. See the + * documentation for {@link ChronicleHashCorruption} for more information. + * + * @param file the persistence file for existing of future hash container + * @param sameLibraryVersion if this builder is configured with the same configurations, as the + * builder, which created the Chronicle Map in the given file initially, and the same version + * of the Chronicle Map library was used. In this case, the header of the file is overridden + * (with presumably the same configurations), protecting from {@link + * ChronicleHashRecoveryFailedException}, if the header is corrupted. + * @return a {@code ChronicleHash} instance, mapped to the given instance + * @throws NullPointerException if the given file or corruptionListener is null + * @throws IOException if any IO error occurs on reading data from the file, or related to + * off-heap memory allocation or file mapping, or establishing replication connections. Probably + * the file is corrupted on OS level, and should be recovered on that level first, before + * calling this procedure. + * @throws ChronicleHashRecoveryFailedException if recovery is impossible + * @see #createOrRecoverPersistedTo(File, boolean) + * @see #recoverPersistedTo(File, boolean) + */ + H createOrRecoverPersistedTo( + File file, boolean sameLibraryVersion, + ChronicleHashCorruption.Listener corruptionListener) throws IOException; + /** * Recovers and opens the hash container, persisted to the specified file. This method * should be used to open a persisted Chronicle Map after an accessor process crash (that might @@ -597,11 +638,11 @@ public interface ChronicleHashBuilder, * which were never put into the Chronicle Hash). * *

This method, unlike {@link #createPersistedTo(File)} or {@link - * #createOrRecoverPersistedTo(File, boolean)} methods, expects that the mapping file already + * #createOrRecoverPersistedTo(File, boolean)} methods, expects that the given file already * exists. * - * "Recovery" of the hash container is changing the memory of the data structure so that after - * the recovery the hash container is in some correct state: with "clean" locks, coherent + *

"Recovery" of the hash container is changing the memory of the data structure so that + * after the recovery the hash container is in some correct state: with "clean" locks, coherent * entry counters, not containing provably corrupt entries, etc. If {@link * #checksumEntries(boolean) checksumEntries(true)} is configured for the chronicle hash * container, recovery procedure checks for each entry that the checksums is correct, otherwise @@ -609,8 +650,15 @@ public interface ChronicleHashBuilder, * Recovery section in the * Chronicle Map tutorial for more information. * + *

The difference between this method and {@link #recoverPersistedTo(File, boolean, + * ChronicleHashCorruption.Listener)} is that this method just logs the encountered corruptions, + * instead of passing them to the specified corruption listener. + * *

At the moment this method is called and executed, no other thread or process should be - * mapping to the given file, and trying to access the given file. + * mapping to the given file, and trying to access the given file. Otherwise the outcomes of + * the {@code recoverPersistedTo()} call, as well as the behaviour of the concurrent thread or + * process, accessing the same Chronicle Map, are unspecified: exception or error could be + * thrown, the Chronicle Map persisted to the given file could be further corrupted. * *

It is strongly recommended to configure this builder with the same configurations, as * the builder, that created the given file for the first time, and pass {@code true} as the @@ -628,6 +676,7 @@ public interface ChronicleHashBuilder, * In this case, the header of the file is overridden (with presumably the same configurations), * protecting from {@link ChronicleHashRecoveryFailedException}, if the header is corrupted. * @return a recovered Chronicle Hash instance, mapped to the given file + * @throws NullPointerException if the given file is null * @throws FileNotFoundException if the file doesn't exist * @throws IOException if any IO error occurs on reading data from the file, or related to * off-heap memory allocation or file mapping, or establishing replication connections. Probably @@ -636,9 +685,64 @@ public interface ChronicleHashBuilder, * @throws ChronicleHashRecoveryFailedException if recovery is impossible * @see #createOrRecoverPersistedTo(File, boolean) * @see #createPersistedTo(File) + * @see #recoverPersistedTo(File, boolean, ChronicleHashCorruption.Listener) */ H recoverPersistedTo(File file, boolean sameBuilderConfigAndLibraryVersion) throws IOException; + /** + * Recovers and opens the hash container, persisted to the specified file. This method + * should be used to open a persisted Chronicle Map after an accessor process crash (that might + * leave some locks "acquired" therefore some segments inaccessible), or an accessor process + * external termination (that, in addition to inaccessible segments, might lead to leaks in the + * Chronicle Hash memory), or a sudden power loss, or a file corruption (that, in addition to + * the already mentioned consequences, might lead to data corruption, i. e. presence of entries + * which were never put into the Chronicle Hash). + * + *

This method, unlike {@link #createPersistedTo(File)} or {@link + * #createOrRecoverPersistedTo(File, boolean, ChronicleHashCorruption.Listener)} methods, + * expects that the given file already exists. + * + *

"Recovery" of the hash container is changing the memory of the data structure so that + * after the recovery the hash container is in some correct state: with "clean" locks, coherent + * entry counters, not containing provably corrupt entries, etc. If {@link + * #checksumEntries(boolean) checksumEntries(true)} is configured for the chronicle hash + * container, recovery procedure checks for each entry that the checksums is correct, otherwise + * it assumes the entry is corrupt and deletes it from the Chronicle Hash. See the + * Recovery section in the + * Chronicle Map tutorial for more information. + * + *

If this procedure encounters corruptions, it fixes them (recovers from them) and + * notifies the provided corruption listener with the details about the corruption. See the + * documentation for {@link ChronicleHashCorruption} for more information. + * + *

At the moment this method is called and executed, no other thread or process should be + * mapping to the given file, and trying to access the given file. Otherwise the outcomes of + * the {@code recoverPersistedTo()} call, as well as the behaviour of the concurrent thread or + * process, accessing the same Chronicle Map, are unspecified: exception or error could be + * thrown, the Chronicle Map persisted to the given file could be further corrupted. + * + * @param file a hash container was mapped to the given file + * @param sameBuilderConfigAndLibraryVersion if this builder is configured with the same + * configurations, as the builder, which created the file (the persisted Chronicle Hash + * instance) for the first time, and with the same version of the Chronicle Map library. + * In this case, the header of the file is overridden (with presumably the same configurations), + * protecting from {@link ChronicleHashRecoveryFailedException}, if the header is corrupted. + * @return a recovered Chronicle Hash instance, mapped to the given file + * @throws NullPointerException if the given file or corruptionListener is null + * @throws FileNotFoundException if the file doesn't exist + * @throws IOException if any IO error occurs on reading data from the file, or related to + * off-heap memory allocation or file mapping, or establishing replication connections. Probably + * the file is corrupted on OS level, and should be recovered on that level first, before + * calling this procedure. + * @throws ChronicleHashRecoveryFailedException if recovery is impossible + * @see #recoverPersistedTo(File, boolean) + * @see #createOrRecoverPersistedTo(File, boolean, ChronicleHashCorruption.Listener) + * @see #createPersistedTo(File) + */ + H recoverPersistedTo( + File file, boolean sameBuilderConfigAndLibraryVersion, + ChronicleHashCorruption.Listener corruptionListener) throws IOException; + /** * @deprecated don't use private API in the client code */ diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashCorruption.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashCorruption.java new file mode 100644 index 000000000..524783e83 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashCorruption.java @@ -0,0 +1,86 @@ +/* + * Copyright (C) 2016 Roman Leventov + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.hash; + +import org.jetbrains.annotations.Nullable; + +import java.io.File; + +/** + * Information about a corruption, encountered in a persisted Chronicle Map during recovery. + * + *

Recovery procedure doesn't guarantee accuracy of the corruption events. Only two things are + * guaranteed: + *

    + *
  1. if {@link Listener} didn't receive any corruption events, the recovered Chronicle Map + * was not corrupted; + *
    2. + *
  2. if {@link Listener} received some corruption events, the recovered Chronicle Map was + * corrupted.
    3. + *
+ * + *

{@code ChronicleHashCorruption} objects, passed to {@link Listener}, shouldn't be saved and + * used outside of the {@link Listener#onCorruption(ChronicleHashCorruption)} method body, because + * {@code ChronicleHashCorruption} objects could be reused during the recovery procedure. + * + * + * @see ChronicleHashBuilder#recoverPersistedTo(File, boolean, Listener) + * @see ChronicleHashBuilder#createOrRecoverPersistedTo(File, boolean, Listener) + */ +@Beta +public interface ChronicleHashCorruption { + + /** + * Returns the message, explaining this corruption. + */ + String message(); + + /** + * Returns the exception, associated with this corruption, if any, or {@code null} if there is + * no exception. + */ + @Nullable + Throwable exception(); + + /** + * Returns the index of the segment, with which this corruption is associated, or -1, if not + * applicable. E. g. if this is a segment lock word corruption, returns the index of the + * segment, guarded by the corrupted lock. If this is an entry data corruption, returns the + * index of the segment, in which the corrupted entry is stored. If the corruption is not + * associated with a particular segment, returns -1, e. g. if this is a ChronicleHash + * header corruption. + */ + int segmentIndex(); + + /** + * Listener of {@link ChronicleHashCorruption} events. + */ + @Beta + interface Listener { + /** + * Called, when recovery + * procedure encounters a corruption of a persisted Chronicle Map. + * + * @param corruption the corruption object, must not be saved and used outside the body of + * the {@code #onCorruption()} method, because during the recovery + * procedure, the same corruption object could be reused. + * @see ChronicleHashCorruption + */ + void onCorruption(ChronicleHashCorruption corruption); + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java index 482b577ad..8f50db9cf 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -31,14 +31,13 @@ import net.openhft.chronicle.hash.serialization.DataAccess; import net.openhft.chronicle.hash.serialization.SizeMarshaller; import net.openhft.chronicle.hash.serialization.SizedReader; import net.openhft.chronicle.hash.serialization.impl.SerializationBuilder; +import net.openhft.chronicle.map.ChronicleHashCorruptionImpl; import net.openhft.chronicle.map.ChronicleMapBuilder; import net.openhft.chronicle.values.Values; import net.openhft.chronicle.wire.Marshallable; import net.openhft.chronicle.wire.WireIn; import net.openhft.chronicle.wire.WireOut; import org.jetbrains.annotations.NotNull; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import sun.misc.Cleaner; import java.io.Closeable; @@ -60,14 +59,14 @@ import static net.openhft.chronicle.algo.MemoryUnit.*; import static net.openhft.chronicle.algo.bytes.Access.nativeAccess; import static net.openhft.chronicle.core.OS.pageAlign; import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.*; +import static net.openhft.chronicle.map.ChronicleHashCorruptionImpl.format; +import static net.openhft.chronicle.map.ChronicleHashCorruptionImpl.report; public abstract class VanillaChronicleHash, SC extends HashSegmentContext, ECQ extends ExternalHashQueryContext> implements ChronicleHash, Marshallable { - private static final Logger LOG = LoggerFactory.getLogger(VanillaChronicleHash.class); - public static final long TIER_COUNTERS_AREA_SIZE = 64; public static final long RESERVED_GLOBAL_MUTABLE_STATE_BYTES = 1024; @@ -505,7 +504,9 @@ public abstract class VanillaChronicleHash + format("global mutable state lock of map at {} is not clear: {}", + file, lockingStrategy.toString(lockState)) + ); lockingStrategy.reset(nativeAccess(), null, lockAddr); } } - private void recoverAllocatedExtraTierBulks(int allocatedExtraTierBulks) { + private void recoverAllocatedExtraTierBulks( + int allocatedExtraTierBulks, ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { if (globalMutableState.getAllocatedExtraTierBulks() != allocatedExtraTierBulks) { - LOG.error("allocated extra tier bulks counter corrupted, or the map file {} " + - "is truncated. stored: {}, should be: {}", file, - globalMutableState.getAllocatedExtraTierBulks(), allocatedExtraTierBulks); + report(corruptionListener, corruption, -1, () -> + format("allocated extra tier bulks counter corrupted, or the map file {} " + + "is truncated. stored: {}, should be: {}", + file, globalMutableState.getAllocatedExtraTierBulks(), + allocatedExtraTierBulks) + ); globalMutableState.setAllocatedExtraTierBulks(allocatedExtraTierBulks); } } - private void recoverSegmentHeadersOffset(long segmentHeadersOffset) { + private void recoverSegmentHeadersOffset( + long segmentHeadersOffset, ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { if (globalMutableState.getSegmentHeadersOffset() != segmentHeadersOffset) { - LOG.error("segment headers offset of map at {} corrupted. stored: {}, should be: {}", - file, globalMutableState.getSegmentHeadersOffset(), segmentHeadersOffset); + report(corruptionListener, corruption, -1, () -> + format("segment headers offset of map at {} corrupted. stored: {}, should be: {}", + file, globalMutableState.getSegmentHeadersOffset(), segmentHeadersOffset) + ); globalMutableState.setSegmentHeadersOffset(segmentHeadersOffset); } } - private void recoverDataStoreSize(long dataStoreSize) { + private void recoverDataStoreSize( + long dataStoreSize, ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { if (globalMutableState.getDataStoreSize() != dataStoreSize) { - LOG.error("data store size of map at {} corrupted. stored: {}, should be: {}", - file, globalMutableState.getDataStoreSize(), dataStoreSize); + report(corruptionListener, corruption, -1, () -> + format("data store size of map at {} corrupted. stored: {}, should be: {}", + file, globalMutableState.getDataStoreSize(), dataStoreSize) + ); globalMutableState.setDataStoreSize(dataStoreSize); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/SegmentsRecovery.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/SegmentsRecovery.java index 10024cafb..7bccfd2fa 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/SegmentsRecovery.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/SegmentsRecovery.java @@ -17,34 +17,40 @@ package net.openhft.chronicle.hash.impl.stage.iter; +import net.openhft.chronicle.hash.ChronicleHashCorruption; import net.openhft.chronicle.hash.VanillaGlobalMutableState; import net.openhft.chronicle.hash.impl.TierCountersArea; import net.openhft.chronicle.hash.impl.VanillaChronicleHash; import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; -import net.openhft.chronicle.hash.impl.stage.hash.LogHolder; +import net.openhft.chronicle.map.ChronicleHashCorruptionImpl; import net.openhft.chronicle.map.impl.IterationContext; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; -import org.slf4j.Logger; + +import static net.openhft.chronicle.map.ChronicleHashCorruptionImpl.format; +import static net.openhft.chronicle.map.ChronicleHashCorruptionImpl.report; @Staged public abstract class SegmentsRecovery implements IterationContext { - @StageRef VanillaChronicleHashHolder hh; - @StageRef SegmentStages s; - @StageRef TierRecovery tierRecovery; - @StageRef LogHolder lh; + @StageRef + VanillaChronicleHashHolder hh; + @StageRef + SegmentStages s; + @StageRef + TierRecovery tierRecovery; @Override - public void recoverSegments() { - Logger log = lh.LOG; + public void recoverSegments( + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { VanillaChronicleHash h = hh.h(); for (int segmentIndex = 0; segmentIndex < h.actualSegments; segmentIndex++) { s.initSegmentIndex(segmentIndex); - resetSegmentLock(); - zeroOutFirstSegmentTierCountersArea(); - tierRecovery.recoverTier(segmentIndex); + resetSegmentLock(corruptionListener, corruption); + zeroOutFirstSegmentTierCountersArea(corruptionListener, corruption); + tierRecovery.recoverTier(segmentIndex, corruptionListener, corruption); } VanillaGlobalMutableState globalMutableState = h.globalMutableState(); @@ -58,14 +64,17 @@ public abstract class SegmentsRecovery implements IterationContext { long tierIndex = h.extraTierIndexToTierIndex(extraTierIndex); // `tier` is unused in recoverTier(), 0 should be a safe value s.initSegmentTier(0, tierIndex); - int segmentIndex = tierRecovery.recoverTier(-1); + int segmentIndex = tierRecovery.recoverTier( + -1, corruptionListener, corruption); if (segmentIndex >= 0) { long tierCountersAreaAddr = s.tierCountersAreaAddr(); int storedSegmentIndex = TierCountersArea.segmentIndex(tierCountersAreaAddr); if (storedSegmentIndex != segmentIndex) { - log.error("wrong segment index stored in tier counters area " + - "of tier with index {}: {}, should be, based on entries: {}", - tierIndex, storedSegmentIndex, segmentIndex); + report(corruptionListener, corruption, segmentIndex, () -> + format("wrong segment index stored in tier counters area " + + "of tier with index {}: {}, should be, based on entries: {}", + tierIndex, storedSegmentIndex, segmentIndex) + ); TierCountersArea.segmentIndex(tierCountersAreaAddr, segmentIndex); } s.nextTierIndex(0); @@ -84,8 +93,11 @@ public abstract class SegmentsRecovery implements IterationContext { } if (storedExtraTiersInUse != actualExtraTiersInUse) { - log.error("wrong number of actual tiers in use in global mutable state, stored: {}, " + - "should be: {}", storedExtraTiersInUse, actualExtraTiersInUse); + long finalActualExtraTiersInUse = actualExtraTiersInUse; + report(corruptionListener, corruption, -1, () -> + format("wrong number of actual tiers in use in global mutable state, stored: {}, " + + "should be: {}", storedExtraTiersInUse, finalActualExtraTiersInUse) + ); globalMutableState.setExtraTiersInUse(actualExtraTiersInUse); } @@ -105,22 +117,26 @@ public abstract class SegmentsRecovery implements IterationContext { } long storedFirstFreeTierIndex = globalMutableState.getFirstFreeTierIndex(); if (storedFirstFreeTierIndex != firstFreeTierIndex) { - log.error("wrong first free tier index in global mutable state, stored: {}, " + - "should be: {}", storedFirstFreeTierIndex, firstFreeTierIndex); + report(corruptionListener, corruption, -1, () -> + format("wrong first free tier index in global mutable state, stored: {}, " + + "should be: {}", storedFirstFreeTierIndex, firstFreeTierIndex) + ); globalMutableState.setFirstFreeTierIndex(firstFreeTierIndex); } - removeDuplicatesInSegments(); + removeDuplicatesInSegments(corruptionListener, corruption); } - private void removeDuplicatesInSegments() { + private void removeDuplicatesInSegments( + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { VanillaChronicleHash h = hh.h(); for (int segmentIndex = 0; segmentIndex < h.actualSegments; segmentIndex++) { s.initSegmentIndex(segmentIndex); s.initSegmentTier(); s.goToLastTier(); while (true) { - tierRecovery.removeDuplicatesInSegment(); + tierRecovery.removeDuplicatesInSegment(corruptionListener, corruption); if (s.tier > 0) { s.prevTier(); } else { @@ -130,31 +146,43 @@ public abstract class SegmentsRecovery implements IterationContext { } } - private void resetSegmentLock() { + private void resetSegmentLock( + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { long lockState = s.segmentHeader.getLockState(s.segmentHeaderAddress); if (lockState != s.segmentHeader.resetLockState()) { - lh.LOG.error("lock of segment {} is not clear: {}", - s.segmentIndex, s.segmentHeader.lockStateToString(lockState)); + report(corruptionListener, corruption, s.segmentIndex, () -> + format("lock of segment {} is not clear: {}", + s.segmentIndex, s.segmentHeader.lockStateToString(lockState)) + ); s.segmentHeader.resetLock(s.segmentHeaderAddress); } } - private void zeroOutFirstSegmentTierCountersArea() { + private void zeroOutFirstSegmentTierCountersArea( + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { s.nextTierIndex(0); if (s.prevTierIndex() != 0) { - lh.LOG.error("stored prev tier index in first tier of segment {}: {}, should be 0", - s.segmentIndex, s.prevTierIndex()); + report(corruptionListener, corruption, s.segmentIndex, () -> + format("stored prev tier index in first tier of segment {}: {}, should be 0", + s.segmentIndex, s.prevTierIndex()) + ); s.prevTierIndex(0); } long tierCountersAreaAddr = s.tierCountersAreaAddr(); if (TierCountersArea.segmentIndex(tierCountersAreaAddr) != 0) { - lh.LOG.error("stored segment index in first tier of segment {}: {}, should be 0", - s.segmentIndex, TierCountersArea.segmentIndex(tierCountersAreaAddr)); + report(corruptionListener, corruption, s.segmentIndex, () -> + format("stored segment index in first tier of segment {}: {}, should be 0", + s.segmentIndex, TierCountersArea.segmentIndex(tierCountersAreaAddr)) + ); TierCountersArea.segmentIndex(tierCountersAreaAddr, 0); } if (TierCountersArea.tier(tierCountersAreaAddr) != 0) { - lh.LOG.error("stored tier in first tier of segment {}: {}, should be 0", - s.segmentIndex, TierCountersArea.tier(tierCountersAreaAddr)); + report(corruptionListener, corruption, s.segmentIndex, () -> + format("stored tier in first tier of segment {}: {}, should be 0", + s.segmentIndex, TierCountersArea.tier(tierCountersAreaAddr)) + ); TierCountersArea.tier(tierCountersAreaAddr, 0); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/TierRecovery.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/TierRecovery.java index ce8ea31ea..f092638dc 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/TierRecovery.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/TierRecovery.java @@ -18,12 +18,13 @@ package net.openhft.chronicle.hash.impl.stage.iter; import net.openhft.chronicle.bytes.BytesUtil; +import net.openhft.chronicle.hash.ChronicleHashCorruption; import net.openhft.chronicle.hash.ChronicleHashRecoveryFailedException; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable; import net.openhft.chronicle.hash.impl.VanillaChronicleHash; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; -import net.openhft.chronicle.hash.impl.stage.hash.LogHolder; +import net.openhft.chronicle.map.ChronicleHashCorruptionImpl; import net.openhft.chronicle.map.ExternalMapQueryContext; import net.openhft.chronicle.map.MapEntry; import net.openhft.chronicle.map.VanillaChronicleMap; @@ -31,21 +32,22 @@ import net.openhft.chronicle.map.impl.VanillaChronicleMapHolder; import net.openhft.chronicle.map.impl.stage.entry.MapEntryStages; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; -import org.slf4j.Logger; + +import static net.openhft.chronicle.map.ChronicleHashCorruptionImpl.*; @Staged public class TierRecovery { - @StageRef LogHolder lh; @StageRef VanillaChronicleMapHolder mh; @StageRef SegmentStages s; @StageRef MapEntryStages e; @StageRef IterationKeyHashCode khc; - public int recoverTier(int segmentIndex) { + public int recoverTier( + int segmentIndex, ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { s.freeList.clearAll(); - Logger log = lh.LOG; VanillaChronicleHash h = mh.h(); CompactOffHeapLinearHashTable hl = h.hashLookup; long hlAddr = s.tierBaseAddr; @@ -59,14 +61,17 @@ public class TierRecovery { // (*) hl.clearEntry(hlAddr, hlPos); if (validEntries >= h.maxEntriesPerHashLookup) { - log.error("Too many entries in tier with index {}, max is {}", - s.tierIndex, h.maxEntriesPerHashLookup); + report(corruptionListener, corruption, segmentIndex, () -> + format("Too many entries in tier with index {}, max is {}", + s.tierIndex, h.maxEntriesPerHashLookup) + ); break nextHlPos; } long searchKey = hl.key(hlEntry); long entryPos = hl.value(hlEntry); - int si = checkEntry(searchKey, entryPos, segmentIndex); + int si = checkEntry(searchKey, entryPos, segmentIndex, + corruptionListener, corruption); if (si < 0) { break nextHlPos; } else { @@ -106,15 +111,18 @@ public class TierRecovery { if (insertPos >= 0 && insertPos < hlPos) { // insertPos already checked e.readExistingEntry(anotherEntryPos); - } else if (checkEntry(searchKey, anotherEntryPos, segmentIndex) < 0) { + } else if (checkEntry(searchKey, anotherEntryPos, segmentIndex, + corruptionListener, corruption) < 0) { break checkDuplicateKeys; } if (e.keySize == currentKeySize && BytesUtil.bytesEqual(s.segmentBS, currentKeyOffset, s.segmentBS, e.keyOffset, currentKeySize)) { - log.error("Entries with duplicate keys within a tier: " + + report(corruptionListener, corruption, segmentIndex, () -> + format("Entries with duplicate keys within a tier: " + "at pos {} and {} with key {}, first value is {}", - entryPos, anotherEntryPos, e.key(), e.value()); + entryPos, anotherEntryPos, e.key(), e.value()) + ); s.freeList.clearRange( entryPos, entryPos + currentEntrySizeInChunks); break nextHlPos; @@ -163,7 +171,9 @@ public class TierRecovery { } while (hlPos != 0 || steps == 0); } - public void removeDuplicatesInSegment() { + public void removeDuplicatesInSegment( + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { long startHlPos = 0L; VanillaChronicleMap m = mh.m(); CompactOffHeapLinearHashTable hashLookup = m.hashLookup; @@ -188,11 +198,13 @@ public class TierRecovery { long keyAddress = key.bytes().address(key.offset()); long key2Address = key2.bytes().address(key2.offset()); if (key2Address != keyAddress) { - lh.LOG.error("entries with duplicate key {} in segment {}: " + - "with values {} and {}, removing the latter", + report(corruptionListener, corruption, s.segmentIndex, () -> + format("entries with duplicate key {} in segment {}: " + + "with values {} and {}, removing the latter", key, c.segmentIndex(), entry2 != null ? ((MapEntry) c).value() : "", - !e.entryDeleted() ? e.value() : ""); + !e.entryDeleted() ? e.value() : "") + ); if (hashLookup.remove(currentTierBaseAddr, hlPos) != hlPos) { hlPos = hashLookup.stepBack(hlPos); steps--; @@ -208,64 +220,84 @@ public class TierRecovery { // entry } while (hlPos != startHlPos || steps == 0); - recoverTierEntriesCounter(entries); - recoverLowestPossibleFreeChunkTiered(); + recoverTierEntriesCounter(entries, corruptionListener, corruption); + recoverLowestPossibleFreeChunkTiered(corruptionListener, corruption); } - private void recoverTierEntriesCounter(long entries) { + private void recoverTierEntriesCounter( + long entries, ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { if (s.tierEntries() != entries) { - lh.LOG.error("Wrong number of entries counter for tier with index {}, " + - "stored: {}, should be: {}", s.tierIndex, s.tierEntries(), entries); + report(corruptionListener, corruption, s.segmentIndex, () -> + format("Wrong number of entries counter for tier with index {}, " + + "stored: {}, should be: {}", s.tierIndex, s.tierEntries(), entries) + ); s.tierEntries(entries); } } - private void recoverLowestPossibleFreeChunkTiered() { + private void recoverLowestPossibleFreeChunkTiered( + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { long lowestFreeChunk = s.freeList.nextClearBit(0); if (lowestFreeChunk == -1) lowestFreeChunk = mh.m().actualChunksPerSegmentTier; if (s.lowestPossiblyFreeChunk() != lowestFreeChunk) { - lh.LOG.error("wrong lowest free chunk for tier with index {}, " + - "stored: {}, should be: {}", - s.tierIndex, s.lowestPossiblyFreeChunk(), lowestFreeChunk); + long finalLowestFreeChunk = lowestFreeChunk; + report(corruptionListener, corruption, s.segmentIndex, () -> + format("wrong lowest free chunk for tier with index {}, " + + "stored: {}, should be: {}", + s.tierIndex, s.lowestPossiblyFreeChunk(), finalLowestFreeChunk) + ); s.lowestPossiblyFreeChunk(lowestFreeChunk); } } - private int checkEntry(long searchKey, long entryPos, int segmentIndex) { - Logger log = lh.LOG; + private int checkEntry( + long searchKey, long entryPos, int segmentIndex, + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { VanillaChronicleHash h = mh.h(); if (entryPos < 0 || entryPos >= h.actualChunksPerSegmentTier) { - log.error("Entry pos is out of range: {}, should be 0-{}", - entryPos, h.actualChunksPerSegmentTier - 1); + report(corruptionListener, corruption, segmentIndex, () -> + format("Entry pos is out of range: {}, should be 0-{}", + entryPos, h.actualChunksPerSegmentTier - 1) + ); return -1; } try { e.readExistingEntry(entryPos); } catch (Exception e) { - log.error("Exception while reading entry key size:", e); + reportException(corruptionListener, corruption, segmentIndex, + () -> "Exception while reading entry key size", e); return -1; } if (e.keyEnd() > s.segmentBytes.capacity()) { - log.error("Wrong key size: {}", e.keySize); + report(corruptionListener, corruption, segmentIndex, () -> + format("Wrong key size: {}", e.keySize) + ); return -1; } long keyHashCode = khc.keyHashCode(); int segmentIndexFromKey = h.hashSplitting.segmentIndex(keyHashCode); if (segmentIndexFromKey < 0 || segmentIndexFromKey >= h.actualSegments) { - log.error("Segment index from the entry key hash code is out of range: {}, " + - "should be 0-{}, entry key: {}", - segmentIndexFromKey, h.actualSegments - 1, e.key()); + report(corruptionListener, corruption, segmentIndex, () -> + format("Segment index from the entry key hash code is out of range: {}, " + + "should be 0-{}, entry key: {}", + segmentIndexFromKey, h.actualSegments - 1, e.key()) + ); return -1; } long segmentHashFromKey = h.hashSplitting.segmentHash(keyHashCode); long searchKeyFromKey = h.hashLookup.maskUnsetKey(segmentHashFromKey); if (searchKey != searchKeyFromKey) { - log.error("HashLookup searchKey: {}, HashLookup searchKey " + + report(corruptionListener, corruption, segmentIndex, () -> + format("HashLookup searchKey: {}, HashLookup searchKey " + "from the entry key hash code: {}, entry key: {}, entry pos: {}", - searchKey, searchKeyFromKey, e.key(), entryPos); + searchKey, searchKeyFromKey, e.key(), entryPos) + ); return -1; } @@ -273,26 +305,33 @@ public class TierRecovery { // e.entryEnd() implicitly reads the value size, to be computed long entryAndChecksumEnd = e.entryEnd() + e.checksumStrategy.extraEntryBytes(); if (entryAndChecksumEnd > s.segmentBytes.capacity()) { - log.error("Wrong value size: {}, key: {}", e.valueSize, e.key()); + report(corruptionListener, corruption, segmentIndex, () -> + format("Wrong value size: {}, key: {}", e.valueSize, e.key()) + ); return -1; } } catch (Exception ex) { - log.error("Exception while reading entry value size, key: " + e.key(), ex); + reportException(corruptionListener, corruption, segmentIndex, () -> + "Exception while reading entry value size, key: " + e.key(), ex); return -1; } int storedChecksum = e.checksumStrategy.storedChecksum(); int checksumFromEntry = e.checksumStrategy.computeChecksum(); if (storedChecksum != checksumFromEntry) { - log.error("Checksum doesn't match, stored: {}, should be from " + + report(corruptionListener, corruption, segmentIndex, () -> + format("Checksum doesn't match, stored: {}, should be from " + "the entry bytes: {}, key: {}, value: {}", - storedChecksum, checksumFromEntry, e.key(), e.value()); + storedChecksum, checksumFromEntry, e.key(), e.value()) + ); return -1; } if (!s.freeList.isRangeClear(entryPos, entryPos + e.entrySizeInChunks)) { - log.error("Overlapping entry: positions {}-{}, key: {}, value: {}", - entryPos, entryPos + e.entrySizeInChunks - 1, e.key(), e.value()); + report(corruptionListener, corruption, segmentIndex, () -> + format("Overlapping entry: positions {}-{}, key: {}, value: {}", + entryPos, entryPos + e.entrySizeInChunks - 1, e.key(), e.value()) + ); return -1; } @@ -300,9 +339,11 @@ public class TierRecovery { return segmentIndexFromKey; } else { if (segmentIndex != segmentIndexFromKey) { - log.error("Expected segment index: {}, segment index from the entry key: {}, " + - "key: {}, value: {}", segmentIndex, searchKeyFromKey, e.key(), - e.value()); + report(corruptionListener, corruption, segmentIndex, () -> + format("Expected segment index: {}, segment index from the entry key: {}, " + + "key: {}, value: {}", + segmentIndex, searchKeyFromKey, e.key(), e.value()) + ); return -1; } else { return segmentIndex; diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleHashCorruptionImpl.java b/src/main/java/net/openhft/chronicle/map/ChronicleHashCorruptionImpl.java new file mode 100644 index 000000000..ee00c3385 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/ChronicleHashCorruptionImpl.java @@ -0,0 +1,77 @@ +/* + * Copyright (C) 2016 Roman Leventov + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program. If not, see . + */ + +package net.openhft.chronicle.map; + +import net.openhft.chronicle.hash.ChronicleHashCorruption; +import org.jetbrains.annotations.Nullable; +import org.slf4j.helpers.MessageFormatter; + +import java.util.function.Supplier; + +public class ChronicleHashCorruptionImpl implements ChronicleHashCorruption { + + public static void report( + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption, int segmentIndex, + Supplier messageSupplier) { + corruption.set(segmentIndex, messageSupplier, null); + corruptionListener.onCorruption(corruption); + } + + public static void reportException( + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption, int segmentIndex, + Supplier messageSupplier, Throwable exception) { + corruption.set(segmentIndex, messageSupplier, exception); + corruptionListener.onCorruption(corruption); + } + + public static String format(String message, Object... args) { + return MessageFormatter.arrayFormat(message, args).getMessage(); + } + + private int segmentIndex; + private Supplier messageSupplier; + private Throwable exception; + private String message; + + private void set(int segmentIndex, Supplier messageSupplier, Throwable exception) { + this.segmentIndex = segmentIndex; + this.messageSupplier = messageSupplier; + this.exception = exception; + this.message = null; + } + + @Override + public String message() { + if (message == null) { + message = messageSupplier.get(); + } + return message; + } + + @Nullable + @Override + public Throwable exception() { + return exception; + } + + @Override + public int segmentIndex() { + return segmentIndex; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 208eb04ff..77cf0a1d7 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -24,6 +24,7 @@ import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.core.Jvm; import net.openhft.chronicle.core.OS; import net.openhft.chronicle.hash.ChronicleHashBuilder; +import net.openhft.chronicle.hash.ChronicleHashCorruption; import net.openhft.chronicle.hash.ChronicleHashRecoveryFailedException; import net.openhft.chronicle.hash.impl.*; import net.openhft.chronicle.hash.impl.stage.entry.ChecksumStrategy; @@ -65,6 +66,8 @@ import static net.openhft.chronicle.hash.impl.VanillaChronicleHash.throwRecovery import static net.openhft.chronicle.hash.impl.util.FileIOUtils.readFully; import static net.openhft.chronicle.hash.impl.util.FileIOUtils.writeFully; import static net.openhft.chronicle.hash.impl.util.Objects.builderEquals; +import static net.openhft.chronicle.map.ChronicleHashCorruptionImpl.format; +import static net.openhft.chronicle.map.ChronicleHashCorruptionImpl.report; import static net.openhft.chronicle.map.DefaultSpi.mapEntryOperations; import static net.openhft.chronicle.map.DefaultSpi.mapRemoteOperations; import static net.openhft.chronicle.map.VanillaChronicleMap.alignAddr; @@ -151,6 +154,15 @@ public final class ChronicleMapBuilder implements new ConcurrentHashMap<>(128); private static int MAX_BOOTSTRAPPING_HEADER_SIZE = (int) MemoryUnit.KILOBYTES.toBytes(16); + private static final Logger chronicleMapLogger = LoggerFactory.getLogger(ChronicleMap.class); + private static final ChronicleHashCorruption.Listener defaultChronicleMapCorruptionListener = + corruption -> { + if (corruption.exception() != null) { + chronicleMapLogger.error(corruption.message(), corruption.exception()); + } else { + chronicleMapLogger.error(corruption.message()); + } + }; private String name; SerializationBuilder keyBuilder; @@ -1469,7 +1481,7 @@ public final class ChronicleMapBuilder implements // clone() to make this builder instance thread-safe, because createWithFile() method // computes some state based on configurations, but doesn't synchronize on configuration // changes. - return clone().createWithFile(file, false, false); + return clone().createWithFile(file, false, false, null); } @Override @@ -1480,8 +1492,16 @@ public final class ChronicleMapBuilder implements @Override public ChronicleMap createOrRecoverPersistedTo(File file, boolean sameLibraryVersion) throws IOException { + return createOrRecoverPersistedTo(file, sameLibraryVersion, + defaultChronicleMapCorruptionListener); + } + + @Override + public ChronicleMap createOrRecoverPersistedTo( + File file, boolean sameLibraryVersion, + ChronicleHashCorruption.Listener corruptionListener) throws IOException { if (file.exists()) { - return recoverPersistedTo(file, sameLibraryVersion); + return recoverPersistedTo(file, sameLibraryVersion, corruptionListener); } else { return createPersistedTo(file); } @@ -1490,7 +1510,16 @@ public final class ChronicleMapBuilder implements @Override public ChronicleMap recoverPersistedTo( File file, boolean sameBuilderConfigAndLibraryVersion) throws IOException { - return clone().createWithFile(file, true, sameBuilderConfigAndLibraryVersion); + return recoverPersistedTo(file, sameBuilderConfigAndLibraryVersion, + defaultChronicleMapCorruptionListener); + } + + @Override + public ChronicleMap recoverPersistedTo( + File file, boolean sameBuilderConfigAndLibraryVersion, + ChronicleHashCorruption.Listener corruptionListener) throws IOException { + return clone().createWithFile(file, true, sameBuilderConfigAndLibraryVersion, + corruptionListener); } @Override @@ -1502,7 +1531,8 @@ public final class ChronicleMapBuilder implements } private ChronicleMap createWithFile( - File file, boolean recover, boolean overrideBuilderConfig) throws IOException { + File file, boolean recover, boolean overrideBuilderConfig, + ChronicleHashCorruption.Listener corruptionListener) throws IOException { if (overrideBuilderConfig && !recover) throw new AssertionError("recover -> overrideBuilderConfig"); replicated = replicationIdentifier != -1; @@ -1522,7 +1552,8 @@ public final class ChronicleMapBuilder implements try { VanillaChronicleMap result; if (raf.length() > 0) { - result = openWithExistingFile(file, raf, resources, recover, overrideBuilderConfig); + result = openWithExistingFile(file, raf, resources, recover, overrideBuilderConfig, + corruptionListener); } else { // Single-element arrays allow to modify variables within lambda @@ -1548,7 +1579,7 @@ public final class ChronicleMapBuilder implements headerSize); } else { result = openWithExistingFile(file, raf, resources, recover, - overrideBuilderConfig); + overrideBuilderConfig, corruptionListener); } } prepareMapPublication(result); @@ -1639,7 +1670,9 @@ public final class ChronicleMapBuilder implements * @return ByteBuffer, in [position, limit) range the self bootstrapping header is read */ private static ByteBuffer readSelfBootstrappingHeader( - File file, RandomAccessFile raf, int headerSize, boolean recover) throws IOException { + File file, RandomAccessFile raf, int headerSize, boolean recover, + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) throws IOException { if (raf.length() < headerSize + SELF_BOOTSTRAPPING_HEADER_OFFSET) { throw throwRecoveryOrReturnIOException(file, "The file is shorter than the header size: " + headerSize + @@ -1658,7 +1691,10 @@ public final class ChronicleMapBuilder implements int sizeWord = headerBuffer.getInt(SIZE_WORD_OFFSET); if (!SizePrefixedBlob.isReady(sizeWord)) { if (recover) { - LOG.error("file={}: size-prefixed blob readiness bit is set to NOT_COMPLETE", file); + report(corruptionListener, corruption, -1, () -> + format("file={}: size-prefixed blob readiness bit is set to NOT_COMPLETE", + file) + ); // the bit will be overwritten to READY in the end of recovery procedure, so nothing // to fix right here } else { @@ -1688,12 +1724,15 @@ public final class ChronicleMapBuilder implements private VanillaChronicleMap openWithExistingFile( File file, RandomAccessFile raf, ChronicleHashResources resources, - boolean recover, boolean overrideBuilderConfig) + boolean recover, boolean overrideBuilderConfig, + ChronicleHashCorruption.Listener corruptionListener) throws IOException { + ChronicleHashCorruptionImpl corruption = recover ? new ChronicleHashCorruptionImpl() : null; try { int headerSize = waitUntilReady(raf, file, recover); FileChannel fileChannel = raf.getChannel(); - ByteBuffer headerBuffer = readSelfBootstrappingHeader(file, raf, headerSize, recover); + ByteBuffer headerBuffer = readSelfBootstrappingHeader( + file, raf, headerSize, recover, corruptionListener, corruption); if (headerSize != headerBuffer.remaining()) throw new AssertionError(); boolean headerCorrect = checkSumSelfBootstrappingHeader(headerBuffer, headerSize); @@ -1728,7 +1767,7 @@ public final class ChronicleMapBuilder implements } else { if (!headerWritten) writeNotComplete(fileChannel, headerBuffer, headerSize); - map.recover(resources); + map.recover(resources, corruptionListener, corruption); commitChronicleMapReady(map, raf, headerBuffer, headerSize); } return map; diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 1ddbfe430..079316be8 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -24,6 +24,7 @@ import net.openhft.chronicle.bytes.PointerBytesStore; import net.openhft.chronicle.core.Jvm; import net.openhft.chronicle.core.io.Closeable; import net.openhft.chronicle.hash.ChronicleHashClosedException; +import net.openhft.chronicle.hash.ChronicleHashCorruption; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.*; import net.openhft.chronicle.hash.impl.stage.entry.LocksInterface; @@ -163,10 +164,12 @@ public class VanillaChronicleMap initOwnTransients(); } - public void recover(ChronicleHashResources resources) throws IOException { - basicRecover(resources); + public void recover( + ChronicleHashResources resources, ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) throws IOException { + basicRecover(resources, corruptionListener, corruption); try (IterationContext iterationContext = iterationContext()) { - iterationContext.recoverSegments(); + iterationContext.recoverSegments(corruptionListener, corruption); } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/IterationContext.java b/src/main/java/net/openhft/chronicle/map/impl/IterationContext.java index 753a8734f..67fc4acec 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/IterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/IterationContext.java @@ -17,6 +17,8 @@ package net.openhft.chronicle.map.impl; +import net.openhft.chronicle.hash.ChronicleHashCorruption; +import net.openhft.chronicle.map.ChronicleHashCorruptionImpl; import net.openhft.chronicle.map.MapEntry; import net.openhft.chronicle.map.MapSegmentContext; @@ -28,5 +30,7 @@ public interface IterationContext extends MapEntry, MapSegmentCon void initSegmentIndex(int segmentIndex); - void recoverSegments(); + void recoverSegments( + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedTierRecovery.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedTierRecovery.java index 1912288bf..b45d7bef1 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedTierRecovery.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedTierRecovery.java @@ -18,33 +18,40 @@ package net.openhft.chronicle.map.impl.stage.iter; import net.openhft.chronicle.algo.bitset.ReusableBitSet; +import net.openhft.chronicle.hash.ChronicleHashCorruption; import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable; import net.openhft.chronicle.hash.impl.VanillaChronicleHash; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; -import net.openhft.chronicle.hash.impl.stage.hash.LogHolder; import net.openhft.chronicle.hash.impl.stage.iter.TierRecovery; +import net.openhft.chronicle.map.ChronicleHashCorruptionImpl; import net.openhft.chronicle.map.ReplicatedChronicleMap; import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; +import static net.openhft.chronicle.map.ChronicleHashCorruptionImpl.format; +import static net.openhft.chronicle.map.ChronicleHashCorruptionImpl.report; + @Staged public class ReplicatedTierRecovery extends TierRecovery { @StageRef ReplicatedChronicleMapHolder rh; @StageRef SegmentStages s; @StageRef ReplicatedMapEntryStages e; - @StageRef LogHolder lh; @Override - public void removeDuplicatesInSegment() { - super.removeDuplicatesInSegment(); - recoverTierDeleted(); + public void removeDuplicatesInSegment( + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { + super.removeDuplicatesInSegment(corruptionListener, corruption); + recoverTierDeleted(corruptionListener, corruption); cleanupModificationIterationBits(); } - private void recoverTierDeleted() { + private void recoverTierDeleted( + ChronicleHashCorruption.Listener corruptionListener, + ChronicleHashCorruptionImpl corruption) { VanillaChronicleHash h = rh.h(); CompactOffHeapLinearHashTable hl = h.hashLookup; long hlAddr = s.tierBaseAddr; @@ -62,8 +69,11 @@ public class ReplicatedTierRecovery extends TierRecovery { hlPos = hl.step(hlPos); } while (hlPos != 0); if (s.tierDeleted() != deleted) { - lh.LOG.error("wrong deleted counter for tier with index {}, stored: {}, should be: {}", - s.tierIndex, s.tierDeleted(), deleted); + long finalDeleted = deleted; + report(corruptionListener, corruption, s.segmentIndex, () -> + format("wrong deleted counter for tier with index {}, stored: {}, should be: {}", + s.tierIndex, s.tierDeleted(), finalDeleted) + ); s.tierDeleted(deleted); } } diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index e9c058109..5f8de25c1 100755 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -19,11 +19,14 @@ package net.openhft.chronicle.set; import net.openhft.chronicle.hash.ChronicleHashBuilder; import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI; +import net.openhft.chronicle.hash.ChronicleHashCorruption; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.serialization.*; import net.openhft.chronicle.map.*; import net.openhft.chronicle.map.replication.MapRemoteOperations; import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; @@ -31,7 +34,7 @@ import java.io.IOException; /** * {@code ChronicleSetBuilder} manages the whole set of {@link ChronicleSet} configurations, could * be used as a classic builder and/or factory. - * + *

*

{@code ChronicleMapBuilder} is mutable, see a note in {@link * ChronicleHashBuilder} interface documentation. * @@ -42,6 +45,16 @@ import java.io.IOException; public final class ChronicleSetBuilder implements ChronicleHashBuilder, ChronicleSetBuilder> { + private static final Logger chronicleSetLogger = LoggerFactory.getLogger(ChronicleSet.class); + private static final ChronicleHashCorruption.Listener defaultChronicleSetCorruptionListener = + corruption -> { + if (corruption.exception() != null) { + chronicleSetLogger.error(corruption.message(), corruption.exception()); + } else { + chronicleSetLogger.error(corruption.message()); + } + }; + private ChronicleMapBuilder chronicleMapBuilder; private ChronicleSetBuilderPrivateAPI privateAPI; @@ -60,9 +73,9 @@ public final class ChronicleSetBuilder * Returns a new {@code ChronicleSetBuilder} instance which is able to {@linkplain #create() * create} sets with the specified key class. * - * @param keyClass class object used to infer key type and discover it's properties via - * reflection - * @param key type of the sets, created by the returned builder + * @param keyClass class object used to infer key type and discover it's properties via + * reflection + * @param key type of the sets, created by the returned builder * @return a new builder for the given key class */ public static ChronicleSetBuilder of(Class keyClass) { @@ -125,9 +138,9 @@ public final class ChronicleSetBuilder * encoded (and each character takes 2 bytes on-heap), because default off-heap {@link String} * encoding is UTF-8 in {@code ChronicleSet}.) * + * @param averageKeySize the average size in bytes of the key * @see #constantKeySizeBySample(Object) * @see #actualChunkSize(int) - * @param averageKeySize the average size in bytes of the key */ @Override public ChronicleSetBuilder averageKeySize(double averageKeySize) { @@ -143,7 +156,7 @@ public final class ChronicleSetBuilder /** * {@inheritDoc} - * + *

*

For example, if your keys are Git commit hashes:

{@code
      * Set gitCommitsOfInterest = ChronicleSetBuilder.of(byte[].class)
      *     .constantKeySizeBySample(new byte[20])
@@ -306,23 +319,38 @@ public final class ChronicleSetBuilder
 
     @Override
     public ChronicleSet createOrRecoverPersistedTo(File file) throws IOException {
-        ChronicleMap map = chronicleMapBuilder.createOrRecoverPersistedTo(file);
-        return new SetFromMap<>((VanillaChronicleMap) map);
+        return createOrRecoverPersistedTo(file, true);
     }
 
     @Override
     public ChronicleSet createOrRecoverPersistedTo(File file, boolean sameLibraryVersion)
             throws IOException {
-        ChronicleMap map =
-                chronicleMapBuilder.createOrRecoverPersistedTo(file, sameLibraryVersion);
+        return createOrRecoverPersistedTo(file, sameLibraryVersion,
+                defaultChronicleSetCorruptionListener);
+    }
+
+    @Override
+    public ChronicleSet createOrRecoverPersistedTo(
+            File file, boolean sameLibraryVersion,
+            ChronicleHashCorruption.Listener corruptionListener) throws IOException {
+        ChronicleMap map = chronicleMapBuilder.createOrRecoverPersistedTo(
+                file, sameLibraryVersion, corruptionListener);
         return new SetFromMap<>((VanillaChronicleMap) map);
     }
 
     @Override
     public ChronicleSet recoverPersistedTo(File file, boolean sameBuilderConfigAndLibraryVersion)
             throws IOException {
-        ChronicleMap map =
-                chronicleMapBuilder.recoverPersistedTo(file, sameBuilderConfigAndLibraryVersion);
+        return recoverPersistedTo(file, sameBuilderConfigAndLibraryVersion,
+                defaultChronicleSetCorruptionListener);
+    }
+
+    @Override
+    public ChronicleSet recoverPersistedTo(
+            File file, boolean sameBuilderConfigAndLibraryVersion,
+            ChronicleHashCorruption.Listener corruptionListener) throws IOException {
+        ChronicleMap map = chronicleMapBuilder.recoverPersistedTo(
+                file, sameBuilderConfigAndLibraryVersion, corruptionListener);
         return new SetFromMap<>((VanillaChronicleMap) map);
     }
 
diff --git a/src/test/java/net/openhft/chronicle/map/RecoverTest.java b/src/test/java/net/openhft/chronicle/map/RecoverTest.java
index a21eca920..fd00cb67d 100644
--- a/src/test/java/net/openhft/chronicle/map/RecoverTest.java
+++ b/src/test/java/net/openhft/chronicle/map/RecoverTest.java
@@ -17,8 +17,12 @@
 
 package net.openhft.chronicle.map;
 
-import net.openhft.chronicle.hash.ChronicleHashBuilder;
+import net.openhft.chronicle.core.values.LongValue;
+import net.openhft.chronicle.hash.ChecksumEntry;
 import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI;
+import net.openhft.chronicle.hash.ChronicleHashCorruption;
+import net.openhft.chronicle.values.Values;
+import org.junit.Assert;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -32,8 +36,10 @@ import java.nio.channels.FileChannel;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
 
-import static org.junit.Assert.assertNull;
+import static net.openhft.chronicle.map.ChronicleMapTest.getPersistenceFile;
+import static org.junit.Assert.*;
 
 public class RecoverTest {
 
@@ -117,4 +123,47 @@ public class RecoverTest {
             }
         }
     }
+
+    @Test
+    public void testCorruptedEntryRecovery() throws IOException {
+        File file = getPersistenceFile();
+        try (ChronicleMap map = ChronicleMap
+                .of(Integer.class, LongValue.class)
+                .entries(1)
+                .createPersistedTo(file)) {
+
+            LongValue value = Values.newHeapInstance(LongValue.class);
+            value.setValue(42);
+            map.put(1, value);
+
+            try (ExternalMapQueryContext c = map.queryContext(1)) {
+                // Update lock required for calling ChecksumEntry.checkSum()
+                c.updateLock().lock();
+                MapEntry entry = c.entry();
+                assertNotNull(entry);
+                ChecksumEntry checksumEntry = (ChecksumEntry) entry;
+                assertTrue(checksumEntry.checkSum());
+
+                // to access off-heap bytes, should call value().getUsing() with Native value
+                // provided. Simple get() return Heap value by default
+                LongValue nativeValue =
+                        entry.value().getUsing(Values.newNativeReference(LongValue.class));
+                // This value bytes update bypass Chronicle Map internals, so checksum is not
+                // updated automatically
+                nativeValue.setValue(43);
+                Assert.assertFalse(checksumEntry.checkSum());
+            }
+        }
+
+        AtomicInteger corruptionCounter = new AtomicInteger(0);
+        ChronicleHashCorruption.Listener corruptionListener =
+                corruption -> corruptionCounter.incrementAndGet();
+        //noinspection EmptyTryBlock
+        try (ChronicleMap ignore = ChronicleMap
+                .of(Integer.class, LongValue.class)
+                .entries(1)
+                .createOrRecoverPersistedTo(file, true, corruptionListener)) {
+        }
+        assertTrue(corruptionCounter.get() > 0);
+    }
 }