From 97e4c12de1abd3268d18d413af50f0cff263baed Mon Sep 17 00:00:00 2001 From: leventov Date: Mon, 4 Jan 2016 23:24:39 +0300 Subject: [PATCH] HCOLL-378 #36 Add ChronicleHashBuilder.recoverPersistedTo() and ChronicleHashInstanceBuilder.recover() functionality --- .../openhft/chronicle/hash/AbstractData.java | 22 +- .../chronicle/hash/ChronicleHashBuilder.java | 44 +++ .../hash/ChronicleHashInstanceBuilder.java | 37 ++- .../ChronicleHashRecoveryFailedException.java | 52 ++++ .../chronicle/hash/impl/BigSegmentHeader.java | 20 ++ .../impl/CompactOffHeapLinearHashTable.java | 19 +- .../IntCompactOffHeapLinearHashTable.java | 4 +- .../LongCompactOffHeapLinearHashTable.java | 4 +- .../chronicle/hash/impl/SegmentHeader.java | 5 + .../hash/impl/VanillaChronicleHash.java | 138 +++++++--- .../impl/stage/entry/ChecksumStrategy.java | 4 + .../entry/HashEntryChecksumStrategy.java | 10 +- .../impl/stage/entry/HashEntryStages.java | 2 +- .../impl/stage/entry/HashLookupSearch.java | 4 +- .../impl/stage/entry/NoChecksumStrategy.java | 10 + .../hash/impl/stage/entry/SegmentStages.java | 10 +- .../impl/stage/iter/HashSegmentIteration.java | 2 +- .../impl/stage/iter/SegmentsRecovery.java | 160 +++++++++++ .../hash/impl/stage/iter/TierRecovery.java | 256 ++++++++++++++++++ .../chronicle/map/ChronicleMapBuilder.java | 97 +++++-- .../chronicle/map/MapInstanceBuilder.java | 44 ++- .../chronicle/map/ReplicatedChronicleMap.java | 30 +- .../chronicle/map/VanillaChronicleMap.java | 7 + .../chronicle/map/impl/IterationContext.java | 5 + .../map/impl/MapIterationContext.java | 7 +- .../impl/ReplicatedMapIterationContext.java | 9 +- .../map/impl/stage/entry/MapEntryStages.java | 2 +- .../impl/stage/iter/MapSegmentIteration.java | 16 ++ .../stage/iter/ReplicatedTierRecovery.java | 99 +++++++ .../chronicle/set/ChronicleSetBuilder.java | 9 + .../chronicle/set/SetInstanceBuilder.java | 19 +- 31 files changed, 1026 insertions(+), 121 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/hash/ChronicleHashRecoveryFailedException.java create mode 100644 src/main/java/net/openhft/chronicle/hash/impl/stage/iter/SegmentsRecovery.java create mode 100644 src/main/java/net/openhft/chronicle/hash/impl/stage/iter/TierRecovery.java create mode 100644 src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedTierRecovery.java diff --git a/src/main/java/net/openhft/chronicle/hash/AbstractData.java b/src/main/java/net/openhft/chronicle/hash/AbstractData.java index 732fcf1f6..56335c995 100644 --- a/src/main/java/net/openhft/chronicle/hash/AbstractData.java +++ b/src/main/java/net/openhft/chronicle/hash/AbstractData.java @@ -16,6 +16,8 @@ package net.openhft.chronicle.hash; +import net.openhft.chronicle.bytes.RandomDataInput; + /** * Defines reasonable defaults for {@code Data}'s {@code equals()}, {@code hashCode()} and * {@code toString()}. They should be default implementations in the {@code Data} interface itself, @@ -47,10 +49,26 @@ public abstract class AbstractData implements Data { } /** - * Delegates to {@code Data}'s object {@code toString()}. + * Delegates to {@code Data}'s object {@code toString()}. If deserialization fails with + * exception (e. g. if data bytes are corrupted, and represent not a valid serialized form of + * an object), traces the data's bytes and the exception. */ @Override public String toString() { - return get().toString(); + T object; + try { + object = get(); + } catch (Exception e) { + StringBuilder sb = new StringBuilder(); + sb.append("failed to deserialize object from data with bytes: ["); + RandomDataInput bs = bytes(); + for (long off = offset(), lim = offset() + size(); off < lim; off++) { + sb.append(bs.printable(off)); + } + sb.append("], exception: "); + sb.append(e); + return sb.toString(); + } + return object.toString(); } } diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java index 3ae8fa5b6..f8c3038d9 100755 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java @@ -29,6 +29,7 @@ import net.openhft.chronicle.set.ChronicleSetBuilder; import org.jetbrains.annotations.NotNull; import java.io.File; +import java.io.FileNotFoundException; import java.io.IOException; import java.util.HashMap; import java.util.HashSet; @@ -615,11 +616,54 @@ public interface ChronicleHashBuilder, * @see ChronicleHash#close() * @see #create() * @see ChronicleHashInstanceBuilder#persistedTo(File) + * @see #recoverPersistedTo(File, boolean) */ H createPersistedTo(File file) throws IOException; + /** + * Recovers and opens the hash container, persisted to the specified file, using {@link + * #createPersistedTo(File)} method. This method should be used to open a persisted Chronicle + * Map after accessor process crash (that might leave some locks "acquired" therefore some + * segments inaccessible), or accessor process external termination (that, in addition to + * inaccessible segments, might lead to leaks in the Chronicle Hash memory), or sudden power + * loss, or 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)}, expects the mapping file already + * exists. + * + *

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. + * + *

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 + * {@code sameBuilderConfig} argument (another requirement is running the same version of the + * Chronicle Map library). Otherwise, if the header of the given persisted Chronicle Hash file + * is corrupted, this method is likely to be unable to recover and throw {@link + * ChronicleHashRecoveryFailedException}, or even worse, to corrupt the file further. However, + * the header shouldn't be corrupted on process crash/termination or power loss, only on direct + * file corruption. + * + * @param file a hash container was mapped to the given file + * @param sameBuilderConfig 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 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 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 #createPersistedTo(File) + */ + H recoverPersistedTo(File file, boolean sameBuilderConfig) throws IOException; + /** * @deprecated don't use private API in the client code */ + @Deprecated ChronicleHashBuilderPrivateAPI privateAPI(); } diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java index 9d9710bc5..33a91c503 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceBuilder.java @@ -23,6 +23,7 @@ import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.set.ChronicleSet; import java.io.File; +import java.io.FileNotFoundException; import java.io.IOException; /** @@ -101,17 +102,47 @@ public interface ChronicleHashInstanceBuilder { /** * Creates (or opens existing, if {@link #persistedTo(File)} is configured with the existing - * file) a Chronicle Hash from this builder. After this method is called once the builder - * couldn't be used to create new Chronicle Hash instances, a new {@code + * file) and returns a Chronicle Hash from this builder. After this method is called once the + * builder couldn't be used to create new Chronicle Hash instances, a new {@code * ChronicleHashInstanceBuilder} should be created via {@link ChronicleHashBuilder#instance()}. * * @return a new in-memory or persisted Chronicle Hash instance * @throws IOException if any IO error, related to off-heap memory allocation or file mapping, * or establishing replication connections, occurs - * @throws IllegalStateException if {@code create()} method has already been called on this + * @throws IllegalStateException if {@code create()} or {@link #recover(boolean)} method has + * already been called on this * {@code ChronicleHashInstanceBuilder} * @see ChronicleHashBuilder#create() * @see ChronicleHashBuilder#createPersistedTo(File) */ H create() throws IOException; + + /** + * Recovers and returns a Chronicle Hash instance, persisted to the file, configured via {@link + * #persistedTo(File)} method. See documentation to {@link + * ChronicleHashBuilder#recoverPersistedTo(File, boolean)} for more information of Chronicle + * Hash instance recovery. + * + * @param sameBuilderConfig if the {@link ChronicleHashBuilder}, from which this {@code + * ChronicleHashInstanceBuilder} inherits (i. e. this {@code ChronicleHashInstanceBuilder} is + * created via calling {@link ChronicleHashBuilder#instance()} on that {@link + * ChronicleHashBuilder}) is configured with the same configurations, as the builder, which + * created the persistence 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 specified file + * @throws IllegalStateException if persistence file is not configured (or explicitly configured + * to {@code null}) for this {@code ChronicleHashInstanceBuilder}, using {@link + * #persistedTo(File)} method, or if {@link #create()} or {@code recover()} method has already + * been called on this builder + * @throws FileNotFoundException if the specified persistence 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 ChronicleHashBuilder#recoverPersistedTo(File, boolean) + */ + H recover(boolean sameBuilderConfig) throws IOException; } diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashRecoveryFailedException.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashRecoveryFailedException.java new file mode 100644 index 000000000..e0a70840c --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashRecoveryFailedException.java @@ -0,0 +1,52 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * 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.io.File; + +/** + * This exception is thrown, when a Chronicle Hash recovery using {@link + * ChronicleHashBuilder#recoverPersistedTo(File, boolean)} or {@link + * ChronicleHashInstanceBuilder#recover(boolean)} method is impossible, for example, if the + * persistence file is corrupted too much. + * + * @see ChronicleHashBuilder#recoverPersistedTo(File, boolean) + * @see ChronicleHashInstanceBuilder#recover(boolean) + */ +public final class ChronicleHashRecoveryFailedException extends RuntimeException { + private static final long serialVersionUID = 0L; + + /** + * Constructs a new {@code ChronicleHashRecoveryFailedException} with the specified cause. + * + * @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method) + */ + public ChronicleHashRecoveryFailedException(Throwable cause) { + super(cause); + } + + /** + * Constructs a new {@code ChronicleHashRecoveryFailedException} with the specified detail + * message. + * + * @param message the detail message. The detail message is saved for later retrieval by the + * {@link #getMessage()} method. + */ + public ChronicleHashRecoveryFailedException(String message) { + super(message); + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java b/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java index 5b8d4ff4b..e66cc693f 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/BigSegmentHeader.java @@ -377,4 +377,24 @@ public final class BigSegmentHeader implements SegmentHeader { public void downgradeWriteToReadLock(long address) { LOCK.downgradeWriteToReadLock(A, null, address + LOCK_OFFSET); } + + @Override + public void resetLock(long address) { + LOCK.reset(A, null, address + LOCK_OFFSET); + } + + @Override + public long resetLockState() { + return LOCK.resetState(); + } + + @Override + public long getLockState(long address) { + return LOCK.getState(A, null, address + LOCK_OFFSET); + } + + @Override + public String lockStateToString(long lockState) { + return LOCK.toString(lockState); + } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/CompactOffHeapLinearHashTable.java b/src/main/java/net/openhft/chronicle/hash/impl/CompactOffHeapLinearHashTable.java index 34a1443e8..9e155082d 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/CompactOffHeapLinearHashTable.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/CompactOffHeapLinearHashTable.java @@ -121,7 +121,7 @@ public abstract class CompactOffHeapLinearHashTable { return (entry >>> keyBits) & valueMask; } - long entry(long key, long value) { + public long entry(long key, long value) { return key | (value << keyBits); } @@ -143,9 +143,9 @@ public abstract class CompactOffHeapLinearHashTable { writeEntryVolatile(addr, pos, key(currentEntry), value); } - abstract void writeEntry(long addr, long pos, long newEntry); + public abstract void writeEntry(long addr, long pos, long newEntry); - abstract void clearEntry(long addr, long pos); + public abstract void clearEntry(long addr, long pos); /** * Returns "insert" position in terms of consequent putValue() @@ -176,4 +176,17 @@ public abstract class CompactOffHeapLinearHashTable { clearEntry(addr, posToRemove); return posToRemove; } + + public void insert(long addr, long key, long value) { + long startPos = hlPos(key); + long pos = startPos; + do { + if (empty(readEntry(addr, pos))) { + writeEntry(addr, pos, entry(key, value)); + return; + } + pos = step(pos); + } while (pos != startPos); + throw new AssertionError("HashLookup overflow should never occur"); + } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/IntCompactOffHeapLinearHashTable.java b/src/main/java/net/openhft/chronicle/hash/impl/IntCompactOffHeapLinearHashTable.java index 5bb0e1c68..16188a369 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/IntCompactOffHeapLinearHashTable.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/IntCompactOffHeapLinearHashTable.java @@ -52,12 +52,12 @@ public final class IntCompactOffHeapLinearHashTable extends CompactOffHeapLinear } @Override - void writeEntry(long addr, long pos, long newEntry) { + public void writeEntry(long addr, long pos, long newEntry) { OS.memory().writeInt(addr + pos, (int) newEntry); } @Override - void clearEntry(long addr, long pos) { + public void clearEntry(long addr, long pos) { OS.memory().writeInt(addr + pos, 0); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/LongCompactOffHeapLinearHashTable.java b/src/main/java/net/openhft/chronicle/hash/impl/LongCompactOffHeapLinearHashTable.java index 8b1365fce..d580f49d5 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/LongCompactOffHeapLinearHashTable.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/LongCompactOffHeapLinearHashTable.java @@ -52,12 +52,12 @@ public final class LongCompactOffHeapLinearHashTable extends CompactOffHeapLinea } @Override - void writeEntry(long addr, long pos, long newEntry) { + public void writeEntry(long addr, long pos, long newEntry) { OS.memory().writeLong(addr + pos, newEntry); } @Override - void clearEntry(long addr, long pos) { + public void clearEntry(long addr, long pos) { OS.memory().writeLong(addr + pos, 0L); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/SegmentHeader.java b/src/main/java/net/openhft/chronicle/hash/impl/SegmentHeader.java index 32873becb..bed2c9a01 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/SegmentHeader.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/SegmentHeader.java @@ -59,4 +59,9 @@ public interface SegmentHeader { void writeUnlock(long address); void downgradeWriteToUpdateLock(long address); void downgradeWriteToReadLock(long address); + + void resetLock(long address); + long resetLockState(); + long getLockState(long address); + String lockStateToString(long lockState); } 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 0fc746afa..833a9e1e0 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -16,7 +16,6 @@ package net.openhft.chronicle.hash.impl; -import net.openhft.chronicle.algo.bytes.Access; import net.openhft.chronicle.algo.locks.*; import net.openhft.chronicle.bytes.BytesStore; import net.openhft.chronicle.bytes.MappedBytesStoreFactory; @@ -30,6 +29,7 @@ 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.ChronicleMapBuilder; +import net.openhft.chronicle.map.impl.IterationContext; import net.openhft.chronicle.values.Values; import net.openhft.chronicle.wire.Marshallable; import net.openhft.chronicle.wire.WireIn; @@ -43,12 +43,15 @@ import java.nio.ByteBuffer; import java.nio.channels.FileChannel; import java.util.ArrayList; import java.util.List; +import java.util.Set; import java.util.concurrent.TimeUnit; import static java.lang.Long.numberOfTrailingZeros; import static java.lang.Math.max; import static java.nio.channels.FileChannel.MapMode.READ_WRITE; +import static java.util.stream.Collectors.toSet; import static net.openhft.chronicle.algo.MemoryUnit.*; +import static net.openhft.chronicle.algo.bytes.Access.nativeAccess; import static net.openhft.chronicle.bytes.NativeBytesStore.lazyNativeBytesStoreWithFixedCapacity; import static net.openhft.chronicle.core.OS.pageAlign; import static net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable.*; @@ -120,7 +123,7 @@ public abstract class VanillaChronicleHash 0) { - appendBulkData(0, globalMutableState.getAllocatedExtraTierBulks() - 1, - bs, sizeInBytesWithoutTiers()); - } + initBulks(); } } + private void initBulks() { + if (globalMutableState.getAllocatedExtraTierBulks() > 0) { + appendBulkData(0, globalMutableState.getAllocatedExtraTierBulks() - 1, + bs, sizeInBytesWithoutTiers()); + } + } + + private void initBytesStoreAndHeadersViews(BytesStore bytesStore) { + if (bytesStore.start() != 0) { + throw new AssertionError("bytes store " + bytesStore + " starts from " + + bytesStore.start() + ", 0 expected"); + } + this.bs = bytesStore; + globalMutableState.bytesStore(bs, headerSize + GLOBAL_MUTABLE_STATE_VALUE_OFFSET, + globalMutableState.maxSize()); + + onHeaderCreated(); + } + public final void createMappedStoreAndSegments(File file, RandomAccessFile raf) throws IOException { // TODO this method had been moved -- not clear where @@ -447,6 +459,50 @@ public abstract class VanillaChronicleHash 0) { @@ -695,10 +759,20 @@ public abstract class VanillaChronicleHash implements HashEntry, ChecksumEntry return keyOffset + keySize; } - protected long entryEnd() { + public long entryEnd() { return keyEnd(); } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupSearch.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupSearch.java index 08e7c4465..54081a0ee 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupSearch.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/HashLookupSearch.java @@ -67,9 +67,7 @@ public abstract class HashLookupSearch { return hl().value(entry); } } - throw new IllegalStateException("MultiMap is full, that most likely means you " + - "misconfigured entrySize/chunkSize, and entries tend to take less chunks than " + - "expected"); + throw new IllegalStateException("HashLookup overflow should never occur"); } public void found() { diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/NoChecksumStrategy.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/NoChecksumStrategy.java index 738118c20..cf477de79 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/NoChecksumStrategy.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/NoChecksumStrategy.java @@ -29,6 +29,16 @@ public enum NoChecksumStrategy implements ChecksumStrategy { return true; } + @Override + public int computeChecksum() { + return 0; + } + + @Override + public int storedChecksum() { + return 0; + } + @Override public long extraEntryBytes() { return 0; // no extra bytes to store checksum diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java index e750ba20d..53b104729 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/entry/SegmentStages.java @@ -58,8 +58,8 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface { public abstract boolean segmentIndexInit(); - @Stage("SegmentHeader") long segmentHeaderAddress; - @Stage("SegmentHeader") SegmentHeader segmentHeader = null; + @Stage("SegmentHeader") public long segmentHeaderAddress; + @Stage("SegmentHeader") public SegmentHeader segmentHeader = null; private void initSegmentHeader() { segmentHeaderAddress = hh.h().segmentHeaderAddress(segmentIndex); @@ -82,7 +82,7 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface { } } - long lowestPossiblyFreeChunk() { + public long lowestPossiblyFreeChunk() { if (tier == 0) { return segmentHeader.lowestPossiblyFreeChunk(segmentHeaderAddress); } else { @@ -513,7 +513,7 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface { tier = 0; } - private void initSegmentTier(int tier, long tierIndex) { + public void initSegmentTier(int tier, long tierIndex) { this.tier = tier; this.tierIndex = tierIndex; assert tierIndex > 0; @@ -584,7 +584,7 @@ public abstract class SegmentStages implements SegmentLock, LocksInterface { @Stage("Segment") public final PointerBytesStore segmentBS = new PointerBytesStore(); @Stage("Segment") public final Bytes segmentBytes = new VanillaBytes(segmentBS); - @Stage("Segment") private final ReusableBitSet freeList = new ReusableBitSet( + @Stage("Segment") public final ReusableBitSet freeList = new ReusableBitSet( new SingleThreadedFlatBitSetFrame(LONGS.align(hh.h().actualChunksPerSegmentTier, BITS)), Access.nativeAccess(), null, 0); @Stage("Segment") long entrySpaceOffset = 0; diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java index 6c0daf2e2..f5b73ab79 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/HashSegmentIteration.java @@ -98,7 +98,7 @@ public abstract class HashSegmentIteration> } } - private boolean forEachTierEntryWhile( + public boolean forEachTierEntryWhile( Predicate predicate, int currentTier, long currentTierBaseAddr, long tierIndex) { long leftEntries = tierEntriesForIteration(); 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 new file mode 100644 index 000000000..eb09cb7e4 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/SegmentsRecovery.java @@ -0,0 +1,160 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * 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.impl.stage.iter; + +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.impl.IterationContext; +import net.openhft.sg.StageRef; +import net.openhft.sg.Staged; +import org.slf4j.Logger; + +@Staged +public abstract class SegmentsRecovery implements IterationContext { + + @StageRef VanillaChronicleHashHolder hh; + @StageRef SegmentStages s; + @StageRef TierRecovery tierRecovery; + @StageRef LogHolder lh; + @StageRef HashSegmentIteration iteration; + + @Override + public void recoverSegments() { + Logger log = lh.LOG; + VanillaChronicleHash h = hh.h(); + for (int segmentIndex = 0; segmentIndex < h.actualSegments; segmentIndex++) { + s.initSegmentIndex(segmentIndex); + resetSegmentLock(); + zeroOutFirstSegmentTierCountersArea(); + tierRecovery.recoverTier(segmentIndex); + } + + VanillaGlobalMutableState globalMutableState = h.globalMutableState(); + long storedExtraTiersInUse = globalMutableState.getExtraTiersInUse(); + long allocatedExtraTiers = globalMutableState.getAllocatedExtraTierBulks() * h.tiersInBulk; + long expectedExtraTiersInUse = + Math.max(0, Math.min(storedExtraTiersInUse, allocatedExtraTiers)); + long actualExtraTiersInUse = 0; + long firstFreeExtraTierIndex = -1; + for (long extraTierIndex = 0; extraTierIndex < expectedExtraTiersInUse; extraTierIndex++) { + 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); + 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); + TierCountersArea.segmentIndex(tierCountersAreaAddr, segmentIndex); + } + s.nextTierIndex(0); + + s.initSegmentIndex(segmentIndex); + s.goToLastTier(); + s.nextTierIndex(tierIndex); + + TierCountersArea.prevTierIndex(tierCountersAreaAddr, s.tierIndex); + TierCountersArea.tier(tierCountersAreaAddr, s.tier + 1); + actualExtraTiersInUse = extraTierIndex + 1; + } else { + firstFreeExtraTierIndex = extraTierIndex; + break; + } + } + + if (storedExtraTiersInUse != actualExtraTiersInUse) { + log.error("wrong number of actual tiers in use in global mutable state, stored: {}, " + + "should be: {}", storedExtraTiersInUse, actualExtraTiersInUse); + globalMutableState.setExtraTiersInUse(actualExtraTiersInUse); + } + + long firstFreeTierIndex; + if (firstFreeExtraTierIndex == -1) { + if (allocatedExtraTiers > expectedExtraTiersInUse) { + firstFreeTierIndex = h.extraTierIndexToTierIndex(expectedExtraTiersInUse); + + } else { + firstFreeTierIndex = 0; + } + } else { + firstFreeTierIndex = h.extraTierIndexToTierIndex(firstFreeExtraTierIndex); + } + if (firstFreeTierIndex > 0) { + long lastTierIndex = h.extraTierIndexToTierIndex(allocatedExtraTiers - 1); + h.linkAndZeroOutFreeTiers(firstFreeTierIndex, lastTierIndex); + } + long storedFirstFreeTierIndex = globalMutableState.getFirstFreeTierIndex(); + if (storedFirstFreeTierIndex != firstFreeTierIndex) { + log.error("wrong first free tier index in global mutable state, stored: {}, " + + "should be: {}", storedFirstFreeTierIndex, firstFreeTierIndex); + } + + removeDuplicatesInSegments(); + } + + private void removeDuplicatesInSegments() { + VanillaChronicleHash h = hh.h(); + for (int segmentIndex = 0; segmentIndex < h.actualSegments; segmentIndex++) { + s.initSegmentIndex(segmentIndex); + s.initSegmentTier(); + s.goToLastTier(); + while (true) { + tierRecovery.removeDuplicatesInSegment(); + if (s.tier > 0) { + s.prevTier(); + } else { + break; + } + } + } + } + + private void resetSegmentLock() { + 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)); + s.segmentHeader.resetLock(s.segmentHeaderAddress); + } + } + + private void zeroOutFirstSegmentTierCountersArea() { + 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()); + 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)); + 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)); + } + } +} 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 new file mode 100644 index 000000000..4904316ca --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/iter/TierRecovery.java @@ -0,0 +1,256 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * 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.impl.stage.iter; + +import net.openhft.chronicle.bytes.BytesUtil; +import net.openhft.chronicle.hash.ChronicleHashRecoveryFailedException; +import net.openhft.chronicle.hash.Data; +import net.openhft.chronicle.hash.ExternalHashQueryContext; +import net.openhft.chronicle.hash.HashEntry; +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.ExternalMapQueryContext; +import net.openhft.chronicle.map.MapEntry; +import net.openhft.chronicle.map.VanillaChronicleMap; +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; + +@Staged +public class TierRecovery { + + @StageRef LogHolder lh; + @StageRef VanillaChronicleMapHolder mh; + @StageRef SegmentStages s; + @StageRef MapEntryStages e; + @StageRef IterationKeyHashCode khc; + + public int recoverTier(int segmentIndex) { + s.freeList.clearAll(); + + Logger log = lh.LOG; + VanillaChronicleHash h = mh.h(); + CompactOffHeapLinearHashTable hl = h.hashLookup; + long hlAddr = s.tierBaseAddr; + + long hlPos = 0; + do { + long hlEntry = hl.readEntry(hlAddr, hlPos); + nextHlPos: + if (!hl.empty(hlEntry)) { + hl.clearEntry(hlAddr, hlPos); + + long searchKey = hl.key(hlEntry); + long entryPos = hl.value(hlEntry); + int si = checkEntry(searchKey, entryPos, segmentIndex); + if (si < 0) { + break nextHlPos; + } else { + segmentIndex = si; + } + + // The entry has passed all checks, re-insert: + long startInsertPos = hl.hlPos(searchKey); + long insertPos = startInsertPos; + do { + long hlInsertEntry = hl.readEntry(hlAddr, insertPos); + if (hl.empty(hlInsertEntry)) { + hl.writeEntry(hlAddr, insertPos, hl.entry(searchKey, entryPos)); + break nextHlPos; + } + if (insertPos == hlPos) { + throw new ChronicleHashRecoveryFailedException( + "Concurrent modification of ChronicleMap at " + h.file() + + " while recovery procedure is in progress"); + } + if (hl.key(hlInsertEntry) == searchKey) { + long anotherEntryPos = hl.value(hlInsertEntry); + if (anotherEntryPos == entryPos) + break nextHlPos; + long currentKeyOffset = e.keyOffset; + long currentKeySize = e.keySize; + int currentEntrySizeInChunks = e.entrySizeInChunks; + if (checkEntry(searchKey, anotherEntryPos, segmentIndex) > 0) { + if (e.keySize == currentKeySize && + BytesUtil.bytesEqual(s.segmentBS, currentKeyOffset, + s.segmentBS, e.keyOffset, currentKeySize)) { + log.error("Entries with duplicate keys within a tier: " + + "at pos {} and {} with key {}, first value is {}", + entryPos, anotherEntryPos, e.key(), e.value()); + s.freeList.clearRange( + entryPos, entryPos + currentEntrySizeInChunks); + break nextHlPos; + } + } + } + insertPos = hl.step(insertPos); + } while (insertPos != startInsertPos); + throw new ChronicleHashRecoveryFailedException( + "HashLookup overflow should never occur. " + + "It might also be concurrent access to ChronicleMap at " + + h.file() + " while recovery procedure is in progress"); + } + hlPos = hl.step(hlPos); + } while (hlPos != 0); + return segmentIndex; + } + + public void removeDuplicatesInSegment() { + long startHlPos = 0L; + VanillaChronicleMap m = mh.m(); + CompactOffHeapLinearHashTable hashLookup = m.hashLookup; + long currentTierBaseAddr = s.tierBaseAddr; + while (!hashLookup.empty(hashLookup.readEntry(currentTierBaseAddr, startHlPos))) { + startHlPos = hashLookup.step(startHlPos); + } + long hlPos = startHlPos; + int steps = 0; + long entries = 0; + tierIteration: + do { + hlPos = hashLookup.step(hlPos); + steps++; + long entry = hashLookup.readEntry(currentTierBaseAddr, hlPos); + if (!hashLookup.empty(entry)) { + e.readExistingEntry(hashLookup.value(entry)); + Data key = (Data) e.key(); + try (ExternalMapQueryContext c = m.queryContext(key)) { + MapEntry entry2 = c.entry(); + Data key2 = entry2.key(); + if (key2.bytes().address(key2.offset()) != key.bytes().address(key.offset())) { + lh.LOG.error("entries with duplicate key {} in segment {}: " + + "with values {} and {}, removing the latter", + key, c.segmentIndex(), entry2.value(), e.value()); + if (hashLookup.remove(currentTierBaseAddr, hlPos) != hlPos) { + hlPos = hashLookup.stepBack(hlPos); + steps--; + } + continue tierIteration; + } + } + entries++; + } + // the `steps == 0` condition and this variable updates in the loop fix the bug, when + // shift deletion occurs on the first entry of the tier, and the hlPos + // becomes equal to start pos without making the whole loop, but only visiting a single + // entry + } while (hlPos != startHlPos || steps == 0); + + recoverTierEntriesCounter(entries); + recoverLowestPossibleFreeChunkTiered(); + } + + private void recoverTierEntriesCounter(long entries) { + if (s.tierEntries() != entries) { + lh.LOG.error("Wrong number of entries counter for tier with index {}, " + + "stored: {}, should be: {}", s.tierIndex, s.tierEntries(), entries); + s.tierEntries(entries); + } + } + + private void recoverLowestPossibleFreeChunkTiered() { + 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); + s.lowestPossiblyFreeChunk(lowestFreeChunk); + } + } + + private int checkEntry(long searchKey, long entryPos, int segmentIndex) { + Logger log = lh.LOG; + 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); + } + try { + e.readExistingEntry(entryPos); + } catch (Exception e) { + log.error("Exception while reading entry key size: {}", e); + return -1; + } + if (e.keyEnd() > s.segmentBytes.capacity()) { + log.error("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()); + return -1; + } + + long segmentHashFromKey = h.hashSplitting.segmentHash(keyHashCode); + long searchKeyFromKey = h.hashLookup.maskUnsetKey(segmentHashFromKey); + if (searchKey != searchKeyFromKey) { + log.error("HashLookup searchKey: {}, HashLookup searchKey " + + "from the entry key hash code: {}, entry key: {}", + searchKey, searchKeyFromKey, e.key()); + return -1; + } + + try { + long entryAndChecksumEnd = e.entryEnd() + e.checksumStrategy.extraEntryBytes(); + if (entryAndChecksumEnd > s.segmentBytes.capacity()) { + log.error("Wrong value size: {}, key: {}", e.valueSize, e.key()); + return -1; + } + } catch (Exception ex) { + log.error("Exception while reading entry value size: {}, key: {}", ex, e.key()); + 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 " + + "the entry bytes: {}, key: {}, 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()); + } + s.freeList.setRange(entryPos, entryPos + e.entrySizeInChunks); + + if (segmentIndex < 0) { + 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()); + return -1; + } else { + 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 0775b37db..4d1b8f0b6 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.core.OS; import net.openhft.chronicle.hash.ChronicleHashBuilder; import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI; import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; +import net.openhft.chronicle.hash.ChronicleHashRecoveryFailedException; import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable; import net.openhft.chronicle.hash.impl.stage.entry.ChecksumStrategy; import net.openhft.chronicle.hash.impl.util.math.PoissonDistribution; @@ -47,7 +48,6 @@ import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import static java.lang.Double.isNaN; import static java.lang.Math.round; @@ -267,6 +267,7 @@ public final class ChronicleMapBuilder implements * @deprecated don't use private API in the client code */ @Override + @Deprecated public ChronicleHashBuilderPrivateAPI privateAPI() { return privateAPI; } @@ -1358,8 +1359,7 @@ public final class ChronicleMapBuilder implements @Override public ChronicleHashInstanceBuilder> instance() { - return new MapInstanceBuilder<>(this.clone(), singleHashReplication, null, null, null, - new AtomicBoolean(false)); + return new MapInstanceBuilder<>(this.clone(), singleHashReplication); } @Override @@ -1367,7 +1367,13 @@ 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, singleHashReplication, null); + return clone().createWithFile(file, singleHashReplication, null, false, false); + } + + @Override + public ChronicleMap recoverPersistedTo(File file, boolean sameBuilderConfig) + throws IOException { + return clone().createWithFile(file, singleHashReplication, null, true, sameBuilderConfig); } @Override @@ -1378,9 +1384,12 @@ public final class ChronicleMapBuilder implements return clone().createWithoutFile(singleHashReplication, null); } - ChronicleMap create(MapInstanceBuilder ib) throws IOException { + ChronicleMap create( + MapInstanceBuilder ib, boolean recover, boolean sameBuilderConfig) + throws IOException { if (ib.file != null) { - return createWithFile(ib.file, ib.singleHashReplication, ib.channel); + return createWithFile( + ib.file, ib.singleHashReplication, ib.channel, recover, sameBuilderConfig); } else { return createWithoutFile(ib.singleHashReplication, ib.channel); } @@ -1388,7 +1397,10 @@ public final class ChronicleMapBuilder implements ChronicleMap createWithFile( File file, SingleChronicleHashReplication singleHashReplication, - ReplicationChannel channel) throws IOException { + ReplicationChannel channel, + boolean recover, boolean overrideBuilderConfig) throws IOException { + if (overrideBuilderConfig && !recover) + throw new AssertionError("recover -> overrideBuilderConfig"); replicated = singleHashReplication != null || channel != null; persisted = true; @@ -1397,6 +1409,12 @@ public final class ChronicleMapBuilder implements if (fileLength > 0) { try (RandomAccessFile raf = new RandomAccessFile(file, "rw")) { FileChannel fileChannel = raf.getChannel(); + int headerSize = 0; + if (overrideBuilderConfig) { + VanillaChronicleMap mapObjectForHeaderOverwrite = + newMap(singleHashReplication, channel); + headerSize = writeHeader(fileChannel, mapObjectForHeaderOverwrite); + } ByteBuffer headerBuffer = ByteBuffer.allocate(MAX_BOOTSTRAPPING_HEADER_SIZE); while (headerBuffer.remaining() > 0 && headerBuffer.position() < fileChannel.size()) { @@ -1408,24 +1426,42 @@ public final class ChronicleMapBuilder implements headerBytes.readLimit(headerBuffer.limit()); Wire wire = new BinaryWire(headerBytes); VanillaChronicleMap map = wire.getValueIn().typedMarshallable(); - map.initTransientsFromBuilder(this); - initTransientsFromReplication(map, singleHashReplication, channel); - - map.initBeforeMapping(fileChannel, headerBytes.readPosition()); + long readHeaderSize = headerBytes.readPosition(); + if (overrideBuilderConfig && headerSize != readHeaderSize) { + throw new ChronicleHashRecoveryFailedException("read header size " + + readHeaderSize + " != overwritten header size " + headerSize); + } + map.initBeforeMapping(file, fileChannel, readHeaderSize); long expectedFileLength = map.expectedFileSize(); - if (expectedFileLength != fileLength) { + if (!recover && expectedFileLength != fileLength) { throw new IOException("The file " + file + " the map is serialized from " + "has unexpected length " + fileLength + ", probably corrupted. " + "Expected length is " + expectedFileLength); } - map.createMappedStoreAndSegments(file, raf); + + map.initTransientsFromBuilder(this); + initTransientsFromReplication(map, singleHashReplication, channel); + + if (!recover) { + map.createMappedStoreAndSegments(file, raf); + } else { + map.recover(file, raf); + } + establishReplication(map, singleHashReplication, channel); + fileChannel.force(true); // TODO according to Self Boostrapping Data spec, should write "init complete" // byte *after* the above force instruction, see HCOLL-396 return map; + } catch (Exception e) { + if (recover && !(e instanceof IOException)) + throw new ChronicleHashRecoveryFailedException(e); + throw e; } } + if (recover) + throw new FileNotFoundException("file " + file + " should exist for recovery"); if (file.createNewFile() || fileLength == 0) { break; } @@ -1443,21 +1479,28 @@ public final class ChronicleMapBuilder implements try (RandomAccessFile raf = new RandomAccessFile(file, "rw")) { FileChannel fileChannel = raf.getChannel(); - ByteBuffer headerBuffer = ByteBuffer.allocate(MAX_BOOTSTRAPPING_HEADER_SIZE); - Bytes headerBytes = Bytes.wrapForWrite(headerBuffer); - Wire wire = new BinaryWire(headerBytes); - wire.getValueOut().typedMarshallable(map); + int headerSize = writeHeader(fileChannel, map); + map.initBeforeMapping(file, fileChannel, headerSize); - int headerSize = (int) headerBytes.writePosition(); - headerBuffer.limit(headerSize); - while (headerBuffer.remaining() > 0) { - fileChannel.write(headerBuffer, headerBuffer.position()); - } - map.initBeforeMapping(fileChannel, headerSize); map.createMappedStoreAndSegments(file, raf); + establishReplication(map, singleHashReplication, channel); + return map; } + } - return establishReplication(map, singleHashReplication, channel); + private static int writeHeader(FileChannel fileChannel, VanillaChronicleMap map) + throws IOException { + ByteBuffer headerBuffer = ByteBuffer.allocate(MAX_BOOTSTRAPPING_HEADER_SIZE); + Bytes headerBytes = Bytes.wrapForWrite(headerBuffer); + Wire wire = new BinaryWire(headerBytes); + wire.getValueOut().typedMarshallable(map); + + int headerSize = (int) headerBytes.writePosition(); + headerBuffer.limit(headerSize); + while (headerBuffer.remaining() > 0) { + fileChannel.write(headerBuffer, headerBuffer.position()); + } + return headerSize; } private static void initTransientsFromReplication( @@ -1492,7 +1535,8 @@ public final class ChronicleMapBuilder implements BytesStore bytesStore = lazyNativeBytesStoreWithFixedCapacity(map.sizeInBytesWithoutTiers()); map.createMappedStoreAndSegments(bytesStore); - return establishReplication(map, singleHashReplication, channel); + establishReplication(map, singleHashReplication, channel); + return map; } catch (IOException e) { // file-less version should never trigger an IOException. throw new AssertionError(e); @@ -1552,7 +1596,7 @@ public final class ChronicleMapBuilder implements && actualChunkSize > 0; } - private ChronicleMap establishReplication( + private void establishReplication( VanillaChronicleMap map, SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) throws IOException { @@ -1591,7 +1635,6 @@ public final class ChronicleMapBuilder implements result.addCloseable(token); } } - return map; } private void establishCleanupThread(ReplicatedChronicleMap map) { diff --git a/src/main/java/net/openhft/chronicle/map/MapInstanceBuilder.java b/src/main/java/net/openhft/chronicle/map/MapInstanceBuilder.java index b6e06e2c4..16629d3a5 100644 --- a/src/main/java/net/openhft/chronicle/map/MapInstanceBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/MapInstanceBuilder.java @@ -24,7 +24,6 @@ import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import java.io.File; import java.io.IOException; import java.io.Serializable; -import java.util.concurrent.atomic.AtomicBoolean; final class MapInstanceBuilder implements ChronicleHashInstanceBuilder>, Serializable { @@ -35,18 +34,12 @@ final class MapInstanceBuilder transient File file; String name; - final AtomicBoolean used; + volatile boolean used = false; MapInstanceBuilder(ChronicleMapBuilder mapBuilder, - SingleChronicleHashReplication singleHashReplication, - ReplicationChannel channel, - File file, String name, AtomicBoolean used) { + SingleChronicleHashReplication singleHashReplication) { this.mapBuilder = mapBuilder; this.singleHashReplication = singleHashReplication; - this.channel = channel; - this.file = file; - this.name = name; - this.used = used; } @Override @@ -84,14 +77,35 @@ final class MapInstanceBuilder @Override public synchronized ChronicleMap create() throws IOException { - if (!used.getAndSet(true)) { - return mapBuilder.create(this); + if (!used) { + ChronicleMap map = mapBuilder.create(this, false, false); + // Set used = true, if map created successfully. + // If exception is thrown in mapBuilder.create(), the next line is not executed. + used = true; + return map; } else { - throw new IllegalStateException( - "A ChronicleMap has already been created using this instance config chain. " + - "Create a new instance config (builder.instance()) to create a new " + - "ChronicleMap instance"); + throw alreadyCreated(); } } + @Override + public synchronized ChronicleMap recover(boolean sameBuilderConfig) throws IOException { + if (file == null) + throw new IllegalStateException("persistedTo(file) must be configured for " + + "ChronicleHashInstanceBuilder, for recover()"); + if (!used) { + ChronicleMap map = mapBuilder.create(this, true, sameBuilderConfig); + used = true; + return map; + } else { + throw alreadyCreated(); + } + } + + private static IllegalStateException alreadyCreated() { + return new IllegalStateException( + "A ChronicleHash has already been created using this " + + "ChronicleHashInstanceBuilder. Create a new instance builder " + + "(chronicleHashBuilder.instance()) to create another Chronicle Hash"); + } } diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 17639cecd..fb2f07fa0 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -172,7 +172,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap raise change for all raiseChangeForAllExcept(tierIndex, pos, (byte) -1); @@ -601,7 +610,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap> log2TiersInBulk); + long offsetToTierBitSet = + (extraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; + TierBulkData tierBulkData = tierBulkOffsets.get(bulkIndex); + long bitSetAddr = bitSetsAddr(tierBulkData) + offsetToTierBitSet; + tierModIterFrame.clearRange(nativeAccess(), null, bitSetAddr, pos, endPosExclusive); + } + } } } \ No newline at end of file diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index a1f8aee5a..594a87154 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -36,8 +36,10 @@ import net.openhft.chronicle.wire.WireIn; import net.openhft.chronicle.wire.WireOut; import org.jetbrains.annotations.NotNull; +import java.io.File; import java.io.IOException; import java.io.ObjectInputStream; +import java.io.RandomAccessFile; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.Set; @@ -148,6 +150,11 @@ public class VanillaChronicleMap initOwnTransients(); } + public void recover(File file, RandomAccessFile raf) throws IOException { + basicRecover(file, raf); + iterationContext().recoverSegments(); + } + private void initOwnTransients() { couldNotDetermineAlignmentBeforeAllocation = greatestCommonDivisor((int) chunkSize, alignment) != alignment; 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 340f65720..8feb3f726 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/IterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/IterationContext.java @@ -19,8 +19,13 @@ package net.openhft.chronicle.map.impl; import net.openhft.chronicle.map.MapEntry; import net.openhft.chronicle.map.MapSegmentContext; +import java.util.List; +import java.util.Set; + public interface IterationContext extends MapEntry, MapSegmentContext { long pos(); void initSegmentIndex(int segmentIndex); + + void recoverSegments(); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java b/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java index 26fccc0aa..ac8ce8a94 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java @@ -21,9 +21,7 @@ import net.openhft.chronicle.hash.impl.stage.entry.*; import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop; import net.openhft.chronicle.hash.impl.stage.hash.LogHolder; import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder; -import net.openhft.chronicle.hash.impl.stage.iter.IterationAlloc; -import net.openhft.chronicle.hash.impl.stage.iter.IterationKeyHashCode; -import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages; +import net.openhft.chronicle.hash.impl.stage.iter.*; import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData; import net.openhft.chronicle.map.impl.stage.data.bytes.WrappedValueBytesData; @@ -61,6 +59,9 @@ import net.openhft.sg.Staged; MapEntryOperationsDelegation.class, IterationAlloc.class, + + TierRecovery.class, + SegmentsRecovery.class, }, nested = { ReadLock.class, UpdateLock.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java index 8d6ad5751..053df607f 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java @@ -24,16 +24,14 @@ import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder; import net.openhft.chronicle.hash.impl.stage.iter.IterationAlloc; import net.openhft.chronicle.hash.impl.stage.iter.IterationKeyHashCode; import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages; +import net.openhft.chronicle.hash.impl.stage.iter.SegmentsRecovery; import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData; import net.openhft.chronicle.map.impl.stage.data.bytes.WrappedValueBytesData; import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceDataHolder; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; -import net.openhft.chronicle.map.impl.stage.iter.IterationCheckOnEachPublicOperation; -import net.openhft.chronicle.map.impl.stage.iter.ReplicatedMapAbsentDelegatingForIteration; -import net.openhft.chronicle.map.impl.stage.iter.ReplicatedMapEntryDelegating; -import net.openhft.chronicle.map.impl.stage.iter.ReplicatedMapSegmentIteration; +import net.openhft.chronicle.map.impl.stage.iter.*; import net.openhft.chronicle.map.impl.stage.map.*; import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate; import net.openhft.sg.Context; @@ -68,6 +66,9 @@ import net.openhft.sg.Staged; DefaultValue.class, IterationAlloc.class, + + ReplicatedTierRecovery.class, + SegmentsRecovery.class, }, nested = { ReadLock.class, UpdateLock.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java index 7cb51c055..1ee38eedd 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/MapEntryStages.java @@ -112,7 +112,7 @@ public abstract class MapEntryStages extends HashEntryStages } @Override - protected long entryEnd() { + public long entryEnd() { return valueOffset + valueSize; } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java index 5bd3bcb06..48f981e6f 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java @@ -26,6 +26,11 @@ import net.openhft.sg.StageRef; import net.openhft.sg.Staged; import org.jetbrains.annotations.NotNull; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + @Staged public abstract class MapSegmentIteration extends HashSegmentIteration> implements MapEntry, IterationContext { @@ -48,4 +53,15 @@ public abstract class MapSegmentIteration extends HashSegmentIteration< public WrappedValueInstanceDataHolderAccess context() { return wrappedValueInstanceDataHolderAccess; } +// +// @Override +// public void recover(List> usedTierIndexesBySegment) { +// for (int visitedSegmentIndex = 0; visitedSegmentIndex < usedTierIndexesBySegment.size(); +// visitedSegmentIndex++) { +// Set intersection = new HashSet<>( +// usedTierIndexesBySegment.get(visitedSegmentIndex)); +// intersection.retainAll(usedTierIndexes); +// if (Collections.dis) +// } +// } } 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 new file mode 100644 index 000000000..bd5d531c9 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedTierRecovery.java @@ -0,0 +1,99 @@ +/* + * Copyright (C) 2015 higherfrequencytrading.com + * + * 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.impl.stage.iter; + +import net.openhft.chronicle.algo.bitset.ReusableBitSet; +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.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; + +@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(); + cleanupModificationIterationBits(); + } + + private void recoverTierDeleted() { + VanillaChronicleHash h = rh.h(); + CompactOffHeapLinearHashTable hl = h.hashLookup; + long hlAddr = s.tierBaseAddr; + + long deleted = 0; + long hlPos = 0; + do { + long hlEntry = hl.readEntry(hlAddr, hlPos); + if (!hl.empty(hlEntry)) { + e.readExistingEntry(hl.value(hlEntry)); + if (e.entryDeleted()) { + deleted++; + } + } + 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); + s.tierDeleted(deleted); + } + } + + private void cleanupModificationIterationBits() { + ReplicatedChronicleMap m = rh.m(); + ReplicatedChronicleMap.ModificationIterator[] its = + m.acquireAllModificationIterators(); + ReusableBitSet freeList = s.freeList; + for (long pos = 0;;) { + long nextPos = freeList.nextSetBit(pos); + if (nextPos > pos) { + for (ReplicatedChronicleMap.ModificationIterator it : its) { + it.clearRange0(s.tierIndex, pos, nextPos); + } + } + if (nextPos > 0) { + e.readExistingEntry(nextPos); + if (e.entrySizeInChunks > 1) { + for (ReplicatedChronicleMap.ModificationIterator it : its) { + it.clearRange0(s.tierIndex, nextPos + 1, nextPos + e.entrySizeInChunks); + } + } + pos = nextPos + e.entrySizeInChunks; + } else { + if (pos < m.actualChunksPerSegmentTier) { + for (ReplicatedChronicleMap.ModificationIterator it : its) { + it.clearRange0(s.tierIndex, pos, m.actualChunksPerSegmentTier); + } + } + } + } + } +} diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index 6a3d2dcec..be956da46 100755 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -333,12 +333,21 @@ public final class ChronicleSetBuilder return new SetFromMap<>((VanillaChronicleMap) map); } + @Override + public ChronicleSet recoverPersistedTo(File file, boolean sameBuilderConfig) + throws IOException { + ChronicleMap map = + chronicleMapBuilder.recoverPersistedTo(file, sameBuilderConfig); + return new SetFromMap<>((VanillaChronicleMap) map); + } + /** * @deprecated don't use private API in the client code */ @Deprecated @Override public ChronicleHashBuilderPrivateAPI privateAPI() { + //noinspection deprecation return chronicleMapBuilder.privateAPI(); } } diff --git a/src/main/java/net/openhft/chronicle/set/SetInstanceBuilder.java b/src/main/java/net/openhft/chronicle/set/SetInstanceBuilder.java index a36dfcbe4..6869f86e4 100644 --- a/src/main/java/net/openhft/chronicle/set/SetInstanceBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/SetInstanceBuilder.java @@ -69,17 +69,12 @@ final class SetInstanceBuilder implements ChronicleHashInstanceBuilder create() throws IOException { - try { - return new SetFromMap<>((VanillaChronicleMap) mapBuilder.create()); - } catch (IllegalStateException e) { - if (e.getMessage() != null && e.getMessage().startsWith("A ChronicleMap")) { - throw new IllegalStateException( - "A ChronicleSet has already been created using this instance config. " + - "Create a new instance config (builder.instance()) to create " + - "a new ChronicleSet instance"); - } else { - throw e; - } - } + return new SetFromMap<>((VanillaChronicleMap) mapBuilder.create()); + } + + @Override + public synchronized ChronicleSet recover(boolean sameBuilderConfig) throws IOException { + ChronicleMap map = mapBuilder.recover(sameBuilderConfig); + return new SetFromMap<>((VanillaChronicleMap) map); } }