Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Add ChronicleHashCorruption listening interface, and Beta annotation

Browse Source
This commit is contained in:
leventov
2017-02-26 21:02:34 -06:00
parent 0cf26fb9b6
commit 7f3cca96d2
14 changed files with 669 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/Beta.java
+42
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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."
*
* <p>This annotation is a copy of Guava's <a href="
* https://google.github.io/guava/releases/19.0/api/docs/com/google/common/annotations/Beta.html">
* {@code Beta}</a>.
*/
@Retention(value = RetentionPolicy.CLASS)
@Target(value = {ANNOTATION_TYPE, CONSTRUCTOR, FIELD,METHOD, TYPE})
@Documented
@Beta
public @interface Beta {
}
src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
+108 -4
View File
@@ -528,6 +528,7 @@ public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?, ?>,
* @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<K, H extends ChronicleHash<K, ?, ?, ?>,
*
* @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<K, H extends ChronicleHash<K, ?, ?, ?>,
* @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<K, H extends ChronicleHash<K, ?, ?, ?>,
* file doesn't exist, and {@link #recoverPersistedTo(File, boolean)
* recoverPersistedTo(file, sameLibraryVersion)}, if the file exists.
*
* <p>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 <b>the same version
@@ -577,6 +584,7 @@ public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?, ?>,
* (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<K, H extends ChronicleHash<K, ?, ?, ?>,
* @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.
*
* <p>If this procedure encounters corruptions, it fixes them (<i>recovers</i> 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 <b>the same version
* of the Chronicle Map library was used</b>. 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;
/**
* <i>Recovers</i> 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<K, H extends ChronicleHash<K, ?, ?, ?>,
* which were never put into the Chronicle Hash).
*
* <p>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
* <p>"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. <i>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<K, H extends ChronicleHash<K, ?, ?, ?>,
* <a href="https://github.com/OpenHFT/Chronicle-Map#recovery">Recovery section</a> in the
* Chronicle Map tutorial for more information.
*
* <p>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.
*
* <p>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.
*
* <p>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<K, H extends ChronicleHash<K, ?, ?, ?>,
* 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<K, H extends ChronicleHash<K, ?, ?, ?>,
* @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;
/**
* <i>Recovers</i> 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).
*
* <p>This method, unlike {@link #createPersistedTo(File)} or {@link
* #createOrRecoverPersistedTo(File, boolean, ChronicleHashCorruption.Listener)} methods,
* expects that the given file already exists.
*
* <p>"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. <i>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.</i> See the
* <a href="https://github.com/OpenHFT/Chronicle-Map#recovery">Recovery section</a> in the
* Chronicle Map tutorial for more information.
*
* <p>If this procedure encounters corruptions, it fixes them (<i>recovers</i> from them) and
* notifies the provided corruption listener with the details about the corruption. See the
* documentation for {@link ChronicleHashCorruption} for more information.
*
* <p>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 <b>with the same version of the Chronicle Map library</b>.
* 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
*/
src/main/java/net/openhft/chronicle/hash/ChronicleHashCorruption.java
+86
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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 <a
* href="https://github.com/OpenHFT/Chronicle-Map#recovery">recovery</a>.
*
* <p>Recovery procedure doesn't guarantee accuracy of the corruption events. Only two things are
* guaranteed:
* <ol>
* <li>if {@link Listener} didn't receive any corruption events, the recovered Chronicle Map
* was not corrupted;
* </li>
* <li>if {@link Listener} received some corruption events, the recovered Chronicle Map was
* corrupted.</li>
* </ol>
*
* <p>{@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 <a href="https://github.com/OpenHFT/Chronicle-Map#recovery">recovery</a>
* 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);
}
}
src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
+40 -22
View File
@@ -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<K,
C extends HashEntry<K>, SC extends HashSegmentContext<K, ?>,
ECQ extends ExternalHashQueryContext<K>>
implements ChronicleHash<K, C, SC, ECQ>, 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<K,
createStoreAndSegments(map(dataStoreSize(), 0));
}
public final void basicRecover(ChronicleHashResources resources)
public final void basicRecover(
ChronicleHashResources resources, ChronicleHashCorruption.Listener corruptionListener,
ChronicleHashCorruptionImpl corruption)
throws IOException {
this.resources = resources;
long segmentHeadersOffset = globalMutableState().getSegmentHeadersOffset();
@@ -525,45 +526,62 @@ public abstract class VanillaChronicleHash<K,
}
initBytesStoreAndHeadersViews(map(dataStoreSize, 0));
resetGlobalMutableStateLock();
recoverAllocatedExtraTierBulks(allocatedExtraTierBulks);
recoverSegmentHeadersOffset(segmentHeadersOffset);
recoverDataStoreSize(dataStoreSize);
resetGlobalMutableStateLock(corruptionListener, corruption);
recoverAllocatedExtraTierBulks(allocatedExtraTierBulks, corruptionListener, corruption);
recoverSegmentHeadersOffset(segmentHeadersOffset, corruptionListener, corruption);
recoverDataStoreSize(dataStoreSize, corruptionListener, corruption);
initOffsetsAndBulks();
}
private void resetGlobalMutableStateLock() {
private void resetGlobalMutableStateLock(
ChronicleHashCorruption.Listener corruptionListener,
ChronicleHashCorruptionImpl corruption) {
long lockAddr = globalMutableStateAddress() + GLOBAL_MUTABLE_STATE_LOCK_OFFSET;
LockingStrategy lockingStrategy = globalMutableStateLockingStrategy;
long lockState = lockingStrategy.getState(nativeAccess(), null, lockAddr);
if (lockState != lockingStrategy.resetState()) {
LOG.error("global mutable state lock of map at {} is not clear: {}",
file, lockingStrategy.toString(lockState));
report(corruptionListener, corruption, -1, () ->
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);
}
}
src/main/java/net/openhft/chronicle/hash/impl/stage/iter/SegmentsRecovery.java
+60 -32
View File
@@ -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);
}
}
src/main/java/net/openhft/chronicle/hash/impl/stage/iter/TierRecovery.java
+85 -44
View File
@@ -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() : "<deleted>",
!e.entryDeleted() ? e.value() : "<deleted>");
!e.entryDeleted() ? e.value() : "<deleted>")
);
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;
src/main/java/net/openhft/chronicle/map/ChronicleHashCorruptionImpl.java
+77
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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<String> messageSupplier) {
corruption.set(segmentIndex, messageSupplier, null);
corruptionListener.onCorruption(corruption);
}
public static void reportException(
ChronicleHashCorruption.Listener corruptionListener,
ChronicleHashCorruptionImpl corruption, int segmentIndex,
Supplier<String> 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<String> messageSupplier;
private Throwable exception;
private String message;
private void set(int segmentIndex, Supplier<String> 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;
}
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+50 -11
View File
@@ -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<K, V> 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<K> keyBuilder;
@@ -1469,7 +1481,7 @@ public final class ChronicleMapBuilder<K, V> 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<K, V> implements
@Override
public ChronicleMap<K, V> createOrRecoverPersistedTo(File file, boolean sameLibraryVersion)
throws IOException {
return createOrRecoverPersistedTo(file, sameLibraryVersion,
defaultChronicleMapCorruptionListener);
}
@Override
public ChronicleMap<K, V> 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<K, V> implements
@Override
public ChronicleMap<K, V> recoverPersistedTo(
File file, boolean sameBuilderConfigAndLibraryVersion) throws IOException {
return clone().createWithFile(file, true, sameBuilderConfigAndLibraryVersion);
return recoverPersistedTo(file, sameBuilderConfigAndLibraryVersion,
defaultChronicleMapCorruptionListener);
}
@Override
public ChronicleMap<K, V> 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<K, V> implements
}
private ChronicleMap<K, V> 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<K, V> implements
try {
VanillaChronicleMap<K, V, ?> 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<K, V> implements
headerSize);
} else {
result = openWithExistingFile(file, raf, resources, recover,
overrideBuilderConfig);
overrideBuilderConfig, corruptionListener);
}
}
prepareMapPublication(result);
@@ -1639,7 +1670,9 @@ public final class ChronicleMapBuilder<K, V> 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<K, V> 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<K, V> implements
private VanillaChronicleMap<K, V, ?> 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<K, V> implements
} else {
if (!headerWritten)
writeNotComplete(fileChannel, headerBuffer, headerSize);
map.recover(resources);
map.recover(resources, corruptionListener, corruption);
commitChronicleMapReady(map, raf, headerBuffer, headerSize);
}
return map;
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+6 -3
View File
@@ -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<K, V, R>
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<K, V, ?> iterationContext = iterationContext()) {
iterationContext.recoverSegments();
iterationContext.recoverSegments(corruptionListener, corruption);
}
}
src/main/java/net/openhft/chronicle/map/impl/IterationContext.java
+5 -1
View File
@@ -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<K, V, R> extends MapEntry<K, V>, MapSegmentCon
void initSegmentIndex(int segmentIndex);
void recoverSegments();
void recoverSegments(
ChronicleHashCorruption.Listener corruptionListener,
ChronicleHashCorruptionImpl corruption);
}
src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedTierRecovery.java
+18 -8
View File
@@ -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);
}
}
src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
+40 -12
View File
@@ -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.
*
* <p>
* <p>{@code ChronicleMapBuilder} is mutable, see a note in {@link
* ChronicleHashBuilder} interface documentation.
*
@@ -42,6 +45,16 @@ import java.io.IOException;
public final class ChronicleSetBuilder<K>
implements ChronicleHashBuilder<K, ChronicleSet<K>, ChronicleSetBuilder<K>> {
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<K, DummyValue> chronicleMapBuilder;
private ChronicleSetBuilderPrivateAPI<K> privateAPI;
@@ -60,9 +73,9 @@ public final class ChronicleSetBuilder<K>
* 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 <K> 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 <K> key type of the sets, created by the returned builder
* @return a new builder for the given key class
*/
public static <K> ChronicleSetBuilder<K> of(Class<K> keyClass) {
@@ -125,9 +138,9 @@ public final class ChronicleSetBuilder<K>
* 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<K> averageKeySize(double averageKeySize) {
@@ -143,7 +156,7 @@ public final class ChronicleSetBuilder<K>
/**
* {@inheritDoc}
*
* <p>
* <p>For example, if your keys are Git commit hashes:<pre>{@code
* Set<byte[]> gitCommitsOfInterest = ChronicleSetBuilder.of(byte[].class)
* .constantKeySizeBySample(new byte[20])
@@ -306,23 +319,38 @@ public final class ChronicleSetBuilder<K>
@Override
public ChronicleSet<K> createOrRecoverPersistedTo(File file) throws IOException {
ChronicleMap<K, DummyValue> map = chronicleMapBuilder.createOrRecoverPersistedTo(file);
return new SetFromMap<>((VanillaChronicleMap<K, DummyValue, ?>) map);
return createOrRecoverPersistedTo(file, true);
}
@Override
public ChronicleSet<K> createOrRecoverPersistedTo(File file, boolean sameLibraryVersion)
throws IOException {
ChronicleMap<K, DummyValue> map =
chronicleMapBuilder.createOrRecoverPersistedTo(file, sameLibraryVersion);
return createOrRecoverPersistedTo(file, sameLibraryVersion,
defaultChronicleSetCorruptionListener);
}
@Override
public ChronicleSet<K> createOrRecoverPersistedTo(
File file, boolean sameLibraryVersion,
ChronicleHashCorruption.Listener corruptionListener) throws IOException {
ChronicleMap<K, DummyValue> map = chronicleMapBuilder.createOrRecoverPersistedTo(
file, sameLibraryVersion, corruptionListener);
return new SetFromMap<>((VanillaChronicleMap<K, DummyValue, ?>) map);
}
@Override
public ChronicleSet<K> recoverPersistedTo(File file, boolean sameBuilderConfigAndLibraryVersion)
throws IOException {
ChronicleMap<K, DummyValue> map =
chronicleMapBuilder.recoverPersistedTo(file, sameBuilderConfigAndLibraryVersion);
return recoverPersistedTo(file, sameBuilderConfigAndLibraryVersion,
defaultChronicleSetCorruptionListener);
}
@Override
public ChronicleSet<K> recoverPersistedTo(
File file, boolean sameBuilderConfigAndLibraryVersion,
ChronicleHashCorruption.Listener corruptionListener) throws IOException {
ChronicleMap<K, DummyValue> map = chronicleMapBuilder.recoverPersistedTo(
file, sameBuilderConfigAndLibraryVersion, corruptionListener);
return new SetFromMap<>((VanillaChronicleMap<K, DummyValue, ?>) map);
}