From 904ecca036298e024aef6050f0a5afa16e10e163 Mon Sep 17 00:00:00 2001 From: leventov Date: Sat, 22 Aug 2015 23:39:53 +0300 Subject: [PATCH] HCOLL-343 HCOLL-227 Remove old entries from ReplicatedChronicleMap --- .../openhft/chronicle/hash/ChronicleHash.java | 23 +++- .../chronicle/hash/ChronicleHashBuilder.java | 21 ++++ .../hash/ReplicatedHashSegmentContext.java | 30 +++++ .../hash/impl/VanillaChronicleHash.java | 17 ++- .../hash/impl/stage/entry/SegmentStages.java | 34 ++++-- .../hash/CheckOnEachPublicOperation.java | 4 + .../impl/stage/iter/HashSegmentIteration.java | 26 ++-- .../hash/impl/stage/query/HashQuery.java | 2 +- .../ReplicableEntryDelegating.java | 59 ++++++++++ .../hash/replication/ReplicableEntry.java | 20 ++++ .../hash/replication/TimeProvider.java | 24 ++++ .../chronicle/map/AbstractChronicleMap.java | 4 +- .../chronicle/map/ChronicleMapBuilder.java | 69 ++++++++++- .../chronicle/map/ChronicleMapIterator.java | 2 +- .../map/OldDeletedEntriesCleanup.java | 111 ++++++++++++++++++ .../chronicle/map/ReplicatedChronicleMap.java | 59 ++++++++-- ...licatedChronicleMapGlobalMutableState.java | 30 +++++ .../chronicle/map/StatelessChronicleMap.java | 10 ++ .../chronicle/map/VanillaChronicleMap.java | 5 - .../chronicle/map/impl/MapQueryContext.java | 6 +- .../ReplicatedIterationContextInterface.java | 6 - .../impl/ReplicatedMapIterationContext.java | 12 +- .../map/impl/ReplicatedMapQueryContext.java | 8 +- .../stage/entry/ReplicatedMapEntryStages.java | 17 ++- ...icatedMapAbsentDelegatingForIteration.java | 63 ++++++++++ .../iter/ReplicatedMapEntryDelegating.java | 68 +++++++++++ .../iter/ReplicatedMapSegmentIteration.java | 66 +++++++++-- .../map/impl/stage/map/DefaultValue.java | 45 +++++++ .../map/impl/stage/query/MapAbsent.java | 15 +-- .../impl/stage/query/ReplicatedMapAbsent.java | 2 +- .../impl/stage/query/ReplicatedMapQuery.java | 13 +- .../query/ReplicatedQuerySegmentStages.java | 71 +++++++++++ .../stage/replication/ReplicationUpdate.java | 4 + .../chronicle/set/ChronicleSetBuilder.java | 8 ++ .../net/openhft/chronicle/set/SetFromMap.java | 10 ++ .../chronicle/map/ReplicationCheckingMap.java | 10 ++ .../map/TCPSocketReplication4WayMapTest.java | 1 + .../chronicle/map/TcpReplicationSoakTest.java | 40 ++++++- .../map/TcpTimeBasedReplicationSoakTest.java | 10 ++ .../jsr166/map/StatelessChronicleMapTest.java | 10 ++ 40 files changed, 945 insertions(+), 90 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/hash/ReplicatedHashSegmentContext.java create mode 100644 src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java create mode 100644 src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java create mode 100644 src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMapGlobalMutableState.java create mode 100644 src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapAbsentDelegatingForIteration.java create mode 100644 src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapEntryDelegating.java create mode 100644 src/main/java/net/openhft/chronicle/map/impl/stage/map/DefaultValue.java create mode 100644 src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java index 7dfb7347f..d45a569fb 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java @@ -80,8 +80,22 @@ public interface ChronicleHash, SC extends HashSegment */ @NotNull EQC queryContext(Data key); + /** + * Returns the context of the segment with the given index. Segments are indexed from 0 to + * {@link #segments()}{@code - 1}. + * + * @see HashSegmentContext + */ SC segmentContext(int segmentIndex); - + + /** + * Returns the number of segments in this {@code ChronicleHash}. + * + * @see ChronicleHashBuilder#minSegments(int) + * @see ChronicleHashBuilder#actualSegments(int) + */ + int segments(); + /** * Checks the given predicate on each entry in this {@code ChronicleHash} until all entries * have been processed or the predicate returns {@code false} for some entry, or throws @@ -133,4 +147,11 @@ public interface ChronicleHash, SC extends HashSegment */ @Override void close(); + + /** + * Tells whether or not this {@code ChronicleHash} (on-heap instance) is open. + * + * @return {@code true} is {@link #close()} is not yet called + */ + boolean isOpen(); } diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java index fc06ae828..0467f423f 100755 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java @@ -305,6 +305,27 @@ public interface ChronicleHashBuilder, */ B timeProvider(TimeProvider timeProvider); + /** + * Configures timeout after which entries, marked as removed in the Chronicle Hash, constructed + * by this builder, are allowed to be completely removed from the data structure. In replicated + * Chronicle nodes, when {@code remove()} on the key is called, the corresponding entry + * is not immediately erased from the data structure, to let the distributed system eventually + * converge on some value for this key (or converge on the fact, that this key is removed). + * Chronicle Hash watch in runtime after the entries, and if one is removed and not updated + * in anyway for this {@code removedEntryCleanupTimeout}, Chronicle is allowed to remove this + * entry completely from the data structure. This timeout should depend on your distributed + * system topology, and typical replication latencies, that should be determined experimentally. + * + *

Default timeout is 1 second. + * + * @param removedEntryCleanupTimeout timeout, after which stale removed entries could be erased + * from Chronicle Hash data structure completely + * @param unit time unit, in which the timeout is given + * @return this builder back + * @throws IllegalArgumentException is the specified timeout is less than 1 millisecond + */ + B removedEntryCleanupTimeout(long removedEntryCleanupTimeout, TimeUnit unit); + /** * Configures a {@link BytesMarshallerFactory} to be used with {@link * BytesMarshallableSerializer}, which is a default {@link #objectSerializer ObjectSerializer}, diff --git a/src/main/java/net/openhft/chronicle/hash/ReplicatedHashSegmentContext.java b/src/main/java/net/openhft/chronicle/hash/ReplicatedHashSegmentContext.java new file mode 100644 index 000000000..5b3383eda --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/ReplicatedHashSegmentContext.java @@ -0,0 +1,30 @@ +/* + * 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 net.openhft.chronicle.hash.replication.ReplicableEntry; + +import java.util.function.Consumer; +import java.util.function.Predicate; + +public interface ReplicatedHashSegmentContext> + extends HashSegmentContext { + + void forEachSegmentReplicableEntry(Consumer action); + + boolean forEachSegmentReplicableEntryWhile(Predicate predicate); +} 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 f3dac08ed..2eae066fb 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java @@ -114,6 +114,8 @@ public abstract class VanillaChronicleHash builder, boolean replicated) { // Version @@ -262,13 +264,21 @@ public abstract class VanillaChronicleHash Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) size; } + + @Override + public int segments() { + return actualSegments; + } } 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 1cc549e63..06977fd5b 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 @@ -373,8 +373,7 @@ public abstract class SegmentStages implements SegmentLock { entrySpaceOffset = 0; } - //TODO refactor/optimize - public long alloc(int chunks) { + public long allocReturnCode(int chunks) { VanillaChronicleHash h = hh.h(); if (chunks > h.maxChunksPerEntry) throw new IllegalArgumentException("Entry is too large: requires " + chunks + @@ -382,23 +381,17 @@ public abstract class SegmentStages implements SegmentLock { long ret = freeList.setNextNContinuousClearBits(nextPosToSearchFrom(), chunks); if (ret == DirectBitSet.NOT_FOUND || ret + chunks > h.actualChunksPerSegment) { if (ret != DirectBitSet.NOT_FOUND && - ret + chunks > h.actualChunksPerSegment && ret < h.actualChunksPerSegment) + ret + chunks > h.actualChunksPerSegment && ret < h.actualChunksPerSegment) { freeList.clear(ret, h.actualChunksPerSegment); + } ret = freeList.setNextNContinuousClearBits(0L, chunks); if (ret == DirectBitSet.NOT_FOUND || ret + chunks > h.actualChunksPerSegment) { if (ret != DirectBitSet.NOT_FOUND && ret + chunks > h.actualChunksPerSegment && - ret < h.actualChunksPerSegment) + ret < h.actualChunksPerSegment) { freeList.clear(ret, h.actualChunksPerSegment); - if (chunks == 1) { - throw new IllegalStateException( - "Segment is full, no free entries found"); - } else { - throw new IllegalStateException( - "Segment is full or has no ranges of " + chunks - + " continuous free chunks" - ); } + return -1; } updateNextPosToSearchFrom(ret, chunks); } else { @@ -412,6 +405,23 @@ public abstract class SegmentStages implements SegmentLock { return ret; } + public long alloc(int chunks) { + long ret = allocReturnCode(chunks); + if (ret >= 0) { + return ret; + } else { + if (chunks == 1) { + throw new IllegalStateException( + "Segment is full, no free entries found"); + } else { + throw new IllegalStateException( + "Segment is full or has no ranges of " + chunks + + " continuous free chunks" + ); + } + } + } + public void free(long fromPos, int chunks) { freeList.clear(fromPos, fromPos + chunks); if (fromPos < nextPosToSearchFrom()) diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java index e5bd9720b..d9eb56806 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/hash/CheckOnEachPublicOperation.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.hash.impl.stage.hash; +import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -23,8 +24,11 @@ import net.openhft.sg.Staged; public abstract class CheckOnEachPublicOperation { @StageRef OwnerThreadHolder holder; + @StageRef VanillaChronicleHashHolder hh; public void checkOnEachPublicOperation() { holder.checkAccessingFromOwnerThread(); + if (!hh.h().isOpen()) + throw new IllegalStateException("Access to Chronicle Hash after close()"); } } 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 8e2316f8d..f6dd85e6d 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 @@ -17,6 +17,7 @@ package net.openhft.chronicle.hash.impl.stage.iter; import net.openhft.chronicle.hash.HashEntry; +import net.openhft.chronicle.hash.HashSegmentContext; import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable; import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages; @@ -30,7 +31,8 @@ import java.util.function.Consumer; import java.util.function.Predicate; @Staged -public abstract class HashSegmentIteration> implements HashEntry { +public abstract class HashSegmentIteration> + implements HashEntry, HashSegmentContext { @StageRef public SegmentStages s; @StageRef HashEntryStages e; @@ -50,6 +52,7 @@ public abstract class HashSegmentIteration> implements this.entryRemovedOnThisIteration = entryRemovedOnThisIteration; } + @Override public boolean forEachSegmentEntryWhile(Predicate action) { s.innerUpdateLock.lock(); try { @@ -88,6 +91,7 @@ public abstract class HashSegmentIteration> implements } } + @Override public void forEachSegmentEntry(Consumer action) { forEachSegmentEntryWhile(e -> { action.accept(e); @@ -103,18 +107,22 @@ public abstract class HashSegmentIteration> implements @Override public void doRemove() { checkOnEachPublicOperation.checkOnEachPublicOperation(); - initEntryRemovedOnThisIteration(true); s.innerWriteLock.lock(); try { - // this condition mean -- some other entry taken place of the removed one - if (hh.h().hashLookup.remove(s.segmentBase, hlp.hashLookupPos) != hlp.hashLookupPos) { - // if so, should make step back, to compensate step forward on the next iteration, - // to consume the shifted entry - hlp.setHashLookupPos(hh.h().hashLookup.stepBack(hlp.hashLookupPos)); - } - e.innerRemoveEntryExceptHashLookupUpdate(); + iterationRemove(); } finally { s.innerWriteLock.unlock(); } + initEntryRemovedOnThisIteration(true); + } + + public void iterationRemove() { + // this condition mean -- some other entry taken place of the removed one + if (hh.h().hashLookup.remove(s.segmentBase, hlp.hashLookupPos) != hlp.hashLookupPos) { + // if so, should make step back, to compensate step forward on the next iteration, + // to consume the shifted entry + hlp.setHashLookupPos(hh.h().hashLookup.stepBack(hlp.hashLookupPos)); + } + e.innerRemoveEntryExceptHashLookupUpdate(); } } diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java index ea213830a..49746607b 100644 --- a/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java @@ -80,7 +80,7 @@ public abstract class HashQuery implements HashEntry { public void doRemove() { checkOnEachPublicOperation.checkOnEachPublicOperation(); s.innerUpdateLock.lock(); - if (entryPresent()) { + if (ks.searchStatePresent()) { // TODO optimize: if shift-deletion is trivial, updateLock.lock() s.innerWriteLock.lock(); hashLookupSearch.remove(); diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java new file mode 100644 index 000000000..c855c05e5 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/replication/ReplicableEntryDelegating.java @@ -0,0 +1,59 @@ +/* + * 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.replication; + +import net.openhft.chronicle.hash.replication.ReplicableEntry; + +public interface ReplicableEntryDelegating extends ReplicableEntry { + + ReplicableEntry d(); + + @Override + default byte originIdentifier() { + return d().originIdentifier(); + } + + @Override + default long originTimestamp() { + return d().originTimestamp(); + } + + @Override + default void updateOrigin(byte newIdentifier, long newTimestamp) { + d().updateOrigin(newIdentifier, newTimestamp); + } + + @Override + default void dropChanged() { + d().dropChanged(); + } + + @Override + default void raiseChanged() { + d().raiseChanged(); + } + + @Override + default boolean isChanged() { + return d().isChanged(); + } + + @Override + default void doRemoveCompletely() { + d().doRemoveCompletely(); + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java index 8bf62ff38..aab386260 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java @@ -86,4 +86,24 @@ public interface ReplicableEntry { * Propagate the entry, schedule it to be replicated over to remote Chronicle nodes. */ void raiseChanged(); + + /** + * Check is the entry is scheduled to be replicated to the remote Chronicle nodes, to which + * the connection is currently established. + * + * @return {@code true} is the entry is "dirty" locally, i. e. should be replicated to remote + * Chronicle nodes, {@code false} otherwise + */ + boolean isChanged(); + + /** + * Completely remove the entry from the Chronicle hash, not just mark is as deleted, what + * {@link MapEntry#doRemove()} does for Replicated Chronicle hash. After calling this method + * the entry won't be replicated over to remote Chronicle nodes, as if {@link #dropChanged()} + * was called, and incoming entry replication event with the key if this entry will be + * processed, as the entry have never been present in the Chronicle hash. After calling this + * method any subsequent call of any method of {@code ReplicableEntry} class throws {@code + * IllegalStateException}. + */ + void doRemoveCompletely(); } diff --git a/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java b/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java index eb90f77de..0e23d014b 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/TimeProvider.java @@ -22,6 +22,7 @@ import net.openhft.chronicle.set.ChronicleSet; import java.io.ObjectStreamException; import java.io.Serializable; +import java.util.concurrent.TimeUnit; /** * TimeProvider was aims to possibly later provide an optimization to @@ -44,14 +45,37 @@ public abstract class TimeProvider implements Serializable { public abstract long currentTime(); + /** + * {@code timeProvider.scale(System.currentTimeMillis(), TimeUnit.MILLISECONDS)} should + * result to {@link #currentTime()}. + */ + public abstract long scale(long time, TimeUnit unit); + + /** + * {@code timeProvider.unscale(timeProvider.currentTime(), TimeUnit.NANOSECONDS)} should result + * to something close to {@code System.nanoTime()}. + */ + public abstract long unscale(long time, TimeUnit toUnit); + private static class System extends TimeProvider { private static final long serialVersionUID = 1L; private static final long SCALE = 1000; + @Override public long currentTime() { return java.lang.System.currentTimeMillis() * SCALE; } + @Override + public long scale(long time, TimeUnit unit) { + return TimeUnit.MILLISECONDS.convert(time, unit) * SCALE; + } + + @Override + public long unscale(long time, TimeUnit toUnit) { + return toUnit.convert(time, TimeUnit.MILLISECONDS) / SCALE; + } + @Override public boolean equals(Object o) { return o != null && o.getClass() == getClass(); diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java index 5b1981d77..636d9f5fe 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMap.java @@ -76,8 +76,6 @@ interface AbstractChronicleMap extends ChronicleMap, Serializable { } } - int actualSegments(); - @Override default boolean containsValue(Object value) { return !forEachEntryWhile(c -> !c.value().equals(c.context().wrapValueAsData((V) value))); @@ -257,7 +255,7 @@ interface AbstractChronicleMap extends ChronicleMap, Serializable { @Override default boolean forEachEntryWhile(final Predicate> action) { boolean interrupt = false; - for (int i = actualSegments() - 1; i >= 0; i--) { + for (int i = segments() - 1; i >= 0; i--) { try (MapSegmentContext c = segmentContext(i)) { if (!c.forEachSegmentEntryWhile(action)) { interrupt = true; diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 29a3efdd8..4cd219c55 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -37,6 +37,7 @@ import net.openhft.lang.io.serialization.impl.NewInstanceObjectFactory; import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory; import net.openhft.lang.model.Byteable; import net.openhft.lang.model.DataValueClasses; +import net.openhft.lang.thread.NamedThreadFactory; import net.openhft.lang.threadlocal.Provider; import net.openhft.lang.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; @@ -48,8 +49,11 @@ import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.net.InetSocketAddress; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.Objects; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -159,6 +163,9 @@ public final class ChronicleMapBuilder implements // replication private TimeProvider timeProvider = TimeProvider.SYSTEM; + long cleanupTimeout = 1; + TimeUnit cleanupTimeoutUnit = TimeUnit.SECONDS; + private BytesMarshallerFactory bytesMarshallerFactory; private ObjectSerializer objectSerializer; private V defaultValue = null; @@ -1056,6 +1063,18 @@ public final class ChronicleMapBuilder implements return this; } + @Override + public ChronicleMapBuilder removedEntryCleanupTimeout( + long removedEntryCleanupTimeout, TimeUnit unit) { + if (unit.toMillis(removedEntryCleanupTimeout) < 1) { + throw new IllegalArgumentException("timeout should be >= 1 millisecond, " + + removedEntryCleanupTimeout + " " + unit + " is given"); + } + cleanupTimeout = removedEntryCleanupTimeout; + cleanupTimeoutUnit = unit; + return this; + } + TimeProvider timeProvider() { return timeProvider; } @@ -1548,10 +1567,12 @@ public final class ChronicleMapBuilder implements SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) throws IOException { if (map instanceof ReplicatedChronicleMap) { - if (singleHashReplication != null && channel != null) { + if (singleHashReplication != null && channel != null) throw new AssertionError("Only one non-null replication should be passed"); - } + ReplicatedChronicleMap result = (ReplicatedChronicleMap) map; + establishCleanupThread(result); + List replicators = new ArrayList<>(2); if (singleHashReplication != null) { if (singleHashReplication.tcpTransportAndNetwork() != null) @@ -1577,6 +1598,50 @@ public final class ChronicleMapBuilder implements return map; } + private void establishCleanupThread(ReplicatedChronicleMap map) { + map.globalStateLock(); + try { + if (!map.globalMutableState.isCurrentlyCleanupIterated()) { + OldDeletedEntriesCleanup cleanup = new OldDeletedEntriesCleanup(map, + timeProvider.scale(cleanupTimeout, cleanupTimeoutUnit)); + NamedThreadFactory threadFactory = + new NamedThreadFactory("cleanup thread for map persisted at " + map.file()); + ExecutorService executor = Executors.newSingleThreadExecutor(threadFactory); + executor.submit(cleanup); + + map.addCloseable(cleanup); + // WARNING this relies on the fact that ReplicatedChronicleMap closes in the same + // order, i. e. OldDeletedEntriesCleanup instance first + map.addCloseable(() -> { + executor.shutdown(); + try { + executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + LOG.error("", e); + } + }); + map.addCloseable(() -> { + map.globalStateLock(); + try { + map.globalMutableState.setCurrentlyCleanupIterated(false); + } finally { + map.globalStateUnlock(); + } + }); + + map.globalStateLock(); + try { + if (!map.globalMutableState.isCurrentlyCleanupIterated()) + map.globalMutableState.setCurrentlyCleanupIterated(true); + } finally { + map.globalStateUnlock(); + } + } + } finally { + map.globalStateUnlock(); + } + } + private long bufferSize(SerializationBuilder builder, double averageSize, boolean replicated) { if (builder.constantSizeMarshaller()) return round(ceil(averageSize)); diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapIterator.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapIterator.java index 11d07b527..fb799f58f 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapIterator.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapIterator.java @@ -36,7 +36,7 @@ abstract class ChronicleMapIterator implements Iterator, Consumer map) { this.map = map; - segmentIndex = map.actualSegments() - 1; + segmentIndex = map.segments() - 1; } private void checkSingleThreaded() { diff --git a/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java b/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.java new file mode 100644 index 000000000..44c324d0a --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/OldDeletedEntriesCleanup.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 net.openhft.chronicle.core.io.Closeable; +import net.openhft.chronicle.hash.ReplicatedHashSegmentContext; +import net.openhft.chronicle.hash.replication.ReplicableEntry; +import net.openhft.chronicle.hash.replication.TimeProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; + +public class OldDeletedEntriesCleanup implements Runnable, Closeable, Predicate { + private static final Logger LOG = LoggerFactory.getLogger(OldDeletedEntriesCleanup.class); + + private static final int PRIME_1 = 1_000_003, PRIME_2 = 999_983; + + private final ReplicatedChronicleMap map; + private final long timeout; + private final int step; + private volatile boolean shutdown; + private long prevSegment0ScanStart = -1; + private long removedCompletely; + + public OldDeletedEntriesCleanup(ReplicatedChronicleMap map, long timeout) { + this.map = map; + this.timeout = timeout; + // step primality guarantees perfect segment permutation + step = map.segments() != PRIME_1 ? PRIME_1 : PRIME_2; + } + + @Override + public void run() { + while (!shutdown) { + int segmentIndex = map.globalMutableState.getCurrentCleanupSegmentIndex(); + TimeProvider timeProvider = map.timeProvider; + if (segmentIndex == 0 && prevSegment0ScanStart >= 0) { + long currentTime = timeProvider.currentTime(); + long mapScanTime = currentTime - prevSegment0ScanStart; + LOG.debug("Old deleted entries scan time: {}", mapScanTime); + if (mapScanTime < timeout) { + long timeToSleep = + timeProvider.unscale(timeout - mapScanTime, TimeUnit.MILLISECONDS); + if (timeToSleep > 0) { + try { + Thread.sleep(timeToSleep); + } catch (InterruptedException e) { + throw new RuntimeException(e); + + } + } else { + timeToSleep = timeProvider.unscale(timeout - mapScanTime, + TimeUnit.NANOSECONDS); + try { + Thread.sleep(0, (int) timeToSleep); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + } + } + try (MapSegmentContext context = map.segmentContext(segmentIndex)) { + if (segmentIndex == 0) + prevSegment0ScanStart = timeProvider.currentTime(); + removedCompletely = 0; + if (((ReplicatedHashSegmentContext) context) + .forEachSegmentReplicableEntryWhile(this)) { + segmentIndex = (segmentIndex + step) % map.segments(); + map.globalMutableState.setCurrentCleanupSegmentIndex(segmentIndex); + } + } + LOG.debug("Removed {} old deleted entries in the segment {}", + removedCompletely, segmentIndex); + } + } + + @Override + public boolean test(ReplicableEntry e) { + if (shutdown) + return false; + if (e instanceof MapAbsentEntry) { + long deleteTimeout = map.timeProvider.currentTime() - e.originTimestamp(); + if (deleteTimeout > timeout && !e.isChanged()) { + e.doRemoveCompletely(); + removedCompletely++; + } + } + return true; + } + + @Override + public void close() { + shutdown = true; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index 699333dc5..9afb5c0f2 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.hash.ReplicatedHashSegmentContext; import net.openhft.chronicle.hash.replication.AbstractReplication; import net.openhft.chronicle.hash.replication.ReplicableEntry; import net.openhft.chronicle.hash.replication.TimeProvider; @@ -28,6 +29,7 @@ import net.openhft.lang.Maths; import net.openhft.lang.collection.ATSDirectBitSet; import net.openhft.lang.collection.SingleThreadedDirectBitSet; import net.openhft.lang.io.Bytes; +import net.openhft.lang.model.DataValueClasses; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -93,16 +95,21 @@ public class ReplicatedChronicleMap closeables; private transient Bytes identifierUpdatedBytes; + transient Bytes globalMutableStateBytes; + transient ReplicatedChronicleMapGlobalMutableState globalMutableState; private transient ATSDirectBitSet modIterSet; private transient AtomicReferenceArray modificationIterators; private transient long startOfModificationIterators; private boolean bootstrapOnlyLocalEntries; + + public long cleanupTimeout; public transient MapRemoteOperations remoteOperations; transient CompiledReplicatedMapQueryContext remoteOpContext; @@ -118,6 +125,8 @@ public class ReplicatedChronicleMap c.context().remove(c)); @@ -227,7 +254,9 @@ public class ReplicatedChronicleMap 0L; + next = modIterSet.nextSetBit(next + 1L)) { + ModificationIterator modificationIterator = + acquireModificationIterator((byte) next); + if (modificationIterator.isChanged(segmentIndex, pos)) + return true; + } + return false; + } + @Override public boolean identifierCheck(@NotNull ReplicableEntry entry, int chronicleId) { return entry.originIdentifier() == localIdentifier; @@ -551,6 +591,10 @@ public class ReplicatedChronicleMap { + c.forEachSegmentReplicableEntry(e -> { if (debugEnabled) { LOG.debug("Bootstrap entry: id {}, key {}, value {}", localIdentifier, c.key(), c.value()); } - MapReplicableEntry re = (MapReplicableEntry) entry; // Bizarrely the next line line cause NPE in JDT compiler //assert re.originTimestamp() > 0L; if (debugEnabled) { LOG.debug("Bootstrap decision: bs ts: {}, entry ts: {}, " + "entry id: {}, local id: {}", - fromTimeStamp, re.originTimestamp(), - re.originIdentifier(), localIdentifier); + fromTimeStamp, e.originTimestamp(), + e.originIdentifier(), localIdentifier); } - if (re.originTimestamp() >= fromTimeStamp && + if (e.originTimestamp() >= fromTimeStamp && (!bootstrapOnlyLocalEntries || - re.originIdentifier() == localIdentifier)) { + e.originIdentifier() == localIdentifier)) { raiseChange(segmentIndex, c.pos(), c.timestamp()); } }); diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMapGlobalMutableState.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMapGlobalMutableState.java new file mode 100644 index 000000000..c12097777 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMapGlobalMutableState.java @@ -0,0 +1,30 @@ +/* + * 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 net.openhft.lang.model.Byteable; + +interface ReplicatedChronicleMapGlobalMutableState extends Byteable { + + int getCurrentCleanupSegmentIndex(); + + void setCurrentCleanupSegmentIndex(int currentCleanupSegmentIndex); + + boolean isCurrentlyCleanupIterated(); + + void setCurrentlyCleanupIterated(boolean isCurrentlyCleanupIterated); +} diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 11f05375c..676f45ad8 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -409,6 +409,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } + @Override + public boolean isOpen() { + return closeables != null; + } + /** * the transaction id are generated as unique timestamps * @@ -558,6 +563,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon throw new UnsupportedOperationException(); } + @Override + public int segments() { + throw new UnsupportedOperationException(); + } + public V get(Object key) { return fetchObject(vClass, GET, (K) key); } diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 05e9a2d33..a923b3aef 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -199,11 +199,6 @@ public class VanillaChronicleMap extends IterationContextInterface { - void forEachReplicableEntry(Consumer action); - void readExistingEntry(long pos); } 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 665d1cc0d..9dc0e2674 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java @@ -26,11 +26,11 @@ import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData; import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceData; 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.map.MapEntryOperationsDelegation; -import net.openhft.chronicle.map.impl.stage.map.ReplicatedChronicleMapHolderImpl; -import net.openhft.chronicle.map.impl.stage.map.ValueBytesInterop; -import net.openhft.chronicle.map.impl.stage.map.WrappedValueInstanceValueHolder; +import net.openhft.chronicle.map.impl.stage.map.*; +import net.openhft.chronicle.map.impl.stage.query.ReplicatedMapAbsent; import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate; import net.openhft.sg.Context; import net.openhft.sg.Staged; @@ -56,6 +56,7 @@ import net.openhft.sg.Staged; MapEntryOperationsDelegation.class, ReplicationUpdate.class, + DefaultValue.class, }, nested = { ReadLock.class, UpdateLock.class, @@ -67,6 +68,9 @@ import net.openhft.sg.Staged; WrappedValueInstanceData.class, DummyValueZeroData.class, + + ReplicatedMapAbsentDelegatingForIteration.class, + ReplicatedMapEntryDelegating.class }) public class ReplicatedMapIterationContext { } diff --git a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java index f4afc1c16..00937a4dc 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java @@ -35,10 +35,7 @@ import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceDa import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; import net.openhft.chronicle.map.impl.stage.input.MapInputBytesValues; import net.openhft.chronicle.map.impl.stage.input.ReplicatedInput; -import net.openhft.chronicle.map.impl.stage.map.MapEntryOperationsDelegation; -import net.openhft.chronicle.map.impl.stage.map.ReplicatedChronicleMapHolderImpl; -import net.openhft.chronicle.map.impl.stage.map.ValueBytesInterop; -import net.openhft.chronicle.map.impl.stage.map.WrappedValueInstanceValueHolder; +import net.openhft.chronicle.map.impl.stage.map.*; import net.openhft.chronicle.map.impl.stage.query.*; import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate; import net.openhft.chronicle.map.impl.stage.ret.BytesReturnValue; @@ -58,7 +55,7 @@ import net.openhft.sg.Staged; ReplicatedChronicleMapHolderImpl.class, KeyBytesInterop.class, - QuerySegmentStages.class, + ReplicatedQuerySegmentStages.class, KeySearch.class, HashLookupSearch.class, QueryHashLookupPos.class, @@ -71,6 +68,7 @@ import net.openhft.sg.Staged; WrappedValueInstanceValueHolder.class, ReplicatedMapQuery.class, ReplicatedMapAbsent.class, + DefaultValue.class, ValueBytesInterop.class, ReplicatedMapAbsentHolder.class, ReplicationUpdate.class, 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 e46be08b9..4d0fd9e49 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 @@ -18,18 +18,20 @@ package net.openhft.chronicle.map.impl.stage.entry; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.replication.TimeProvider; +import net.openhft.chronicle.map.MapAbsentEntry; import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder; import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate; import net.openhft.chronicle.map.replication.MapReplicableEntry; import net.openhft.sg.Stage; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; +import org.jetbrains.annotations.NotNull; import static net.openhft.chronicle.map.ReplicatedChronicleMap.ADDITIONAL_ENTRY_BYTES; @Staged public abstract class ReplicatedMapEntryStages extends MapEntryStages - implements MapReplicableEntry { + implements MapReplicableEntry, MapAbsentEntry { @StageRef ReplicatedChronicleMapHolder mh; @StageRef ReplicationUpdate ru; @@ -116,6 +118,13 @@ public abstract class ReplicatedMapEntryStages extends MapEntryStages extends MapEntryStages absentKey() { + return key(); + } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapAbsentDelegatingForIteration.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapAbsentDelegatingForIteration.java new file mode 100644 index 000000000..e9279c273 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapAbsentDelegatingForIteration.java @@ -0,0 +1,63 @@ +/* + * 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.hash.Data; +import net.openhft.chronicle.hash.impl.stage.replication.ReplicableEntryDelegating; +import net.openhft.chronicle.hash.replication.ReplicableEntry; +import net.openhft.chronicle.map.MapAbsentEntry; +import net.openhft.chronicle.map.MapContext; +import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; +import net.openhft.sg.StageRef; +import net.openhft.sg.Staged; +import org.jetbrains.annotations.NotNull; + +@Staged +public class ReplicatedMapAbsentDelegatingForIteration + implements MapAbsentEntry, ReplicableEntryDelegating { + + @StageRef ReplicatedMapSegmentIteration delegate; + @StageRef ReplicatedMapEntryStages e; + + @NotNull + @Override + public MapContext context() { + return delegate.context(); + } + + @Override + public void doInsert(Data value) { + delegate.doInsert(value); + } + + @NotNull + @Override + public Data defaultValue() { + return delegate.defaultValue(); + } + + @NotNull + @Override + public Data absentKey() { + return delegate.absentKey(); + } + + @Override + public ReplicableEntry d() { + return e; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapEntryDelegating.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapEntryDelegating.java new file mode 100644 index 000000000..6454a7bb0 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapEntryDelegating.java @@ -0,0 +1,68 @@ +/* + * 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.hash.Data; +import net.openhft.chronicle.hash.impl.stage.replication.ReplicableEntryDelegating; +import net.openhft.chronicle.hash.replication.ReplicableEntry; +import net.openhft.chronicle.map.MapContext; +import net.openhft.chronicle.map.MapEntry; +import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; +import net.openhft.sg.StageRef; +import net.openhft.sg.Staged; +import org.jetbrains.annotations.NotNull; + +@Staged +public class ReplicatedMapEntryDelegating + implements MapEntry, ReplicableEntryDelegating { + + @StageRef ReplicatedMapSegmentIteration delegate; + @StageRef ReplicatedMapEntryStages e; + + @NotNull + @Override + public MapContext context() { + return delegate.context(); + } + + @NotNull + @Override + public Data key() { + return delegate.key(); + } + + @NotNull + @Override + public Data value() { + return delegate.value(); + } + + @Override + public void doReplaceValue(Data newValue) { + delegate.doReplaceValue(newValue); + } + + @Override + public void doRemove() { + delegate.doRemove(); + } + + @Override + public ReplicableEntry d() { + return e; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java index 378c164df..cddf981c0 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java @@ -17,9 +17,12 @@ package net.openhft.chronicle.map.impl.stage.iter; import net.openhft.chronicle.hash.Data; +import net.openhft.chronicle.hash.ReplicatedHashSegmentContext; import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable; import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; import net.openhft.chronicle.hash.replication.ReplicableEntry; +import net.openhft.chronicle.map.MapAbsentEntry; +import net.openhft.chronicle.map.MapEntry; import net.openhft.chronicle.map.impl.ReplicatedIterationContextInterface; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; @@ -28,15 +31,19 @@ import net.openhft.sg.StageRef; import net.openhft.sg.Staged; import java.util.function.Consumer; +import java.util.function.Predicate; @Staged public abstract class ReplicatedMapSegmentIteration extends MapSegmentIteration - implements ReplicatedIterationContextInterface, ReplicableEntry { + implements ReplicatedIterationContextInterface, ReplicableEntry, + ReplicatedHashSegmentContext>, MapAbsentEntry { @StageRef VanillaChronicleHashHolder hh; @StageRef ReplicatedMapEntryStages e; @StageRef ReplicationUpdate ru; @StageRef DummyValueZeroData dummyValue; + @StageRef ReplicatedMapAbsentDelegatingForIteration absentEntryDelegating; + @StageRef ReplicatedMapEntryDelegating entryDelegating; @Override public boolean entryIsPresent() { @@ -56,12 +63,14 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI } @Override - public void forEachReplicableEntry(Consumer action) { + public boolean forEachSegmentReplicableEntryWhile( + Predicate predicate) { s.innerUpdateLock.lock(); try { long entries = s.entries(); if (entries == 0) - return; + return true; + boolean interrupted = false; long startPos = 0L; CompactOffHeapLinearHashTable hashLookup = hh.h().hashLookup; while (!hashLookup.empty(hashLookup.readEntry(s.segmentBase, startPos))) { @@ -73,22 +82,36 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI long entry = hashLookup.readEntry(s.segmentBase, hlp.hashLookupPos); if (!hashLookup.empty(entry)) { e.readExistingEntry(hashLookup.value(entry)); - action.accept(this); - if (--entries == 0) + ReplicableEntry e = !this.e.entryDeleted() ? entryDelegating : + absentEntryDelegating; + initEntryRemovedOnThisIteration(false); + if (!predicate.test(e)) { + interrupted = true; break; + } else { + if (--entries == 0) + break; + } } } while (hlp.hashLookupPos != startPos); + return !interrupted; } finally { s.innerReadLock.unlock(); + initEntryRemovedOnThisIteration(false); } } - + @Override + public void forEachSegmentReplicableEntry(Consumer action) { + forEachSegmentReplicableEntryWhile(e -> { + action.accept(e); + return true; + }); + } @Override public void doRemove() { checkOnEachPublicOperation.checkOnEachPublicOperation(); - initEntryRemovedOnThisIteration(true); try { if (e.valueSize > dummyValue.size()) e.innerDefaultReplaceValue(dummyValue); @@ -99,5 +122,34 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI } finally { s.innerWriteLock.unlock(); } + initEntryRemovedOnThisIteration(true); + } + + @Override + public void doRemoveCompletely() { + boolean wasDeleted = e.entryDeleted(); + super.doRemove(); + if (wasDeleted) + s.deleted(s.deleted() - 1); + } + + @Override + public void doInsert(Data value) { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + if (e.entryDeleted()) { + try { + e.innerDefaultReplaceValue(value); + s.deleted(s.deleted() - 1); + s.incrementModCount(); + e.writeEntryPresent(); + ru.updateChange(); + e.updatedReplicationStateOnPresentEntry(); + } finally { + s.innerWriteLock.unlock(); + } + } else { + throw new IllegalStateException( + "Entry is present in the map when doInsert() is called"); + } } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/map/DefaultValue.java b/src/main/java/net/openhft/chronicle/map/impl/stage/map/DefaultValue.java new file mode 100644 index 000000000..02a676de2 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/map/DefaultValue.java @@ -0,0 +1,45 @@ +/* + * 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.map; + +import net.openhft.chronicle.hash.Data; +import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; +import net.openhft.chronicle.map.MapAbsentEntry; +import net.openhft.chronicle.map.impl.VanillaChronicleMapHolder; +import net.openhft.sg.StageRef; +import net.openhft.sg.Staged; +import org.jetbrains.annotations.NotNull; + +@Staged +public abstract class DefaultValue implements MapAbsentEntry { + + @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation; + @StageRef VanillaChronicleMapHolder mh; + + @NotNull + @Override + public Data defaultValue() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + if (mh.m().constantValueProvider == null) { + throw new IllegalStateException("to call acquireUsing(), " + + "or defaultValue() on AbsentEntry, you should configure " + + "ChronicleMapBuilder.defaultValue(), or use one of the 'known' value types: " + + "boxed primitives, or so-called data-value-generated interface as a value"); + } + return context().wrapValueAsData(mh.m().constantValueProvider.defaultValue()); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java index be73f6dd6..030585c6b 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java @@ -32,7 +32,7 @@ import org.jetbrains.annotations.NotNull; import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.PRESENT; @Staged -public class MapAbsent implements MapAbsentEntry { +public abstract class MapAbsent implements MapAbsentEntry { @StageRef public KeySearch ks; @StageRef MapQuery q; @@ -81,17 +81,4 @@ public class MapAbsent implements MapAbsentEntry { "Entry is present in the map when doInsert() is called"); } } - - @NotNull - @Override - public Data defaultValue() { - checkOnEachPublicOperation.checkOnEachPublicOperation(); - if (mh.m().constantValueProvider == null) { - throw new IllegalStateException("to call acquireUsing(), " + - "or defaultValue() on AbsentEntry, you should configure " + - "ChronicleMapBuilder.defaultValue(), or use one of the 'known' value types: " + - "boxed primitives, or so-called data-value-generated interface as a value"); - } - return context().wrapValueAsData(mh.m().constantValueProvider.defaultValue()); - } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java index 0bc56d977..229abe3f9 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsent.java @@ -25,7 +25,7 @@ import net.openhft.sg.Staged; import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.PRESENT; @Staged -public class ReplicatedMapAbsent extends MapAbsent { +public abstract class ReplicatedMapAbsent extends MapAbsent { @StageRef MapQuery q; @StageRef ReplicatedMapEntryStages e; diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java index e99809fec..524c70eab 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.map.impl.stage.query; import net.openhft.chronicle.hash.Data; +import net.openhft.chronicle.hash.replication.ReplicableEntry; import net.openhft.chronicle.map.MapAbsentEntry; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; @@ -28,9 +29,11 @@ import net.openhft.sg.StageRef; import net.openhft.sg.Staged; import org.jetbrains.annotations.Nullable; +import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.DELETED; + @Staged public abstract class ReplicatedMapQuery extends MapQuery - implements MapRemoteQueryContext { + implements MapRemoteQueryContext, ReplicableEntry { @StageRef ReplicatedMapEntryStages e; @StageRef ReplicationUpdate ru; @@ -83,6 +86,14 @@ public abstract class ReplicatedMapQuery extends MapQuery } } + @Override + public void doRemoveCompletely() { + boolean wasDeleted = e.entryDeleted(); + super.doRemove(); + if (wasDeleted) + s.deleted(s.deleted() - 1L); + } + @Override public void doReplaceValue(Data newValue) { super.doReplaceValue(newValue); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java new file mode 100644 index 000000000..750fa0b00 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedQuerySegmentStages.java @@ -0,0 +1,71 @@ +/* + * 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.query; + +import net.openhft.chronicle.hash.ReplicatedHashSegmentContext; +import net.openhft.chronicle.hash.impl.stage.query.QuerySegmentStages; +import net.openhft.chronicle.hash.replication.ReplicableEntry; +import net.openhft.chronicle.map.MapAbsentEntry; +import net.openhft.chronicle.map.MapSegmentContext; +import net.openhft.chronicle.map.ReplicatedChronicleMap; +import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder; +import net.openhft.sg.StageRef; +import net.openhft.sg.Staged; + +import java.util.function.Consumer; + +@Staged +public abstract class ReplicatedQuerySegmentStages extends QuerySegmentStages { + + @StageRef + ReplicatedChronicleMapHolder mh; + + final Consumer cleanupAction = e -> { + ReplicatedChronicleMap map = mh.m(); + if (e instanceof MapAbsentEntry) { + long deleteTimeout = map.timeProvider.currentTime() - e.originTimestamp(); + if (deleteTimeout > map.cleanupTimeout && !e.isChanged()) + e.doRemoveCompletely(); + } + }; + + @Override + public long alloc(int chunks) { + long ret = allocReturnCode(chunks); + if (ret >= 0) { + return ret; + } else { + ReplicatedChronicleMap map = mh.m(); + MapSegmentContext sc = map.segmentContext(segmentIndex); + ((ReplicatedHashSegmentContext) sc).forEachSegmentReplicableEntry(cleanupAction); + ret = allocReturnCode(chunks); + if (ret >= 0) { + return ret; + } else { + if (chunks == 1) { + throw new IllegalStateException( + "Segment is full, no free entries found"); + } else { + throw new IllegalStateException( + "Segment is full or has no ranges of " + chunks + + " continuous free chunks" + ); + } + } + } + } +} diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java index cbda10ef0..5fda8e6df 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/replication/ReplicationUpdate.java @@ -62,6 +62,10 @@ public abstract class ReplicationUpdate implements RemoteOperationContext mh.m().raiseChange(s.segmentIndex, e.pos, e.timestamp()); } + public boolean changed() { + return mh.m().isChanged(s.segmentIndex, e.pos); + } + @Override public long remoteTimestamp() { checkOnEachPublicOperation.checkOnEachPublicOperation(); diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index 4632fe981..c4a0622db 100755 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -37,6 +37,7 @@ import org.jetbrains.annotations.NotNull; import java.io.File; import java.io.IOException; +import java.util.concurrent.TimeUnit; /** * {@code ChronicleSetBuilder} manages the whole set of {@link ChronicleSet} configurations, could @@ -189,6 +190,13 @@ public final class ChronicleSetBuilder return this; } + @Override + public ChronicleSetBuilder removedEntryCleanupTimeout( + long removedEntryCleanupTimeout, TimeUnit unit) { + chronicleMapBuilder.removedEntryCleanupTimeout(removedEntryCleanupTimeout, unit); + return this; + } + @Override public ChronicleSetBuilder bytesMarshallerFactory( BytesMarshallerFactory bytesMarshallerFactory) { diff --git a/src/main/java/net/openhft/chronicle/set/SetFromMap.java b/src/main/java/net/openhft/chronicle/set/SetFromMap.java index 28800afaa..45533bbf0 100644 --- a/src/main/java/net/openhft/chronicle/set/SetFromMap.java +++ b/src/main/java/net/openhft/chronicle/set/SetFromMap.java @@ -144,6 +144,11 @@ class SetFromMap extends AbstractSet throw new UnsupportedOperationException(); } + @Override + public int segments() { + return m.segments(); + } + @Override public boolean forEachEntryWhile(Predicate> predicate) { throw new UnsupportedOperationException(); @@ -163,4 +168,9 @@ class SetFromMap extends AbstractSet public void close() { m.close(); } + + @Override + public boolean isOpen() { + return m.isOpen(); + } } diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index a3bac9b61..44f415199 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -298,6 +298,11 @@ public class ReplicationCheckingMap implements ChronicleMap { throw new UnsupportedOperationException(); } + @Override + public int segments() { + throw new UnsupportedOperationException(); + } + @Override public File file() { throw new UnsupportedOperationException(); @@ -309,6 +314,11 @@ public class ReplicationCheckingMap implements ChronicleMap { map2.close(); } + @Override + public boolean isOpen() { + throw new UnsupportedOperationException(); + } + @Override public boolean equals(Object o) { return map1.equals(o); diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java index 09bab260d..4b9563680 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java @@ -92,6 +92,7 @@ public class TCPSocketReplication4WayMapTest { map1.put(1, "EXAMPLE-1"); map2.put(2, "EXAMPLE-1"); map3.put(3, "EXAMPLE-1"); + map4.put(3, "EXAMPLE-1"); map4.remove(3); // allow time for the recompilation to resolve diff --git a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java index d1bda8263..18979a148 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java @@ -16,11 +16,13 @@ package net.openhft.chronicle.map; +import com.google.common.collect.Maps; import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.model.Byteable; import net.openhft.lang.model.DataValueClasses; +import net.openhft.lang.threadlocal.ThreadLocalCopies; import net.openhft.lang.values.IntValue; import org.junit.*; @@ -28,10 +30,8 @@ import java.io.Closeable; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; -import java.util.HashMap; -import java.util.Map; -import java.util.Random; -import java.util.Set; +import java.util.*; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; /** @@ -60,7 +60,9 @@ public class TcpReplicationSoakTest { map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(Builder.SIZE + Builder.SIZE) + .removedEntryCleanupTimeout(1, TimeUnit.MILLISECONDS) .actualSegments(1) + .averageValue("test" + 1000) .replication(SingleChronicleHashReplication.builder() .tcpTransportAndNetwork(tcpConfig1) .name("map1") @@ -77,6 +79,8 @@ public class TcpReplicationSoakTest { map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(Builder.SIZE + Builder.SIZE) + .removedEntryCleanupTimeout(1, TimeUnit.MILLISECONDS) + .averageValue("test" + 1000) .replication(SingleChronicleHashReplication.builder() .tcpTransportAndNetwork(tcpConfig2) .name("map2") @@ -149,6 +153,34 @@ public class TcpReplicationSoakTest { } + @Test + public void testFloodReplicatedMapWithDeletedEntries() throws InterruptedException { + try { + Random r = ThreadLocalRandom.current(); + NavigableSet put = new TreeSet<>(); + for (int i = 0; i < Builder.SIZE * 10; i++) { + if (r.nextBoolean() || put.isEmpty()) { + int key = r.nextInt(); + map1.put(key, "test"); + put.add(key); + } else { + Integer key = put.pollFirst(); + map1.remove(key); + } +// Thread.sleep(0, 1000); + } + System.out.println("\nwaiting till equal"); + + waitTillEqual(15000); + + if (!map1.equals(map2)) + Assert.assertEquals(Maps.difference(map1, map2).toString(), map1, map2); + } finally { + map1.close(); + map2.close(); + } + } + private void waitTillEqual(final int timeOutMs) throws InterruptedException { diff --git a/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java index a375b2dd0..ebe0ebf31 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java @@ -63,6 +63,16 @@ public class TcpTimeBasedReplicationSoakTest { else return t; } + + @Override + public long scale(long time, TimeUnit unit) { + return TimeUnit.MILLISECONDS.convert(time, unit); + } + + @Override + public long unscale(long time, TimeUnit toUnit) { + return toUnit.convert(time, TimeUnit.MILLISECONDS); + } }; { diff --git a/src/test/java/net/openhft/chronicle/map/jsr166/map/StatelessChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/jsr166/map/StatelessChronicleMapTest.java index d085ee58d..08f4d0af6 100755 --- a/src/test/java/net/openhft/chronicle/map/jsr166/map/StatelessChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/jsr166/map/StatelessChronicleMapTest.java @@ -82,6 +82,11 @@ public class StatelessChronicleMapTest extends JSR166TestCase { } } + @Override + public boolean isOpen() { + return d.isOpen(); + } + @Override public long longSize() { return d.longSize(); @@ -130,6 +135,11 @@ public class StatelessChronicleMapTest extends JSR166TestCase { return d.segmentContext(segmentIndex); } + @Override + public int segments() { + return d.segments(); + } + @Override public V get(Object key) { return d.get(key);