diff --git a/pom.xml b/pom.xml index 2dcd66015..891685714 100755 --- a/pom.xml +++ b/pom.xml @@ -152,7 +152,7 @@ net.openhft stage-compiler - 1.7 + 1.8 provided @@ -238,6 +238,7 @@ net/openhft/chronicle/map/VanillaChronicleMap.java net/openhft/chronicle/map/Replica.java net/openhft/chronicle/map/ReplicatedChronicleMap.java + net/openhft/chronicle/set/DummyValueData.java false -Xdoclint:none 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 a455dae5e..d7e14c7a2 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 @@ -25,6 +25,7 @@ import net.openhft.chronicle.hash.impl.stage.entry.HashLookupSearch; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; import net.openhft.chronicle.hash.serialization.DataAccess; +import net.openhft.chronicle.set.SetEntry; import net.openhft.sg.Stage; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -33,7 +34,7 @@ import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState. import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.PRESENT; @Staged -public abstract class HashQuery implements HashEntry { +public abstract class HashQuery implements SetEntry { @StageRef public VanillaChronicleHashHolder hh; @StageRef public SegmentStages s; diff --git a/src/main/java/net/openhft/chronicle/map/MapContext.java b/src/main/java/net/openhft/chronicle/map/MapContext.java index 261ec0685..3bc67169f 100644 --- a/src/main/java/net/openhft/chronicle/map/MapContext.java +++ b/src/main/java/net/openhft/chronicle/map/MapContext.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.chronicle.hash.HashContext; import net.openhft.chronicle.hash.Data; @@ -35,9 +36,7 @@ public interface MapContext * Returns the accessed {@code ChronicleMap}. Synonym to {@link #map()}. */ @Override - default ChronicleMap hash() { - return map(); - } + ChronicleHash hash(); /** * Returns the accessed {@code ChronicleMap}. Synonym to {@link #hash()}. diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index afe89a5a9..9ba6313fc 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -30,6 +30,7 @@ import net.openhft.chronicle.map.impl.*; import net.openhft.chronicle.map.impl.CompiledMapIterationContext; import net.openhft.chronicle.map.impl.CompiledMapQueryContext; import net.openhft.chronicle.map.impl.ret.InstanceReturnValue; +import net.openhft.chronicle.set.ChronicleSet; import net.openhft.chronicle.values.Values; import net.openhft.chronicle.wire.WireIn; import net.openhft.chronicle.wire.WireOut; @@ -72,6 +73,9 @@ public class VanillaChronicleMap transient boolean removeReturnsNull; transient Set> entrySet; + + /** @see net.openhft.chronicle.set.SetFromMap */ + public transient ChronicleSet chronicleSet; public transient MapEntryOperations entryOperations; public transient MapMethods methods; diff --git a/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java b/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java index bf4756133..2eac4e63e 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java @@ -58,7 +58,6 @@ import net.openhft.sg.Staged; MapAbsent.class, DefaultValue.class, ValueBytesInterop.class, - MapAbsentHolder.class, QueryAlloc.class, }, 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 9be94aba6..638108232 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java @@ -67,7 +67,6 @@ import net.openhft.sg.Staged; ReplicatedMapAbsent.class, DefaultValue.class, ValueBytesInterop.class, - ReplicatedMapAbsentHolder.class, ReplicationUpdate.class, ReplicatedInput.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/VanillaChronicleMapHolder.java b/src/main/java/net/openhft/chronicle/map/impl/VanillaChronicleMapHolder.java index 539cc429e..d9ab52b9a 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/VanillaChronicleMapHolder.java +++ b/src/main/java/net/openhft/chronicle/map/impl/VanillaChronicleMapHolder.java @@ -20,6 +20,7 @@ import net.openhft.chronicle.hash.impl.VanillaChronicleHash; import net.openhft.chronicle.hash.impl.VanillaChronicleHashHolder; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.VanillaChronicleMap; +import net.openhft.chronicle.set.ChronicleSet; public interface VanillaChronicleMapHolder extends VanillaChronicleHashHolder { @@ -33,4 +34,6 @@ public interface VanillaChronicleMapHolder extends VanillaChronicleHash default ChronicleMap map() { return m(); } + + ChronicleSet set(); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java index 0da4ebaba..0ed6de1fb 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java @@ -47,7 +47,7 @@ public abstract class MapSegmentIteration extends HashSegmentIteration< @NotNull @Override - public MapContext context() { + public WrappedValueInstanceValueHolder context() { return wrappedValueInstanceValueHolder; } } 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 index 9666a4802..3809383bb 100644 --- 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 @@ -20,22 +20,23 @@ 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.chronicle.map.impl.stage.map.WrappedValueInstanceValueHolder; +import net.openhft.chronicle.set.SetAbsentEntry; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; import org.jetbrains.annotations.NotNull; @Staged public class ReplicatedMapAbsentDelegatingForIteration - implements MapAbsentEntry, ReplicableEntryDelegating { + implements MapAbsentEntry, SetAbsentEntry, ReplicableEntryDelegating { @StageRef ReplicatedMapSegmentIteration delegate; @StageRef ReplicatedMapEntryStages e; @NotNull @Override - public MapContext context() { + public WrappedValueInstanceValueHolder context() { return delegate.context(); } @@ -44,6 +45,11 @@ public class ReplicatedMapAbsentDelegatingForIteration delegate.doInsert(value); } + @Override + public void doInsert() { + delegate.doInsert(); + } + @NotNull @Override public Data defaultValue() { 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 d1f642f62..ff6e68535 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 @@ -19,14 +19,16 @@ 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.ReplicatedIterationContext; +import net.openhft.chronicle.map.impl.VanillaChronicleMapHolder; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate; +import net.openhft.chronicle.set.DummyValueData; +import net.openhft.chronicle.set.SetAbsentEntry; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -36,9 +38,9 @@ import java.util.function.Predicate; @Staged public abstract class ReplicatedMapSegmentIteration extends MapSegmentIteration implements ReplicatedIterationContext, ReplicableEntry, - ReplicatedHashSegmentContext>, MapAbsentEntry { + ReplicatedHashSegmentContext>, MapAbsentEntry, SetAbsentEntry { - @StageRef VanillaChronicleHashHolder hh; + @StageRef VanillaChronicleMapHolder mh; @StageRef ReplicatedMapEntryStages e; @StageRef ReplicationUpdate ru; @StageRef DummyValueZeroData dummyValue; @@ -73,7 +75,7 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI return true; boolean interrupted = false; long startPos = 0L; - CompactOffHeapLinearHashTable hashLookup = hh.h().hashLookup; + CompactOffHeapLinearHashTable hashLookup = mh.h().hashLookup; while (!hashLookup.empty(hashLookup.readEntry(s.tierBaseAddr, startPos))) { startPos = hashLookup.step(startPos); } @@ -155,4 +157,11 @@ public abstract class ReplicatedMapSegmentIteration extends MapSegmentI "Entry is present in the map when doInsert() is called"); } } + + @Override + public void doInsert() { + if (mh.set() == null) + throw new IllegalStateException("Called SetAbsentEntry.doInsert() from Map context"); + doInsert((Data) DummyValueData.INSTANCE); + } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java b/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java index 0a9dd4ec2..e9c77692c 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java @@ -16,11 +16,13 @@ package net.openhft.chronicle.map.impl.stage.map; +import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.chronicle.hash.impl.stage.hash.Chaining; import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ReplicatedChronicleMap; import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder; +import net.openhft.chronicle.set.ChronicleSet; import net.openhft.sg.Staged; @Staged @@ -49,4 +51,13 @@ public class ReplicatedChronicleMapHolderImpl extends Chaining public ChronicleMap map() { return m(); } + + @Override + public ChronicleSet set() { + return m.chronicleSet; + } + + public ChronicleHash hash() { + return set() != null ? set() : map(); + } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java b/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java index 7360eecac..29b0a5f4a 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java @@ -16,11 +16,13 @@ package net.openhft.chronicle.map.impl.stage.map; +import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.chronicle.hash.impl.stage.hash.Chaining; import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.VanillaChronicleMap; import net.openhft.chronicle.map.impl.VanillaChronicleMapHolder; +import net.openhft.chronicle.set.ChronicleSet; import net.openhft.sg.Staged; @Staged @@ -49,4 +51,13 @@ public class VanillaChronicleMapHolderImpl extends Chaining public ChronicleMap map() { return m(); } + + @Override + public ChronicleSet set() { + return m.chronicleSet; + } + + public ChronicleHash hash() { + return set() != null ? set() : map(); + } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueInstanceValueHolder.java b/src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueInstanceValueHolder.java index d198032c4..33f6e26aa 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueInstanceValueHolder.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueInstanceValueHolder.java @@ -20,11 +20,13 @@ import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; import net.openhft.chronicle.map.MapContext; import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceDataHolder; +import net.openhft.chronicle.set.SetContext; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @Staged -public abstract class WrappedValueInstanceValueHolder implements MapContext { +public abstract class WrappedValueInstanceValueHolder + implements MapContext, SetContext { @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation; @StageRef diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsentHolder.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/Absent.java similarity index 73% rename from src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsentHolder.java rename to src/main/java/net/openhft/chronicle/map/impl/stage/query/Absent.java index 1d73ab68c..a5899b655 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsentHolder.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/Absent.java @@ -17,17 +17,12 @@ package net.openhft.chronicle.map.impl.stage.query; import net.openhft.chronicle.map.MapAbsentEntry; -import net.openhft.chronicle.map.impl.MapAbsentEntryHolder; -import net.openhft.sg.StageRef; -import net.openhft.sg.Staged; +import net.openhft.chronicle.set.SetAbsentEntry; +import org.jetbrains.annotations.NotNull; -@Staged -public class MapAbsentHolder implements MapAbsentEntryHolder { - - @StageRef MapAbsent ab; +public interface Absent extends MapAbsentEntry, SetAbsentEntry { + @NotNull @Override - public MapAbsentEntry absent() { - return ab; - } + MapAndSetContext context(); } 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 68fb59676..e92223b00 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 @@ -23,8 +23,10 @@ import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; import net.openhft.chronicle.hash.impl.stage.query.HashQuery.EntryPresence; import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.map.MapAbsentEntry; -import net.openhft.chronicle.map.MapContext; +import net.openhft.chronicle.map.impl.VanillaChronicleMapHolder; import net.openhft.chronicle.map.impl.stage.entry.MapEntryStages; +import net.openhft.chronicle.set.DummyValueData; +import net.openhft.chronicle.set.SetAbsentEntry; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; import org.jetbrains.annotations.NotNull; @@ -32,7 +34,7 @@ import org.jetbrains.annotations.NotNull; import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState.PRESENT; @Staged -public abstract class MapAbsent implements MapAbsentEntry { +public abstract class MapAbsent implements Absent { @StageRef public KeySearch ks; @StageRef MapQuery q; @@ -40,6 +42,7 @@ public abstract class MapAbsent implements MapAbsentEntry { @StageRef public HashLookupSearch hashLookupSearch; @StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation; @StageRef public SegmentStages s; + @StageRef VanillaChronicleMapHolder mh; void putEntry(Data value) { assert ks.searchStateAbsent(); @@ -52,7 +55,7 @@ public abstract class MapAbsent implements MapAbsentEntry { @NotNull @Override - public MapContext context() { + public MapQuery context() { checkOnEachPublicOperation.checkOnEachPublicOperation(); return q; } @@ -82,4 +85,11 @@ public abstract class MapAbsent implements MapAbsentEntry { "Entry is present in the map when doInsert() is called"); } } + + @Override + public void doInsert() { + if (mh.set() == null) + throw new IllegalStateException("Called SetAbsentEntry.doInsert() from Map context"); + doInsert((Data) DummyValueData.INSTANCE); + } } diff --git a/src/main/java/net/openhft/chronicle/map/impl/MapAbsentEntryHolder.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAndSetContext.java similarity index 76% rename from src/main/java/net/openhft/chronicle/map/impl/MapAbsentEntryHolder.java rename to src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAndSetContext.java index 30970f387..d90eefd33 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapAbsentEntryHolder.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAndSetContext.java @@ -14,11 +14,10 @@ * along with this program. If not, see . */ -package net.openhft.chronicle.map.impl; +package net.openhft.chronicle.map.impl.stage.query; -import net.openhft.chronicle.map.MapAbsentEntry; +import net.openhft.chronicle.map.MapContext; +import net.openhft.chronicle.set.SetContext; -public interface MapAbsentEntryHolder { - - MapAbsentEntry absent(); +public interface MapAndSetContext extends MapContext, SetContext { } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java index 33e2eab70..5629fd356 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java @@ -22,15 +22,13 @@ import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.hash.impl.stage.query.SearchAllocatedChunks; import net.openhft.chronicle.hash.serialization.DataAccess; import net.openhft.chronicle.map.ExternalMapQueryContext; -import net.openhft.chronicle.map.MapAbsentEntry; -import net.openhft.chronicle.map.MapContext; import net.openhft.chronicle.map.MapEntry; -import net.openhft.chronicle.map.impl.MapAbsentEntryHolder; import net.openhft.chronicle.map.impl.QueryContextInterface; import net.openhft.chronicle.map.impl.VanillaChronicleMapHolder; import net.openhft.chronicle.map.impl.stage.entry.MapEntryStages; import net.openhft.chronicle.map.impl.stage.ret.DefaultReturnValue; import net.openhft.chronicle.map.impl.stage.ret.UsingReturnValue; +import net.openhft.chronicle.set.ExternalSetQueryContext; import net.openhft.sg.Stage; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -42,7 +40,7 @@ import static net.openhft.chronicle.hash.impl.stage.query.KeySearch.SearchState. @Staged public abstract class MapQuery extends HashQuery implements MapEntry, ExternalMapQueryContext, - QueryContextInterface { + ExternalSetQueryContext, QueryContextInterface, MapAndSetContext { @StageRef VanillaChronicleMapHolder mh; @StageRef MapEntryStages e; @@ -53,7 +51,7 @@ public abstract class MapQuery extends HashQuery @StageRef public DefaultReturnValue defaultReturnValue; @StageRef public UsingReturnValue usingReturnValue; - @StageRef public MapAbsentEntryHolder absent; + @StageRef public MapAbsent absent; final DataAccess innerInputValueDataAccess = mh.m().valueDataAccess.copy(); @@ -76,16 +74,16 @@ public abstract class MapQuery extends HashQuery } @Override - public MapEntry entry() { + public MapQuery entry() { checkOnEachPublicOperation.checkOnEachPublicOperation(); return entryPresent() ? this : null; } @Nullable @Override - public MapAbsentEntry absentEntry() { + public Absent absentEntry() { checkOnEachPublicOperation.checkOnEachPublicOperation(); - return entryPresent() ? null : absent.absent(); + return entryPresent() ? null : absent; } protected void putPrefix() { @@ -120,7 +118,7 @@ public abstract class MapQuery extends HashQuery @NotNull @Override - public MapContext context() { + public MapQuery context() { checkOnEachPublicOperation.checkOnEachPublicOperation(); return this; } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsentDelegating.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsentDelegating.java index 80cc09079..aed59e26d 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsentDelegating.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsentDelegating.java @@ -17,23 +17,26 @@ package net.openhft.chronicle.map.impl.stage.query; import net.openhft.chronicle.hash.Data; -import net.openhft.chronicle.map.MapAbsentEntry; -import net.openhft.chronicle.map.MapContext; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; import org.jetbrains.annotations.NotNull; @Staged -public class ReplicatedMapAbsentDelegating implements MapAbsentEntry { +public class ReplicatedMapAbsentDelegating implements Absent { @StageRef ReplicatedMapAbsent delegate; @NotNull @Override - public MapContext context() { + public MapQuery context() { return delegate.context(); } + @Override + public void doInsert() { + delegate.doInsert(); + } + @Override public void doInsert(Data value) { delegate.doInsert(value); diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsentHolder.java b/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsentHolder.java deleted file mode 100644 index 05b7eba7b..000000000 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsentHolder.java +++ /dev/null @@ -1,33 +0,0 @@ -/* - * 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.map.MapAbsentEntry; -import net.openhft.chronicle.map.impl.MapAbsentEntryHolder; -import net.openhft.sg.StageRef; -import net.openhft.sg.Staged; - -@Staged -public class ReplicatedMapAbsentHolder implements MapAbsentEntryHolder { - - @StageRef ReplicatedMapAbsent ab; - - @Override - public MapAbsentEntry absent() { - return ab; - } -} 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 4aff702b9..b32101712 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 @@ -18,19 +18,21 @@ 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; import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate; import net.openhft.chronicle.map.replication.MapRemoteQueryContext; import net.openhft.chronicle.map.replication.MapReplicableEntry; +import net.openhft.chronicle.set.replication.SetRemoteQueryContext; +import net.openhft.chronicle.set.replication.SetReplicableEntry; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; import org.jetbrains.annotations.Nullable; @Staged public abstract class ReplicatedMapQuery extends MapQuery - implements MapRemoteQueryContext, ReplicableEntry { + implements MapRemoteQueryContext, SetRemoteQueryContext, + ReplicableEntry, MapReplicableEntry, SetReplicableEntry { @StageRef ReplicatedMapEntryStages e; @StageRef ReplicationUpdate ru; @@ -39,7 +41,7 @@ public abstract class ReplicatedMapQuery extends MapQuery @Nullable @Override - public MapAbsentEntry absentEntry() { + public Absent absentEntry() { checkOnEachPublicOperation.checkOnEachPublicOperation(); if (entryPresent()) { return null; @@ -48,7 +50,7 @@ public abstract class ReplicatedMapQuery extends MapQuery return absentDelegating; } else { assert e.entryDeleted(); - return absent.absent(); + return absent; } } } @@ -61,8 +63,9 @@ public abstract class ReplicatedMapQuery extends MapQuery @StageRef DummyValueZeroData dummyValue; @Override - public MapReplicableEntry entry() { - return (MapReplicableEntry) super.entry(); + public ReplicatedMapQuery entry() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return entryPresent() ? this : null; } @Override diff --git a/src/main/java/net/openhft/chronicle/map/replication/MapRemoteQueryContext.java b/src/main/java/net/openhft/chronicle/map/replication/MapRemoteQueryContext.java index 974e21832..ed5709539 100644 --- a/src/main/java/net/openhft/chronicle/map/replication/MapRemoteQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/replication/MapRemoteQueryContext.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.map.replication; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.replication.RemoteOperationContext; +import net.openhft.chronicle.hash.serialization.SizeMarshaller; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.ChronicleMapBuilder; import net.openhft.chronicle.map.MapEntryOperations; @@ -40,7 +41,7 @@ public interface MapRemoteQueryContext extends MapQueryContext /** * The value used as a tombstone, for removed entries, to save space. This value has {@linkplain - * SizeMarshaller#minEncodableSize() minimum possible size} for {@linkplain + * SizeMarshaller#minStorableSize()} minimum possible size} for {@linkplain * ChronicleMapBuilder#valueSizeMarshaller(SizeMarshaller) the configured value size * marshaller}. * diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index ef6b6cf2b..226ba3349 100755 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -19,12 +19,16 @@ package net.openhft.chronicle.set; import net.openhft.chronicle.hash.ChronicleHashBuilder; import net.openhft.chronicle.hash.ChronicleHashBuilderPrivateAPI; import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; +import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.hash.replication.TimeProvider; import net.openhft.chronicle.hash.serialization.*; -import net.openhft.chronicle.map.ChronicleMap; -import net.openhft.chronicle.map.ChronicleMapBuilder; +import net.openhft.chronicle.map.*; +import net.openhft.chronicle.map.replication.MapRemoteOperations; +import net.openhft.chronicle.map.replication.MapRemoteQueryContext; +import net.openhft.chronicle.set.replication.SetRemoteOperations; +import net.openhft.chronicle.set.replication.SetRemoteQueryContext; import org.jetbrains.annotations.NotNull; import java.io.File; @@ -287,6 +291,38 @@ public final class ChronicleSetBuilder return this; } + public ChronicleSetBuilder entryOperations(SetEntryOperations entryOperations) { + chronicleMapBuilder.entryOperations(new MapEntryOperations() { + @Override + public Object remove(@NotNull MapEntry entry) { + return entryOperations.remove((SetEntry) entry); + } + + @Override + public Object insert( + @NotNull MapAbsentEntry absentEntry, Data value) { + return entryOperations.insert((SetAbsentEntry) absentEntry); + } + }); + return this; + } + + public ChronicleSetBuilder remoteOperations(SetRemoteOperations remoteOperations) { + chronicleMapBuilder.remoteOperations(new MapRemoteOperations() { + @Override + public void remove(MapRemoteQueryContext q) { + remoteOperations.remove((SetRemoteQueryContext) q); + } + + @Override + public void put( + MapRemoteQueryContext q, Data newValue) { + remoteOperations.put((SetRemoteQueryContext) q); + } + }); + return this; + } + @Override public ChronicleHashInstanceBuilder> instance() { return new SetInstanceBuilder<>(chronicleMapBuilder.instance()); @@ -295,13 +331,13 @@ public final class ChronicleSetBuilder @Override public ChronicleSet create() { final ChronicleMap map = chronicleMapBuilder.create(); - return new SetFromMap<>(map); + return new SetFromMap<>((VanillaChronicleMap) map); } @Override public ChronicleSet createPersistedTo(File file) throws IOException { ChronicleMap map = chronicleMapBuilder.createPersistedTo(file); - return new SetFromMap<>(map); + return new SetFromMap<>((VanillaChronicleMap) map); } /** diff --git a/src/main/java/net/openhft/chronicle/set/DummyValueData.java b/src/main/java/net/openhft/chronicle/set/DummyValueData.java index 505308f83..d4008c9a1 100644 --- a/src/main/java/net/openhft/chronicle/set/DummyValueData.java +++ b/src/main/java/net/openhft/chronicle/set/DummyValueData.java @@ -21,9 +21,9 @@ import net.openhft.chronicle.hash.AbstractData; import net.openhft.chronicle.map.impl.stage.data.ZeroBytesStore; import org.jetbrains.annotations.Nullable; -class DummyValueData extends AbstractData { +public class DummyValueData extends AbstractData { - static final DummyValueData INSTANCE = new DummyValueData(); + public static final DummyValueData INSTANCE = new DummyValueData(); private DummyValueData() {} diff --git a/src/main/java/net/openhft/chronicle/set/SetAbsentEntry.java b/src/main/java/net/openhft/chronicle/set/SetAbsentEntry.java index 78b9c883c..d4453d807 100644 --- a/src/main/java/net/openhft/chronicle/set/SetAbsentEntry.java +++ b/src/main/java/net/openhft/chronicle/set/SetAbsentEntry.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.set; import net.openhft.chronicle.hash.HashAbsentEntry; +import org.jetbrains.annotations.NotNull; /** * Low-level operational context for the situations, when the new key is going to be inserted @@ -28,6 +29,7 @@ import net.openhft.chronicle.hash.HashAbsentEntry; * @see SetQueryContext#absentEntry() */ public interface SetAbsentEntry extends HashAbsentEntry { + @NotNull @Override SetContext context(); diff --git a/src/main/java/net/openhft/chronicle/set/SetContext.java b/src/main/java/net/openhft/chronicle/set/SetContext.java index cc2e6491d..d0e026e0f 100644 --- a/src/main/java/net/openhft/chronicle/set/SetContext.java +++ b/src/main/java/net/openhft/chronicle/set/SetContext.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.set; +import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.chronicle.hash.HashContext; /** @@ -32,9 +33,7 @@ public interface SetContext extends HashContext, SetEntryOperations hash() { - return set(); - } + ChronicleHash hash(); /** * Returns the accessed {@code ChronicleSet}. Synonym to {@link #hash()}. diff --git a/src/main/java/net/openhft/chronicle/set/SetFromMap.java b/src/main/java/net/openhft/chronicle/set/SetFromMap.java index 21e39a434..f158d99b6 100644 --- a/src/main/java/net/openhft/chronicle/set/SetFromMap.java +++ b/src/main/java/net/openhft/chronicle/set/SetFromMap.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.set; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.map.ChronicleMap; +import net.openhft.chronicle.map.VanillaChronicleMap; import org.jetbrains.annotations.NotNull; import java.io.File; @@ -32,14 +33,20 @@ import java.util.function.Predicate; import static net.openhft.chronicle.set.DummyValue.DUMMY_VALUE; +/** + * ChronicleSet is implemented through ChronicleMap, with dummy 0-bytes values. This solution + * trades correctness of abstractions for simplicity and minimizing changes before production 3.x + * release. + */ class SetFromMap extends AbstractSet implements ChronicleSet, Serializable { private final ChronicleMap m; // The backing map private transient Set s; // Its keySet - SetFromMap(ChronicleMap map) { + SetFromMap(VanillaChronicleMap map) { m = map; + map.chronicleSet = this; s = map.keySet(); } diff --git a/src/main/java/net/openhft/chronicle/set/SetInstanceBuilder.java b/src/main/java/net/openhft/chronicle/set/SetInstanceBuilder.java index a23911722..a36dfcbe4 100644 --- a/src/main/java/net/openhft/chronicle/set/SetInstanceBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/SetInstanceBuilder.java @@ -21,6 +21,7 @@ import net.openhft.chronicle.hash.replication.ReplicationChannel; import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.map.ChronicleMap; +import net.openhft.chronicle.map.VanillaChronicleMap; import java.io.File; import java.io.IOException; @@ -69,7 +70,7 @@ final class SetInstanceBuilder implements ChronicleHashInstanceBuilder create() throws IOException { try { - return new SetFromMap<>(mapBuilder.create()); + return new SetFromMap<>((VanillaChronicleMap) mapBuilder.create()); } catch (IllegalStateException e) { if (e.getMessage() != null && e.getMessage().startsWith("A ChronicleMap")) { throw new IllegalStateException( diff --git a/src/main/java/net/openhft/chronicle/set/replication/SetRemoteOperations.java b/src/main/java/net/openhft/chronicle/set/replication/SetRemoteOperations.java index 4fdf6f855..b69c67423 100644 --- a/src/main/java/net/openhft/chronicle/set/replication/SetRemoteOperations.java +++ b/src/main/java/net/openhft/chronicle/set/replication/SetRemoteOperations.java @@ -55,7 +55,7 @@ public interface SetRemoteOperations { } } - default void insert(SetRemoteQueryContext q) { + default void put(SetRemoteQueryContext q) { SetReplicableEntry entry = q.entry(); if (entry != null) { if (decideOnRemoteModification(entry, q) == ACCEPT) { diff --git a/src/test/java/net/openhft/chronicle/set/SetEntryOperationsTest.java b/src/test/java/net/openhft/chronicle/set/SetEntryOperationsTest.java new file mode 100644 index 000000000..9d979f6d9 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/set/SetEntryOperationsTest.java @@ -0,0 +1,60 @@ +/* + * 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.set; + +import org.jetbrains.annotations.NotNull; +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.atomic.AtomicInteger; + +public class SetEntryOperationsTest { + + @Test + public void setEntryOperationsTest() { + AtomicInteger insertCounter = new AtomicInteger(); + AtomicInteger remoteCounter = new AtomicInteger(); + try (ChronicleSet fruits = ChronicleSet + .of(String.class) + .entries(3) + .averageKey("apple") + .entryOperations(new SetEntryOperations() { + @Override + public Void remove(@NotNull SetEntry entry) { + remoteCounter.addAndGet(1); + entry.doRemove(); + return null; + } + + @Override + public Void insert(@NotNull SetAbsentEntry entry) { + insertCounter.addAndGet(1); + entry.doInsert(); + return null; + } + }) + .create()) { + fruits.add("apple"); + fruits.add("banana"); + fruits.remove("banana"); + fruits.remove("grapes"); + + Assert.assertEquals(2, insertCounter.get()); + Assert.assertEquals(1, remoteCounter.get()); + } + } +} diff --git a/src/test/java/net/openhft/chronicle/set/SetRemoteOperationsTest.java b/src/test/java/net/openhft/chronicle/set/SetRemoteOperationsTest.java new file mode 100644 index 000000000..3462b7c51 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/set/SetRemoteOperationsTest.java @@ -0,0 +1,239 @@ +/* + * 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.set; + +import net.openhft.chronicle.hash.replication.ReplicableEntry; +import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; +import net.openhft.chronicle.map.ChannelReplicationTest; +import net.openhft.chronicle.set.replication.SetRemoteOperations; +import net.openhft.chronicle.set.replication.SetRemoteQueryContext; +import net.openhft.chronicle.set.replication.SetReplicableEntry; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.io.Closeable; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import static net.openhft.chronicle.hash.replication.AcceptanceDecision.ACCEPT; +import static net.openhft.chronicle.hash.replication.AcceptanceDecision.DISCARD; +import static net.openhft.chronicle.hash.replication.DefaultEventualConsistencyStrategy.decideOnRemoteModification; +import static org.junit.Assert.assertEquals; + +public class SetRemoteOperationsTest { + + static int s_port = 12050; + private ChronicleSet set1; + private ChronicleSet set2; + private AtomicInteger set1PutCounter; + private AtomicInteger set2PutCounter; + private AtomicInteger set1RemoveCounter; + private AtomicInteger set2RemoveCounter; + + @Before + public void setup() throws IOException { + int port = s_port; + set1PutCounter = new AtomicInteger(); + set1RemoveCounter = new AtomicInteger(); + ChronicleSetBuilder set1Builder = ChronicleSet + .of(Integer.class) + .remoteOperations(new SetRemoteOperations() { + @Override + public void remove(SetRemoteQueryContext q) { + SetReplicableEntry entry = q.entry(); + if (entry != null) { + if (decideOnRemoteModification(entry, q) == ACCEPT) { + q.remove(entry); + ReplicableEntry replicableAbsentEntry = + (ReplicableEntry) q.absentEntry(); + replicableAbsentEntry + .updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); + replicableAbsentEntry.dropChanged(); + set1RemoveCounter.addAndGet(1); + } + } else { + SetAbsentEntry absentEntry = q.absentEntry(); + ReplicableEntry replicableAbsentEntry; + if (!(absentEntry instanceof ReplicableEntry)) { + absentEntry.doInsert(); + q.entry().doRemove(); + replicableAbsentEntry = (ReplicableEntry) q.absentEntry(); + } else { + replicableAbsentEntry = (ReplicableEntry) absentEntry; + if (decideOnRemoteModification(replicableAbsentEntry, q) == DISCARD) + return; + } + replicableAbsentEntry + .updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); + replicableAbsentEntry.dropChanged(); + set1RemoveCounter.addAndGet(1); + } + } + + @Override + public void put(SetRemoteQueryContext q) { + SetReplicableEntry entry = q.entry(); + if (entry != null) { + if (decideOnRemoteModification(entry, q) == ACCEPT) { + entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); + entry.dropChanged(); + set1PutCounter.addAndGet(1); + } + } else { + SetAbsentEntry absentEntry = q.absentEntry(); + assert absentEntry != null; + if (!(absentEntry instanceof ReplicableEntry) || + decideOnRemoteModification((ReplicableEntry) absentEntry, q) == + ACCEPT) { + q.insert(absentEntry); + entry = q.entry(); // q.entry() is not null after insert + assert entry != null; + entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); + entry.dropChanged(); + set1PutCounter.addAndGet(1); + } + } + } + }) + .replication((byte) 1, TcpTransportAndNetworkConfig + .of(port, new InetSocketAddress("localhost", port + 1))); + set1 = set1Builder.entries(Builder.SIZE).create(); + + set2PutCounter = new AtomicInteger(); + set2RemoveCounter = new AtomicInteger(); + ChronicleSetBuilder set2Builder = ChronicleSet + .of(Integer.class) + .remoteOperations(new SetRemoteOperations() { + @Override + public void remove(SetRemoteQueryContext q) { + SetReplicableEntry entry = q.entry(); + if (entry != null) { + if (decideOnRemoteModification(entry, q) == ACCEPT) { + q.remove(entry); + ReplicableEntry replicableAbsentEntry = + (ReplicableEntry) q.absentEntry(); + replicableAbsentEntry + .updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); + replicableAbsentEntry.dropChanged(); + set2RemoveCounter.addAndGet(1); + } + } else { + SetAbsentEntry absentEntry = q.absentEntry(); + ReplicableEntry replicableAbsentEntry; + if (!(absentEntry instanceof ReplicableEntry)) { + absentEntry.doInsert(); + q.entry().doRemove(); + replicableAbsentEntry = (ReplicableEntry) q.absentEntry(); + } else { + replicableAbsentEntry = (ReplicableEntry) absentEntry; + if (decideOnRemoteModification(replicableAbsentEntry, q) == DISCARD) + return; + } + replicableAbsentEntry + .updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); + replicableAbsentEntry.dropChanged(); + set2RemoveCounter.addAndGet(1); + } + } + + @Override + public void put(SetRemoteQueryContext q) { + SetReplicableEntry entry = q.entry(); + if (entry != null) { + if (decideOnRemoteModification(entry, q) == ACCEPT) { + entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); + entry.dropChanged(); + set2PutCounter.addAndGet(1); + } + } else { + SetAbsentEntry absentEntry = q.absentEntry(); + assert absentEntry != null; + if (!(absentEntry instanceof ReplicableEntry) || + decideOnRemoteModification((ReplicableEntry) absentEntry, q) == + ACCEPT) { + q.insert(absentEntry); + entry = q.entry(); // q.entry() is not null after insert + assert entry != null; + entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp()); + entry.dropChanged(); + set2PutCounter.addAndGet(1); + } + } + } + }) + .replication((byte) 2, TcpTransportAndNetworkConfig.of(port + 1)); + set2 = set2Builder.entries(Builder.SIZE).create(); + s_port += 2; + } + + @After + public void tearDown() throws InterruptedException { + + for (final Closeable closeable : new Closeable[]{set1, set2}) { + try { + closeable.close(); + } catch (Exception e) { + e.printStackTrace(); + } + } + System.gc(); + } + + Set threads; + + @Before + public void sampleThreads() { + threads = Thread.getAllStackTraces().keySet(); + } + + @After + public void checkThreadsShutdown() { + ChannelReplicationTest.checkThreadsShutdown(threads); + } + + @Test + public void setRemoteOperationsTest() throws InterruptedException { + set1.add(1); + set2.add(2); + + waitTillEqual(5000); + assertEquals(set1, set2); + + set1.remove(2); + set2.remove(1); + + waitTillEqual(5000); + assertEquals(set1, set2); + + assertEquals(1, set1PutCounter.get()); + assertEquals(1, set2PutCounter.get()); + assertEquals(1, set1RemoveCounter.get()); + assertEquals(1, set2RemoveCounter.get()); + } + + private void waitTillEqual(final int timeOutMs) throws InterruptedException { + int t = 0; + for (; t < timeOutMs; t++) { + if (set1.equals(set2)) + break; + Thread.sleep(1); + } + } +}