From adfb833ca5fa9252dd7f61ce5920988f4d395ee3 Mon Sep 17 00:00:00 2001 From: Ivan Rakov Date: Tue, 2 Mar 2021 14:11:24 +0300 Subject: [PATCH] Dirty outdated entries on reconnect without affecting critical activities, closes https://github.com/ChronicleEnterprise/Chronicle-Map-Enterprise/issues/49 --- .../net/openhft/chronicle/map/Replica.java | 4 +- .../chronicle/map/ReplicatedChronicleMap.java | 86 ++++++++++++++++--- 2 files changed, 78 insertions(+), 12 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 256d2a979..511a6aa41 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.Bytes; +import net.openhft.chronicle.core.threads.EventLoop; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.replication.ReplicableEntry; import org.jetbrains.annotations.NotNull; @@ -137,8 +138,9 @@ public interface Replica extends Closeable { * entries to a new remote node as it connects. * * @param fromTimeStamp the timestamp from which all entries should be dirty + * @param eventLoop event loop in which dirtying to be executed */ - void dirtyEntries(long fromTimeStamp); + void dirtyEntries(long fromTimeStamp, @NotNull EventLoop eventLoop); /** * the {@code modificationNotifier} is called when ever there is a change applied to the diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 41542d88c..e1bcc4d8e 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -19,6 +19,11 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.algo.bitset.BitSetFrame; import net.openhft.chronicle.algo.bitset.SingleThreadedFlatBitSetFrame; import net.openhft.chronicle.bytes.Bytes; +import net.openhft.chronicle.core.threads.EventHandler; +import net.openhft.chronicle.core.threads.EventLoop; +import net.openhft.chronicle.core.threads.HandlerPriority; +import net.openhft.chronicle.core.threads.InvalidEventHandlerException; +import net.openhft.chronicle.hash.ChronicleHashClosedException; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.VanillaGlobalMutableState; import net.openhft.chronicle.hash.impl.TierCountersArea; @@ -27,6 +32,7 @@ import net.openhft.chronicle.hash.replication.ReplicableEntry; import net.openhft.chronicle.map.impl.CompiledReplicatedMapIterationContext; import net.openhft.chronicle.map.impl.CompiledReplicatedMapQueryContext; import net.openhft.chronicle.map.replication.MapRemoteOperations; +import net.openhft.chronicle.threads.EventGroup; import net.openhft.chronicle.values.Values; import net.openhft.chronicle.wire.WireIn; import net.openhft.chronicle.wire.WireOut; @@ -42,6 +48,7 @@ import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReferenceArray; +import java.util.function.IntPredicate; import java.util.stream.Stream; import static net.openhft.chronicle.algo.MemoryUnit.*; @@ -961,20 +968,52 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap iterationContext; + + @Override + public @NotNull + HandlerPriority priority() { + return HandlerPriority.CONCURRENT; + } + + public DirtyEntriesHandler(IntPredicate segmentPredicate, long fromTimeStamp) { + this.segmentPredicate = segmentPredicate; + this.fromTimeStamp = fromTimeStamp; + } + + @Override + public boolean action() throws InvalidEventHandlerException { + if (segmentIndex >= actualSegments) + workCompleted(); + + if (!segmentPredicate.test(segmentIndex)) { + segmentIndex++; + + return true; + } - try (CompiledReplicatedMapIterationContext c = iterationContext()) { - // iterate over all the segments and mark bit in the modification iterator - // that correspond to entries with an older timestamp final boolean debugEnabled = LOG.isDebugEnabled(); - for (int segmentIndex = 0; segmentIndex < actualSegments; segmentIndex++) { - c.initSegmentIndex(segmentIndex); - c.forEachSegmentReplicableEntry(e -> { + + try { + throwExceptionIfClosed(); + + if (iterationContext == null) + iterationContext = iterationContext(); + + iterationContext.initSegmentIndex(segmentIndex); + iterationContext.forEachSegmentReplicableEntry(e -> { if (debugEnabled) { LOG.debug("Bootstrap entry: id {}, key {}, value {}", localIdentifier, - c.key(), c.value()); + iterationContext.key(), iterationContext.value()); } // Bizarrely the next line line cause NPE in JDT compiler //assert re.originTimestamp() > 0L; @@ -990,10 +1029,35 @@ public class ReplicatedChronicleMap extends VanillaChronicleMap= fromTimeStamp) { - raiseChange0(c.tierIndex(), c.pos()); + raiseChange0(iterationContext.tierIndex(), iterationContext.pos()); } }); + } catch (ChronicleHashClosedException e) { + workCompleted(); } + + segmentIndex++; + + return true; + } + + private void workCompleted() throws InvalidEventHandlerException { + if (iterationContext != null) + iterationContext.close(); + + throw new InvalidEventHandlerException("removes the handler"); + } + } + + + @Override + public void dirtyEntries(long fromTimeStamp, @NotNull EventLoop eventLoop) { + throwExceptionIfClosed(); + + for (int i = 0; i < EventGroup.CONC_THREADS; i++) { + final int finalI = i; + eventLoop.addHandler(new DirtyEntriesHandler( + segment -> segment % EventGroup.CONC_THREADS == finalI, fromTimeStamp)); } }