Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Add ChronicleSetBuilder.entryOperations() and .remoteOperations() interception configuration methods; Rename SetRemoteOperations.insert() to put(), for consistency with MapRemoteOperations

Browse Source
This commit is contained in:
leventov
2015-11-28 23:20:29 +03:00
parent ec3bd9447e
commit bfa9988551
30 changed files with 464 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pom.xml
+2 -1
View File
@@ -152,7 +152,7 @@
<dependency>
<groupId>net.openhft</groupId>
<artifactId>stage-compiler</artifactId>
<version>1.7</version>
<version>1.8</version>
<scope>provided</scope>
</dependency>
@@ -238,6 +238,7 @@
<exclude>net/openhft/chronicle/map/VanillaChronicleMap.java</exclude>
<exclude>net/openhft/chronicle/map/Replica.java</exclude>
<exclude>net/openhft/chronicle/map/ReplicatedChronicleMap.java</exclude>
<exclude>net/openhft/chronicle/set/DummyValueData.java</exclude>
</sourceFileExcludes>
<failOnError>false</failOnError>
<additionalparam>-Xdoclint:none</additionalparam>
src/main/java/net/openhft/chronicle/hash/impl/stage/query/HashQuery.java
+2 -1
View File
@@ -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<K> implements HashEntry<K> {
public abstract class HashQuery<K> implements SetEntry<K> {
@StageRef public VanillaChronicleHashHolder<K> hh;
@StageRef public SegmentStages s;
src/main/java/net/openhft/chronicle/map/MapContext.java
+2 -3
View File
@@ -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<K, V, R>
* Returns the accessed {@code ChronicleMap}. Synonym to {@link #map()}.
*/
@Override
default ChronicleMap<K, V> hash() {
return map();
}
ChronicleHash<K, ?, ?, ?> hash();
/**
* Returns the accessed {@code ChronicleMap}. Synonym to {@link #hash()}.
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+4
View File
@@ -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<K, V, R>
transient boolean removeReturnsNull;
transient Set<Entry<K, V>> entrySet;
/** @see net.openhft.chronicle.set.SetFromMap */
public transient ChronicleSet<K> chronicleSet;
public transient MapEntryOperations<K, V, R> entryOperations;
public transient MapMethods<K, V, R> methods;
src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java
-1
View File
@@ -58,7 +58,6 @@ import net.openhft.sg.Staged;
MapAbsent.class,
DefaultValue.class,
ValueBytesInterop.class,
MapAbsentHolder.class,
QueryAlloc.class,
},
src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java
-1
View File
@@ -67,7 +67,6 @@ import net.openhft.sg.Staged;
ReplicatedMapAbsent.class,
DefaultValue.class,
ValueBytesInterop.class,
ReplicatedMapAbsentHolder.class,
ReplicationUpdate.class,
ReplicatedInput.class,
src/main/java/net/openhft/chronicle/map/impl/VanillaChronicleMapHolder.java
+3
View File
@@ -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<K, V, R> extends VanillaChronicleHashHolder<K> {
@@ -33,4 +34,6 @@ public interface VanillaChronicleMapHolder<K, V, R> extends VanillaChronicleHash
default ChronicleMap<K, V> map() {
return m();
}
ChronicleSet<K> set();
}
src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java
+1 -1
View File
@@ -47,7 +47,7 @@ public abstract class MapSegmentIteration<K, V, R> extends HashSegmentIteration<
@NotNull
@Override
public MapContext<K, V, ?> context() {
public WrappedValueInstanceValueHolder<K, V, ?> context() {
return wrappedValueInstanceValueHolder;
}
}
src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapAbsentDelegatingForIteration.java
+9 -3
View File
@@ -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<K, V>
implements MapAbsentEntry<K, V>, ReplicableEntryDelegating {
implements MapAbsentEntry<K, V>, SetAbsentEntry<K>, ReplicableEntryDelegating {
@StageRef ReplicatedMapSegmentIteration<K, V, ?> delegate;
@StageRef ReplicatedMapEntryStages<K, V> e;
@NotNull
@Override
public MapContext<K, V, ?> context() {
public WrappedValueInstanceValueHolder<K, V, ?> context() {
return delegate.context();
}
@@ -44,6 +45,11 @@ public class ReplicatedMapAbsentDelegatingForIteration<K, V>
delegate.doInsert(value);
}
@Override
public void doInsert() {
delegate.doInsert();
}
@NotNull
@Override
public Data<V> defaultValue() {
src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapSegmentIteration.java
+13 -4
View File
@@ -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<K, V, R> extends MapSegmentIteration<K, V, R>
implements ReplicatedIterationContext<K, V, R>, ReplicableEntry,
ReplicatedHashSegmentContext<K, MapEntry<K, V>>, MapAbsentEntry<K, V> {
ReplicatedHashSegmentContext<K, MapEntry<K, V>>, MapAbsentEntry<K, V>, SetAbsentEntry<K> {
@StageRef VanillaChronicleHashHolder<?> hh;
@StageRef VanillaChronicleMapHolder<K, V, R> mh;
@StageRef ReplicatedMapEntryStages<K, V> e;
@StageRef ReplicationUpdate<K> ru;
@StageRef DummyValueZeroData<V> dummyValue;
@@ -73,7 +75,7 @@ public abstract class ReplicatedMapSegmentIteration<K, V, R> 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<K, V, R> 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<V>) DummyValueData.INSTANCE);
}
}
src/main/java/net/openhft/chronicle/map/impl/stage/map/ReplicatedChronicleMapHolderImpl.java
+11
View File
@@ -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<K, V, R> extends Chaining
public ChronicleMap<K, V> map() {
return m();
}
@Override
public ChronicleSet<K> set() {
return m.chronicleSet;
}
public ChronicleHash<K, ?, ?, ?> hash() {
return set() != null ? set() : map();
}
}
src/main/java/net/openhft/chronicle/map/impl/stage/map/VanillaChronicleMapHolderImpl.java
+11
View File
@@ -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<K, V, R> extends Chaining
public ChronicleMap<K, V> map() {
return m();
}
@Override
public ChronicleSet<K> set() {
return m.chronicleSet;
}
public ChronicleHash<K, ?, ?, ?> hash() {
return set() != null ? set() : map();
}
}
src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueInstanceValueHolder.java
+3 -1
View File
@@ -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<K, V, R> implements MapContext<K, V, R> {
public abstract class WrappedValueInstanceValueHolder<K, V, R>
implements MapContext<K, V, R>, SetContext<K, R> {
@StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
@StageRef
src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsentHolder.java → src/main/java/net/openhft/chronicle/map/impl/stage/query/Absent.java
+5 -10
View File
@@ -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<K, V> implements MapAbsentEntryHolder<K, V> {
@StageRef MapAbsent<K, V> ab;
public interface Absent<K, V> extends MapAbsentEntry<K, V>, SetAbsentEntry<K> {
@NotNull
@Override
public MapAbsentEntry<K, V> absent() {
return ab;
}
MapAndSetContext<K, V, ?> context();
}
src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAbsent.java
+13 -3
View File
@@ -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<K, V> implements MapAbsentEntry<K, V> {
public abstract class MapAbsent<K, V> implements Absent<K, V> {
@StageRef public KeySearch<K> ks;
@StageRef MapQuery<K, V, ?> q;
@@ -40,6 +42,7 @@ public abstract class MapAbsent<K, V> implements MapAbsentEntry<K, V> {
@StageRef public HashLookupSearch hashLookupSearch;
@StageRef public CheckOnEachPublicOperation checkOnEachPublicOperation;
@StageRef public SegmentStages s;
@StageRef VanillaChronicleMapHolder<K, V, ?> mh;
void putEntry(Data<V> value) {
assert ks.searchStateAbsent();
@@ -52,7 +55,7 @@ public abstract class MapAbsent<K, V> implements MapAbsentEntry<K, V> {
@NotNull
@Override
public MapContext<K, V, ?> context() {
public MapQuery<K, V, ?> context() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return q;
}
@@ -82,4 +85,11 @@ public abstract class MapAbsent<K, V> implements MapAbsentEntry<K, V> {
"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<V>) DummyValueData.INSTANCE);
}
}
src/main/java/net/openhft/chronicle/map/impl/MapAbsentEntryHolder.java → src/main/java/net/openhft/chronicle/map/impl/stage/query/MapAndSetContext.java
+4 -5
View File
@@ -14,11 +14,10 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
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<K, V> {
MapAbsentEntry<K, V> absent();
public interface MapAndSetContext<K, V, R> extends MapContext<K, V, R>, SetContext<K, R> {
}
src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java
+7 -9
View File
@@ -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<K, V, R> extends HashQuery<K>
implements MapEntry<K, V>, ExternalMapQueryContext<K, V, R>,
QueryContextInterface<K, V, R> {
ExternalSetQueryContext<K, R>, QueryContextInterface<K, V, R>, MapAndSetContext<K, V, R> {
@StageRef VanillaChronicleMapHolder<K, V, R> mh;
@StageRef MapEntryStages<K, V> e;
@@ -53,7 +51,7 @@ public abstract class MapQuery<K, V, R> extends HashQuery<K>
@StageRef public DefaultReturnValue<V> defaultReturnValue;
@StageRef public UsingReturnValue<V> usingReturnValue;
@StageRef public MapAbsentEntryHolder<K, V> absent;
@StageRef public MapAbsent<K, V> absent;
final DataAccess<V> innerInputValueDataAccess = mh.m().valueDataAccess.copy();
@@ -76,16 +74,16 @@ public abstract class MapQuery<K, V, R> extends HashQuery<K>
}
@Override
public MapEntry<K, V> entry() {
public MapQuery<K, V, R> entry() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return entryPresent() ? this : null;
}
@Nullable
@Override
public MapAbsentEntry<K, V> absentEntry() {
public Absent<K, V> absentEntry() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return entryPresent() ? null : absent.absent();
return entryPresent() ? null : absent;
}
protected void putPrefix() {
@@ -120,7 +118,7 @@ public abstract class MapQuery<K, V, R> extends HashQuery<K>
@NotNull
@Override
public MapContext<K, V, ?> context() {
public MapQuery<K, V, R> context() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return this;
}
src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsentDelegating.java
+7 -4
View File
@@ -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<K, V> implements MapAbsentEntry<K, V> {
public class ReplicatedMapAbsentDelegating<K, V> implements Absent<K, V> {
@StageRef ReplicatedMapAbsent<K, V> delegate;
@NotNull
@Override
public MapContext<K, V, ?> context() {
public MapQuery<K, V, ?> context() {
return delegate.context();
}
@Override
public void doInsert() {
delegate.doInsert();
}
@Override
public void doInsert(Data<V> value) {
delegate.doInsert(value);
src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapAbsentHolder.java
-33
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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<K, V> implements MapAbsentEntryHolder<K, V> {
@StageRef ReplicatedMapAbsent<K, V> ab;
@Override
public MapAbsentEntry<K, V> absent() {
return ab;
}
}
src/main/java/net/openhft/chronicle/map/impl/stage/query/ReplicatedMapQuery.java
+9 -6
View File
@@ -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<K, V, R> extends MapQuery<K, V, R>
implements MapRemoteQueryContext<K, V, R>, ReplicableEntry {
implements MapRemoteQueryContext<K, V, R>, SetRemoteQueryContext<K, R>,
ReplicableEntry, MapReplicableEntry<K, V>, SetReplicableEntry<K> {
@StageRef ReplicatedMapEntryStages<K, V> e;
@StageRef ReplicationUpdate ru;
@@ -39,7 +41,7 @@ public abstract class ReplicatedMapQuery<K, V, R> extends MapQuery<K, V, R>
@Nullable
@Override
public MapAbsentEntry<K, V> absentEntry() {
public Absent<K, V> absentEntry() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
if (entryPresent()) {
return null;
@@ -48,7 +50,7 @@ public abstract class ReplicatedMapQuery<K, V, R> extends MapQuery<K, V, R>
return absentDelegating;
} else {
assert e.entryDeleted();
return absent.absent();
return absent;
}
}
}
@@ -61,8 +63,9 @@ public abstract class ReplicatedMapQuery<K, V, R> extends MapQuery<K, V, R>
@StageRef DummyValueZeroData<V> dummyValue;
@Override
public MapReplicableEntry<K, V> entry() {
return (MapReplicableEntry<K, V>) super.entry();
public ReplicatedMapQuery<K, V, R> entry() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return entryPresent() ? this : null;
}
@Override
src/main/java/net/openhft/chronicle/map/replication/MapRemoteQueryContext.java
+2 -1
View File
@@ -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<K, V, R> extends MapQueryContext<K, V, R>
/**
* 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}.
*
src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
+40 -4
View File
@@ -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<K>
return this;
}
public ChronicleSetBuilder<K> entryOperations(SetEntryOperations<K, ?> entryOperations) {
chronicleMapBuilder.entryOperations(new MapEntryOperations<K, DummyValue, Object>() {
@Override
public Object remove(@NotNull MapEntry<K, DummyValue> entry) {
return entryOperations.remove((SetEntry<K>) entry);
}
@Override
public Object insert(
@NotNull MapAbsentEntry<K, DummyValue> absentEntry, Data<DummyValue> value) {
return entryOperations.insert((SetAbsentEntry<K>) absentEntry);
}
});
return this;
}
public ChronicleSetBuilder<K> remoteOperations(SetRemoteOperations<K, ?> remoteOperations) {
chronicleMapBuilder.remoteOperations(new MapRemoteOperations<K, DummyValue, Object>() {
@Override
public void remove(MapRemoteQueryContext<K, DummyValue, Object> q) {
remoteOperations.remove((SetRemoteQueryContext) q);
}
@Override
public void put(
MapRemoteQueryContext<K, DummyValue, Object> q, Data<DummyValue> newValue) {
remoteOperations.put((SetRemoteQueryContext) q);
}
});
return this;
}
@Override
public ChronicleHashInstanceBuilder<ChronicleSet<K>> instance() {
return new SetInstanceBuilder<>(chronicleMapBuilder.instance());
@@ -295,13 +331,13 @@ public final class ChronicleSetBuilder<K>
@Override
public ChronicleSet<K> create() {
final ChronicleMap<K, DummyValue> map = chronicleMapBuilder.create();
return new SetFromMap<>(map);
return new SetFromMap<>((VanillaChronicleMap<K, DummyValue, ?>) map);
}
@Override
public ChronicleSet<K> createPersistedTo(File file) throws IOException {
ChronicleMap<K, DummyValue> map = chronicleMapBuilder.createPersistedTo(file);
return new SetFromMap<>(map);
return new SetFromMap<>((VanillaChronicleMap<K, DummyValue, ?>) map);
}
/**
src/main/java/net/openhft/chronicle/set/DummyValueData.java
+2 -2
View File
@@ -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<DummyValue> {
public class DummyValueData extends AbstractData<DummyValue> {
static final DummyValueData INSTANCE = new DummyValueData();
public static final DummyValueData INSTANCE = new DummyValueData();
private DummyValueData() {}
src/main/java/net/openhft/chronicle/set/SetAbsentEntry.java
+2
View File
@@ -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<K> extends HashAbsentEntry<K> {
@NotNull
@Override
SetContext<K, ?> context();
src/main/java/net/openhft/chronicle/set/SetContext.java
+2 -3
View File
@@ -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<K, R> extends HashContext<K>, SetEntryOperations<K,
* Returns the accessed {@code ChronicleSet}. Synonym to {@link #set()}.
*/
@Override
default ChronicleSet<K> hash() {
return set();
}
ChronicleHash<K, ?, ?, ?> hash();
/**
* Returns the accessed {@code ChronicleSet}. Synonym to {@link #hash()}.
src/main/java/net/openhft/chronicle/set/SetFromMap.java
+8 -1
View File
@@ -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<E> extends AbstractSet<E>
implements ChronicleSet<E>, Serializable {
private final ChronicleMap<E, DummyValue> m; // The backing map
private transient Set<E> s; // Its keySet
SetFromMap(ChronicleMap<E, DummyValue> map) {
SetFromMap(VanillaChronicleMap<E, DummyValue, ?> map) {
m = map;
map.chronicleSet = this;
s = map.keySet();
}
src/main/java/net/openhft/chronicle/set/SetInstanceBuilder.java
+2 -1
View File
@@ -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<E> implements ChronicleHashInstanceBuilder<Chroni
@Override
public synchronized ChronicleSet<E> create() throws IOException {
try {
return new SetFromMap<>(mapBuilder.create());
return new SetFromMap<>((VanillaChronicleMap<E, DummyValue, ?>) mapBuilder.create());
} catch (IllegalStateException e) {
if (e.getMessage() != null && e.getMessage().startsWith("A ChronicleMap")) {
throw new IllegalStateException(
src/main/java/net/openhft/chronicle/set/replication/SetRemoteOperations.java
+1 -1
View File
@@ -55,7 +55,7 @@ public interface SetRemoteOperations<K, R> {
}
}
default void insert(SetRemoteQueryContext<K, R> q) {
default void put(SetRemoteQueryContext<K, R> q) {
SetReplicableEntry<K> entry = q.entry();
if (entry != null) {
if (decideOnRemoteModification(entry, q) == ACCEPT) {
src/test/java/net/openhft/chronicle/set/SetEntryOperationsTest.java
+60
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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<String> fruits = ChronicleSet
.of(String.class)
.entries(3)
.averageKey("apple")
.entryOperations(new SetEntryOperations<String, Void>() {
@Override
public Void remove(@NotNull SetEntry<String> entry) {
remoteCounter.addAndGet(1);
entry.doRemove();
return null;
}
@Override
public Void insert(@NotNull SetAbsentEntry<String> 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());
}
}
}
src/test/java/net/openhft/chronicle/set/SetRemoteOperationsTest.java
+239
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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<Integer> set1;
private ChronicleSet<Integer> 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<Integer> set1Builder = ChronicleSet
.of(Integer.class)
.remoteOperations(new SetRemoteOperations<Integer, Void>() {
@Override
public void remove(SetRemoteQueryContext<Integer, Void> q) {
SetReplicableEntry<Integer> 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<Integer> 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<Integer, Void> q) {
SetReplicableEntry<Integer> entry = q.entry();
if (entry != null) {
if (decideOnRemoteModification(entry, q) == ACCEPT) {
entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
entry.dropChanged();
set1PutCounter.addAndGet(1);
}
} else {
SetAbsentEntry<Integer> 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<Integer> set2Builder = ChronicleSet
.of(Integer.class)
.remoteOperations(new SetRemoteOperations<Integer, Void>() {
@Override
public void remove(SetRemoteQueryContext<Integer, Void> q) {
SetReplicableEntry<Integer> 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<Integer> 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<Integer, Void> q) {
SetReplicableEntry<Integer> entry = q.entry();
if (entry != null) {
if (decideOnRemoteModification(entry, q) == ACCEPT) {
entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
entry.dropChanged();
set2PutCounter.addAndGet(1);
}
} else {
SetAbsentEntry<Integer> 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<Thread> 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);
}
}
}