From 6ee78ba92639fe8c4a038b65cb6cd98b6cb75566 Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 6 Jan 2016 00:39:36 +0300 Subject: [PATCH] HCOLL-378 #36 Fix bugs in recovery --- .../hash/impl/VanillaChronicleHash.java | 7 +- .../impl/stage/iter/SegmentsRecovery.java | 1 - .../hash/impl/stage/iter/TierRecovery.java | 8 +- .../map/impl/stage/entry/MapEntryStages.java | 4 + .../stage/entry/ReplicatedMapEntryStages.java | 1 + .../stage/iter/ReplicatedTierRecovery.java | 9 +- .../openhft/chronicle/map/RecoverTest.java | 111 ++++++++++++++++++ 7 files changed, 129 insertions(+), 12 deletions(-) create mode 100644 src/test/java/net/openhft/chronicle/map/RecoverTest.java 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 1ff8332d3..ff4ee4059 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -149,7 +149,7 @@ public abstract class VanillaChronicleHash tierBulkOffsets; public transient long headerSize; - transient long segmentHeadersOffset; + public transient long segmentHeadersOffset; transient long segmentsOffset; public transient CompactOffHeapLinearHashTable hashLookup; @@ -416,7 +416,10 @@ public abstract class VanillaChronicleHash c = m.queryContext(key)) { MapEntry entry2 = c.entry(); - Data key2 = entry2.key(); + Data key2 = ((MapEntry) c).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()); + key, c.segmentIndex(), + entry2 != null ? ((MapEntry) c).value() : "", + !e.entryDeleted() ? e.value() : ""); if (hashLookup.remove(currentTierBaseAddr, hlPos) != hlPos) { hlPos = hashLookup.stepBack(hlPos); steps--; 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 1ee38eedd..118c201b2 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 @@ -266,4 +266,8 @@ public abstract class MapEntryStages extends HashEntryStages initEntrySizeInChunks(allocatedChunks.allocatedChunks); } } + + public boolean entryDeleted() { + return false; + } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java index 82ab0b12a..1bd37accd 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/entry/ReplicatedMapEntryStages.java @@ -69,6 +69,7 @@ public abstract class ReplicatedMapEntryStages extends MapEntryStages.ModificationIterator[] its = m.acquireAllModificationIterators(); ReusableBitSet freeList = s.freeList; - for (long pos = 0;;) { + for (long pos = 0; pos < m.actualChunksPerSegmentTier;) { long nextPos = freeList.nextSetBit(pos); if (nextPos > pos) { for (ReplicatedChronicleMap.ModificationIterator it : its) { @@ -88,11 +88,10 @@ public class ReplicatedTierRecovery extends TierRecovery { } pos = nextPos + e.entrySizeInChunks; } else { - if (pos < m.actualChunksPerSegmentTier) { - for (ReplicatedChronicleMap.ModificationIterator it : its) { - it.clearRange0(s.tierIndex, pos, m.actualChunksPerSegmentTier); - } + for (ReplicatedChronicleMap.ModificationIterator it : its) { + it.clearRange0(s.tierIndex, pos, m.actualChunksPerSegmentTier); } + break; } } } diff --git a/src/test/java/net/openhft/chronicle/map/RecoverTest.java b/src/test/java/net/openhft/chronicle/map/RecoverTest.java new file mode 100644 index 000000000..15146f04c --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/RecoverTest.java @@ -0,0 +1,111 @@ +/* + * 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; + +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.MappedByteBuffer; +import java.nio.channels.FileChannel; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import static org.junit.Assert.assertNull; + +public class RecoverTest { + + ReplicatedChronicleMap map; + + @Test(timeout = 60_000) + public void recoverTest() throws IOException, ExecutionException, InterruptedException { + File mapFile = File.createTempFile("recoverTestFile", ".map"); + mapFile.deleteOnExit(); + + map = (ReplicatedChronicleMap) ChronicleMap + .of(Integer.class, Integer.class) + .entries(2) + .actualSegments(1) + .checksumEntries(true) + .replication((byte) 1) + .cleanupRemovedEntries(false) + .createPersistedTo(mapFile); + + map.acquireModificationIterator((byte) 2); + + // acquires read lock successfully + assertNull(map.get(0)); + + ExecutorService executorService = Executors.newSingleThreadExecutor(); + + executorService.submit(() -> { + ExternalMapQueryContext c = map.queryContext(0); + c.writeLock().lock(); + }).get(); + + try { + map.get(0); + throw new AssertionError("Expected dead lock exception"); + } catch (Exception expected) { + // do nothing + } + + map.close(); + + map = (ReplicatedChronicleMap) ChronicleMap + .of(Integer.class, Integer.class) + .entries(2) + .actualSegments(1) + .checksumEntries(true) + .replication((byte) 1) + .cleanupRemovedEntries(false) + .recoverPersistedTo(mapFile, true); + + // acquires read lock successfully + assertNull(map.get(0)); + + map.put(1, 1); + map.put(2, 2); + map.remove(1); + + long segmentHeadersOffset = this.map.segmentHeadersOffset; + map.close(); + + try (RandomAccessFile raf = new RandomAccessFile(mapFile, "rw")) { + FileChannel ch = raf.getChannel(); + MappedByteBuffer mapBB = ch.map(FileChannel.MapMode.READ_WRITE, 0, mapFile.length()); + for (long offset = segmentHeadersOffset; offset < mapFile.length(); + offset += 8) { + for (int bit = 0; bit < 64; bit++) { + System.out.println("flip bit " + bit + " of word at " + offset); + mapBB.putLong((int) offset, mapBB.getLong((int) offset) ^ (1L << bit)); + try (ChronicleMap recovered = ChronicleMap + .of(Integer.class, Integer.class) + .replication((byte) 1) + .cleanupRemovedEntries(false) + .recoverPersistedTo(mapFile, false)) { + recovered.put(1, 1); + recovered.put(2, 2); + recovered.remove(1); + } + } + } + } + } +}