diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 558cfda2a..1638777f7 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -101,6 +101,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap builder) throws IOException { + public ReplicatedChronicleMap(@NotNull final ChronicleMapBuilder builder) throws IOException { super(builder); tierModIterBitSetSizeInBits = computeTierModIterBitSetSizeInBits(); tierModIterBitSetOuterSize = computeTierModIterBitSetOuterSize(); @@ -139,7 +140,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap "tierModIterBitSetSizeInBits").int64(); @@ -153,8 +154,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap "tierModIterBitSetSizeInBits").int64(tierModIterBitSetSizeInBits); @@ -174,14 +174,14 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap builder) { + void initTransientsFromBuilder(@NotNull final ChronicleMapBuilder builder) { super.initTransientsFromBuilder(builder); this.localIdentifier = builder.replicationIdentifier; //noinspection unchecked @@ -223,12 +223,12 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap raise change for all + // -1 is invalid remoteIdentifier => raise change for all changeCount.incrementAndGet(); raiseChangeForAllExcept(tierIndex, pos, (byte) -1); } - public void raiseChangeFor(long tierIndex, long pos, byte remoteIdentifier) { + public void raiseChangeFor(final long tierIndex, + final long pos, + final byte remoteIdentifier) { throwExceptionIfClosed(); - acquireModificationIterator(remoteIdentifier).raiseChange0(tierIndex, pos); + acquireModificationIterator(remoteIdentifier).raiseChange0(tierIndex, pos); } - public void raiseChangeForAllExcept(long tierIndex, long pos, byte remoteIdentifier) { + public void raiseChangeForAllExcept(final long tierIndex, + final long pos, + final byte remoteIdentifier) { throwExceptionIfClosed(); - updateModificationIteratorsArray(); + updateModificationIteratorsArray(); if (tierIndex <= actualSegments) { - long segmentIndex = tierIndex - 1; - long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize; + final long segmentIndex = tierIndex - 1; + final long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize; for (ModificationIterator it : assignedModificationIterators) { if (it.remoteIdentifier != remoteIdentifier) it.raiseChangeInSegment(offsetToTierBitSet, pos); } } else { - long extraTierIndex = tierIndex - 1 - actualSegments; - int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk); - long offsetToTierBitSet = + final long extraTierIndex = tierIndex - 1 - actualSegments; + final int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk); + final long offsetToTierBitSet = (extraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; for (ModificationIterator it : assignedModificationIterators) { if (it.remoteIdentifier != remoteIdentifier) @@ -401,20 +405,20 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap> log2TiersInBulk); - long offsetToTierBitSet = + final long extraTierIndex = tierIndex - 1 - actualSegments; + final int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk); + final long offsetToTierBitSet = (extraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; for (ModificationIterator modificationIterator : assignedModificationIterators) { modificationIterator.dropChangeInTierBulk(bulkIndex, offsetToTierBitSet, pos); @@ -422,30 +426,35 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap> log2TiersInBulk); - long newOffsetToTierBitSet = + final long newExtraTierIndex = newTierIndex - 1 - actualSegments; + final int newBulkIndex = (int) (newExtraTierIndex >> log2TiersInBulk); + final long newOffsetToTierBitSet = (newExtraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; for (ModificationIterator modificationIterator : assignedModificationIterators) { if (modificationIterator.dropChangeInSegment(oldOffsetToTierBitSet, oldPos)) { @@ -455,13 +464,13 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap> log2TiersInBulk); - long oldOffsetToTierBitSet = + final long oldExtraTierIndex = oldTierIndex - 1 - actualSegments; + final int oldBulkIndex = (int) (oldExtraTierIndex >> log2TiersInBulk); + final long oldOffsetToTierBitSet = (oldExtraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; if (newTierIndex <= actualSegments) { - long newSegmentIndex = newTierIndex - 1; - long newOffsetToTierBitSet = newSegmentIndex * tierModIterBitSetOuterSize; + final long newSegmentIndex = newTierIndex - 1; + final long newOffsetToTierBitSet = newSegmentIndex * tierModIterBitSetOuterSize; for (ModificationIterator modificationIterator : assignedModificationIterators) { if (modificationIterator .dropChangeInTierBulk(oldBulkIndex, oldOffsetToTierBitSet, oldPos)) { @@ -469,9 +478,9 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap> log2TiersInBulk); - long newOffsetToTierBitSet = + final long newExtraTierIndex = newTierIndex - 1 - actualSegments; + final int newBulkIndex = (int) (newExtraTierIndex >> log2TiersInBulk); + final long newOffsetToTierBitSet = (newExtraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; for (ModificationIterator modificationIterator : assignedModificationIterators) { if (modificationIterator @@ -484,21 +493,21 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap> log2TiersInBulk); - long offsetToTierBitSet = + final long extraTierIndex = tierIndex - 1 - actualSegments; + final int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk); + final long offsetToTierBitSet = (extraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; for (ModificationIterator modificationIterator : assignedModificationIterators) { if (modificationIterator.isChangedTierBulk(bulkIndex, offsetToTierBitSet, pos)) @@ -509,24 +518,27 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap keys) { + public void writeExternalEntry(final ReplicableEntry entry, + final Bytes payload, + @NotNull final Bytes destination, + final int chronicleId, + final ArrayList keys) { throwExceptionIfClosed(); - if (payload != null) + if (payload != null) writePayload(payload, destination); if (entry != null) writeExternalEntry0(entry, destination, keys); } - private void writePayload(Bytes payload, Bytes destination) { + private void writePayload(final Bytes payload, final Bytes destination) { destination.writeByte(BOOTSTRAP_TIME_HUNK); destination.write(payload, payload.readPosition(), payload.readRemaining()); } @@ -535,7 +547,9 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap keys) { + private void writeExternalEntry0(final ReplicableEntry entry, + final Bytes destination, + final ArrayList keys) { destination.writeByte(ENTRY_HUNK); destination.writeStopBit(entry.originTimestamp()); @@ -554,9 +568,9 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap" ); + keys.add(key.get().toString()); + } catch (Exception e) { + keys.add(""); } destination.writeBoolean(isDeleted); @@ -565,7 +579,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap - * Once a change occurs to a map, map replication requires that these changes are picked up - * by another thread, this class provides an iterator like interface to poll for such changes. + * Once a change occurs to a map, map replication requires that these changes are picked up + * by another thread, this class provides an iterator like interface to poll for such changes. *

*

- * In most cases the thread that adds data to the node is unlikely to be the same thread - * that replicates the data over to the other nodes, so data will have to be marshaled between - * the main thread storing data to the map, and the thread running the replication. + * In most cases the thread that adds data to the node is unlikely to be the same thread + * that replicates the data over to the other nodes, so data will have to be marshaled between + * the main thread storing data to the map, and the thread running the replication. *

*

- * One way to perform this marshalling, would be to pipe the data into a queue. However, This class - * takes another approach. It uses a bit set, and marks bits which correspond to the indexes of - * the entries that have changed. It then provides an iterator like interface to poll for such - * changes. + * One way to perform this marshalling, would be to pipe the data into a queue. However, This class + * takes another approach. It uses a bit set, and marks bits which correspond to the indexes of + * the entries that have changed. It then provides an iterator like interface to poll for such + * changes. *

* * @author Rob Austin. */ - public class ModificationIterator implements Replica.ModificationIterator { + public final class ModificationIterator implements Replica.ModificationIterator { private final byte remoteIdentifier; private final long segmentBitSetsAddr; private final long offsetToBitSetsWithinATierBulk; @@ -698,7 +712,7 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap c = iterationContext()) { c.initSegmentIndex(segmentIndex); c.updateLock().lock(); @@ -867,11 +888,11 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap extends VanillaChronicleMap c = iterationContext()) { + try (CompiledReplicatedMapIterationContext c = iterationContext()) { // iterate over all the segments and mark bit in the modification iterator // that correspond to entries with an older timestamp - boolean debugEnabled = LOG.isDebugEnabled(); + final boolean debugEnabled = LOG.isDebugEnabled(); for (int segmentIndex = 0; segmentIndex < actualSegments; segmentIndex++) { c.initSegmentIndex(segmentIndex); c.forEachSegmentReplicableEntry(e -> { @@ -954,49 +975,51 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap> log2TiersInBulk); - long offsetToTierBitSet = + final long extraTierIndex = tierIndex - 1 - actualSegments; + final int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk); + final long offsetToTierBitSet = (extraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; raiseChangeInTierBulk(bulkIndex, offsetToTierBitSet, pos); } } - void dropChange0(long tierIndex, long pos) { + void dropChange0(final long tierIndex, final long pos) { if (tierIndex <= actualSegments) { - long segmentIndex = tierIndex - 1; - long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize; + final long segmentIndex = tierIndex - 1; + final long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize; dropChangeInSegment(offsetToTierBitSet, pos); } else { - long extraTierIndex = tierIndex - 1 - actualSegments; - int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk); - long offsetToTierBitSet = + final long extraTierIndex = tierIndex - 1 - actualSegments; + final int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk); + final long offsetToTierBitSet = (extraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; dropChangeInTierBulk(bulkIndex, offsetToTierBitSet, pos); } } - public void clearRange0(long tierIndex, long pos, long endPosExclusive) { + public void clearRange0(final long tierIndex, + final long pos, + final long endPosExclusive) { throwExceptionIfClosed(); - if (tierIndex <= actualSegments) { - long segmentIndex = tierIndex - 1; - long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize; + if (tierIndex <= actualSegments) { + final long segmentIndex = tierIndex - 1; + final long offsetToTierBitSet = segmentIndex * tierModIterBitSetOuterSize; tierModIterFrame.clearRange(nativeAccess(), null, segmentBitSetsAddr + offsetToTierBitSet, pos, endPosExclusive); } else { - long extraTierIndex = tierIndex - 1 - actualSegments; - int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk); - long offsetToTierBitSet = + final long extraTierIndex = tierIndex - 1 - actualSegments; + final int bulkIndex = (int) (extraTierIndex >> log2TiersInBulk); + final long offsetToTierBitSet = (extraTierIndex & (tiersInBulk - 1)) * tierModIterBitSetOuterSize; - TierBulkData tierBulkData = tierBulkOffsets.get(bulkIndex); - long bitSetAddr = bitSetsAddr(tierBulkData) + offsetToTierBitSet; + final TierBulkData tierBulkData = tierBulkOffsets.get(bulkIndex); + final long bitSetAddr = bitSetsAddr(tierBulkData) + offsetToTierBitSet; tierModIterFrame.clearRange(nativeAccess(), null, bitSetAddr, pos, endPosExclusive); } }