diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java index 48674a45f..67a8279c3 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHash.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.hash; +import net.openhft.chronicle.bytes.BytesStore; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.MapMethods; import net.openhft.chronicle.map.MapQueryContext; @@ -80,6 +81,18 @@ public interface ChronicleHash, SC extends HashSegment */ @NotNull EQC queryContext(Data key); + /** + * Returns the context to perform arbitrary operations with the given key, provided in the + * serialized form. See {@link #queryContext(Object)} and {@link #queryContext(Data)} for more + * information on contexts semantics and usage patterns. + * + * @param keyBytes the bytes store with the key bytes to query + * @param offset actual offset of the key bytes within the given BytesStore + * @param size length of the key bytes sequence + * @return the context to perform operations with the key + */ + @NotNull EQC queryContext(BytesStore keyBytes, long offset, long size); + /** * Returns the context of the segment with the given index. Segments are indexed from 0 to * {@link #segments()}{@code - 1}. diff --git a/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/InputKeyBytesData.java b/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/InputKeyBytesData.java new file mode 100644 index 000000000..7f51b4cb7 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/InputKeyBytesData.java @@ -0,0 +1,105 @@ +/* + * 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.data.bytes; + +import net.openhft.chronicle.bytes.*; +import net.openhft.chronicle.hash.AbstractData; +import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; +import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop; +import net.openhft.sg.Stage; +import net.openhft.sg.StageRef; +import net.openhft.sg.Staged; + +import static net.openhft.chronicle.bytes.NoBytesStore.NO_BYTES_STORE; + +@Staged +public class InputKeyBytesData extends AbstractData { + + @StageRef KeyBytesInterop ki; + @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation; + + @Stage("InputKeyBytesStore") private BytesStore inputKeyBytesStore = null; + @Stage("InputKeyBytesStore") private long inputKeyBytesOffset; + @Stage("InputKeyBytesStore") private long inputKeyBytesSize; + + public void initInputKeyBytesStore(BytesStore bytesStore, long offset, long size) { + inputKeyBytesStore = bytesStore; + inputKeyBytesOffset = offset; + inputKeyBytesSize = size; + } + + @Stage("InputKeyBytes") private final VanillaBytes inputKeyBytes = + new VanillaBytes(NO_BYTES_STORE); + @Stage("InputKeyBytes") private boolean inputKeyBytesUsed = false; + + boolean inputKeyBytesInit() { + return inputKeyBytesUsed; + } + + void initInputKeyBytes() { + inputKeyBytes.bytesStore(inputKeyBytesStore, inputKeyBytesOffset, inputKeyBytesSize); + inputKeyBytesUsed = true; + } + + void closeInputKeyBytes() { + inputKeyBytes.bytesStore(NO_BYTES_STORE, 0, 0); + inputKeyBytesUsed = false; + } + + @Stage("CachedInputKey") private K cachedInputKey; + @Stage("CachedInputKey") private boolean cachedInputKeyRead = false; + + private void initCachedInputKey() { + cachedInputKey = innerGetUsing(cachedInputKey); + cachedInputKeyRead = true; + } + + @Override + public RandomDataInput bytes() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return inputKeyBytes.bytesStore(); + } + + @Override + public long offset() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return inputKeyBytesOffset; + } + + @Override + public long size() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return inputKeyBytesSize; + } + + @Override + public K get() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return cachedInputKey; + } + + @Override + public K getUsing(K using) { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return innerGetUsing(using); + } + + private K innerGetUsing(K usingKey) { + inputKeyBytes.readPosition(inputKeyBytesOffset); + return ki.keyReader.read(inputKeyBytes, inputKeyBytesSize, usingKey); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/MapContext.java b/src/main/java/net/openhft/chronicle/map/MapContext.java index 009edac5e..8ac682e4b 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.bytes.BytesStore; import net.openhft.chronicle.hash.ChronicleHash; import net.openhft.chronicle.hash.HashContext; import net.openhft.chronicle.hash.Data; @@ -58,4 +59,18 @@ public interface MapContext * @return the value as {@code Data} */ Data wrapValueAsData(V value); + + /** + * Wraps the given value bytes as a {@code Data}. Useful when you need to pass a value + * to some method accepting {@code Data}, for example, {@link MapEntryOperations#replaceValue( + * MapEntry, Data)}, without allocating manual deserialization and {@code ThreadLocals}. + * + *

The returned {@code Data} object shouldn't outlive this {@code MapContext}. + * + * @param valueBytes the value bytes to wrap + * @param offset offset within the given valueBytes, the actual value bytes start from + * @param size length of the value bytes sequence + * @return the value bytes as {@code Data} + */ + Data wrapValueBytesAsData(BytesStore valueBytes, long offset, long size); } diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 55083e4f7..5faee4264 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.Bytes; +import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.replication.ReplicableEntry; import org.jetbrains.annotations.NotNull; @@ -27,6 +28,13 @@ import java.io.Closeable; */ public interface Replica extends Closeable { + interface QueryContext { + + void remotePut(Data newValue, byte remoteIdentifier, long timestamp); + + void remoteRemove(byte remoteIdentifier, long timestamp); + } + /** * Provides the unique Identifier associated with this map instance.

An identifier is used * to determine which replicating node made the change.

If two nodes update their map at the diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index ee7273afc..a1f8aee5a 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -17,10 +17,12 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.bytes.Bytes; +import net.openhft.chronicle.bytes.BytesStore; import net.openhft.chronicle.core.io.Closeable; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.VanillaChronicleHash; import net.openhft.chronicle.hash.impl.stage.hash.ChainingInterface; +import net.openhft.chronicle.hash.impl.util.Objects; import net.openhft.chronicle.hash.serialization.DataAccess; import net.openhft.chronicle.hash.serialization.SizeMarshaller; import net.openhft.chronicle.hash.serialization.SizedReader; @@ -291,6 +293,16 @@ public class VanillaChronicleMap return q; } + @NotNull + @Override + public ExternalMapQueryContext queryContext( + BytesStore keyBytes, long offset, long size) { + Objects.requireNonNull(keyBytes); + QueryContextInterface q = mapContext(); + q.initInputKey(q.getInputKeyBytesAsData(keyBytes, offset, size)); + return q; + } + @Override public MapSegmentContext segmentContext(int segmentIndex) { IterationContext c = iterationContext(); diff --git a/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java b/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java index 85e560197..26fccc0aa 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java @@ -26,14 +26,12 @@ import net.openhft.chronicle.hash.impl.stage.iter.IterationKeyHashCode; import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages; import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData; +import net.openhft.chronicle.map.impl.stage.data.bytes.WrappedValueBytesData; import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceDataHolder; import net.openhft.chronicle.map.impl.stage.entry.MapEntryStages; import net.openhft.chronicle.map.impl.stage.iter.IterationCheckOnEachPublicOperation; import net.openhft.chronicle.map.impl.stage.iter.MapSegmentIteration; -import net.openhft.chronicle.map.impl.stage.map.MapEntryOperationsDelegation; -import net.openhft.chronicle.map.impl.stage.map.ValueBytesInterop; -import net.openhft.chronicle.map.impl.stage.map.VanillaChronicleMapHolderImpl; -import net.openhft.chronicle.map.impl.stage.map.WrappedValueInstanceValueHolder; +import net.openhft.chronicle.map.impl.stage.map.*; import net.openhft.sg.Context; import net.openhft.sg.Staged; @@ -56,7 +54,8 @@ import net.openhft.sg.Staged; KeySearch.class, HashLookupSearch.class, - WrappedValueInstanceValueHolder.class, + WrappedValueInstanceDataHolderAccess.class, + WrappedValueBytesDataAccess.class, MapEntryStages.class, ValueBytesInterop.class, MapEntryOperationsDelegation.class, @@ -71,6 +70,7 @@ import net.openhft.sg.Staged; EntryValueBytesData.class, WrappedValueInstanceDataHolder.class, + WrappedValueBytesData.class, HashEntryChecksumStrategy.class, }) 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 77cace6a6..31c0146f9 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.map.impl; import net.openhft.chronicle.hash.impl.stage.data.bytes.EntryKeyBytesData; +import net.openhft.chronicle.hash.impl.stage.data.bytes.InputKeyBytesData; import net.openhft.chronicle.hash.impl.stage.entry.*; import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop; import net.openhft.chronicle.hash.impl.stage.hash.LogHolder; @@ -24,6 +25,7 @@ import net.openhft.chronicle.hash.impl.stage.hash.OwnerThreadHolder; import net.openhft.chronicle.hash.impl.stage.query.*; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData; +import net.openhft.chronicle.map.impl.stage.data.bytes.WrappedValueBytesData; import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceDataHolder; import net.openhft.chronicle.map.impl.stage.map.*; import net.openhft.chronicle.map.impl.stage.query.*; @@ -53,7 +55,8 @@ import net.openhft.sg.Staged; QueryMapEntryStages.class, MapEntryOperationsDelegation.class, - WrappedValueInstanceValueHolder.class, + WrappedValueInstanceDataHolderAccess.class, + WrappedValueBytesDataAccess.class, MapQuery.class, MapAbsent.class, DefaultValue.class, @@ -69,7 +72,10 @@ nested = { EntryKeyBytesData.class, EntryValueBytesData.class, + InputKeyBytesData.class, + WrappedValueInstanceDataHolder.class, + WrappedValueBytesData.class, AcquireHandle.class, DefaultReturnValue.class, UsingReturnValue.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/QueryContextInterface.java b/src/main/java/net/openhft/chronicle/map/impl/QueryContextInterface.java index 0d1a1a592..994dc1967 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/QueryContextInterface.java +++ b/src/main/java/net/openhft/chronicle/map/impl/QueryContextInterface.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.map.impl; +import net.openhft.chronicle.bytes.BytesStore; import net.openhft.chronicle.core.io.Closeable; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.serialization.DataAccess; @@ -26,6 +27,8 @@ import net.openhft.chronicle.map.impl.ret.UsableReturnValue; public interface QueryContextInterface extends ExternalMapQueryContext { void initInputKey(Data inputKey); + + Data getInputKeyBytesAsData(BytesStore bytesStore, long offset, long size); DataAccess inputKeyDataAccess(); 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 25bfd30bb..8d6ad5751 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java @@ -27,6 +27,7 @@ import net.openhft.chronicle.hash.impl.stage.iter.IterationSegmentStages; import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData; +import net.openhft.chronicle.map.impl.stage.data.bytes.WrappedValueBytesData; import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceDataHolder; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; import net.openhft.chronicle.map.impl.stage.iter.IterationCheckOnEachPublicOperation; @@ -57,7 +58,8 @@ import net.openhft.sg.Staged; KeySearch.class, HashLookupSearch.class, - WrappedValueInstanceValueHolder.class, + WrappedValueInstanceDataHolderAccess.class, + WrappedValueBytesDataAccess.class, ReplicatedMapEntryStages.class, ValueBytesInterop.class, MapEntryOperationsDelegation.class, @@ -75,6 +77,7 @@ import net.openhft.sg.Staged; EntryValueBytesData.class, WrappedValueInstanceDataHolder.class, + WrappedValueBytesData.class, DummyValueZeroData.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 66f835735..e9b8fa226 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java +++ b/src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java @@ -17,6 +17,7 @@ package net.openhft.chronicle.map.impl; import net.openhft.chronicle.hash.impl.stage.data.bytes.EntryKeyBytesData; +import net.openhft.chronicle.hash.impl.stage.data.bytes.InputKeyBytesData; import net.openhft.chronicle.hash.impl.stage.entry.*; import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop; import net.openhft.chronicle.hash.impl.stage.hash.LogHolder; @@ -29,6 +30,7 @@ import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData; import net.openhft.chronicle.map.impl.stage.data.bytes.ReplicatedInputKeyBytesData; import net.openhft.chronicle.map.impl.stage.data.bytes.ReplicatedInputValueBytesData; +import net.openhft.chronicle.map.impl.stage.data.bytes.WrappedValueBytesData; import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceDataHolder; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; import net.openhft.chronicle.map.impl.stage.input.ReplicatedInput; @@ -62,7 +64,8 @@ import net.openhft.sg.Staged; ReplicatedMapEntryStages.class, MapEntryOperationsDelegation.class, - WrappedValueInstanceValueHolder.class, + WrappedValueInstanceDataHolderAccess.class, + WrappedValueBytesDataAccess.class, ReplicatedMapQuery.class, ReplicatedMapAbsent.class, DefaultValue.class, @@ -80,7 +83,10 @@ nested = { EntryKeyBytesData.class, EntryValueBytesData.class, + InputKeyBytesData.class, + WrappedValueInstanceDataHolder.class, + WrappedValueBytesData.class, AcquireHandle.class, DefaultReturnValue.class, UsingReturnValue.class, diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/WrappedValueBytesData.java b/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/WrappedValueBytesData.java new file mode 100644 index 000000000..9e92c285d --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/WrappedValueBytesData.java @@ -0,0 +1,137 @@ +/* + * 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.data.bytes; + +import net.openhft.chronicle.bytes.BytesStore; +import net.openhft.chronicle.bytes.RandomDataInput; +import net.openhft.chronicle.bytes.VanillaBytes; +import net.openhft.chronicle.hash.AbstractData; +import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; +import net.openhft.chronicle.map.impl.stage.map.ValueBytesInterop; +import net.openhft.sg.Stage; +import net.openhft.sg.StageRef; +import net.openhft.sg.Staged; + +import static net.openhft.chronicle.bytes.NoBytesStore.NO_BYTES_STORE; + +@Staged +public class WrappedValueBytesData extends AbstractData { + + @StageRef ValueBytesInterop vi; + @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation; + + private WrappedValueBytesData next; + + boolean nextInit() { + return true; + } + + void closeNext() { + // do nothing + } + + @Stage("Next") + public WrappedValueBytesData getUnusedWrappedValueBytesData() { + if (!wrappedValueBytesStoreInit()) + return this; + if (next == null) + next = new WrappedValueBytesData<>(); + return next.getUnusedWrappedValueBytesData(); + } + + @Stage("WrappedValueBytesStore") private BytesStore wrappedValueBytesStore; + @Stage("WrappedValueBytesStore") private long wrappedValueBytesOffset; + @Stage("WrappedValueBytesStore") private long wrappedValueBytesSize; + + boolean wrappedValueBytesStoreInit() { + return wrappedValueBytesStore != null; + } + + public void initWrappedValueBytesStore(BytesStore bytesStore, long offset, long size) { + wrappedValueBytesStore = bytesStore; + wrappedValueBytesOffset = offset; + wrappedValueBytesSize = size; + } + + void closeWrappedValueBytesStore() { + wrappedValueBytesStore = null; + if (next != null) + next.closeWrappedValueBytesStore(); + } + + @Stage("WrappedValueBytes") private final VanillaBytes wrappedValueBytes = + new VanillaBytes(NO_BYTES_STORE); + @Stage("WrappedValueBytes") private boolean wrappedValueBytesUsed = false; + + boolean wrappedValueBytesInit() { + return wrappedValueBytesUsed; + } + + void initWrappedValueBytes() { + wrappedValueBytes.bytesStore( + wrappedValueBytesStore, wrappedValueBytesOffset, wrappedValueBytesSize); + wrappedValueBytesUsed = true; + } + + void closeWrappedValueBytes() { + wrappedValueBytes.bytesStore(NO_BYTES_STORE, 0, 0); + wrappedValueBytesUsed = false; + } + + @Stage("CachedWrappedValue") private V cachedWrappedValue; + @Stage("CachedWrappedValue") private boolean cachedWrappedValueRead = false; + + private void initCachedWrappedValue() { + cachedWrappedValue = innerGetUsing(cachedWrappedValue); + cachedWrappedValueRead = true; + } + + @Override + public RandomDataInput bytes() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return wrappedValueBytes.bytesStore(); + } + + @Override + public long offset() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return wrappedValueBytesOffset; + } + + @Override + public long size() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return wrappedValueBytesSize; + } + + @Override + public V get() { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return cachedWrappedValue; + } + + @Override + public V getUsing(V using) { + checkOnEachPublicOperation.checkOnEachPublicOperation(); + return innerGetUsing(using); + } + + private V innerGetUsing(V usingValue) { + wrappedValueBytes.readPosition(wrappedValueBytesOffset); + return vi.valueReader.read(wrappedValueBytes, wrappedValueBytesSize, usingValue); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java b/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java index 309ae762f..9bb49866b 100644 --- a/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java @@ -20,6 +20,7 @@ import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages; import net.openhft.chronicle.hash.replication.RemoteOperationContext; +import net.openhft.chronicle.map.Replica; import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder; import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData; import net.openhft.chronicle.map.impl.stage.data.bytes.ReplicatedInputKeyBytesData; @@ -33,9 +34,9 @@ import net.openhft.sg.Staged; @Staged -public abstract class ReplicatedInput - implements RemoteOperationContext, MapRemoteQueryContext { - +public abstract class ReplicatedInput implements RemoteOperationContext, + MapRemoteQueryContext, Replica.QueryContext { + @StageRef ReplicatedChronicleMapHolder mh; @StageRef ReplicationUpdate ru; @StageRef ReplicatedInputKeyBytesData replicatedInputKeyBytesValue; @@ -90,11 +91,26 @@ public abstract class ReplicatedInput mh.m().setLastModificationTime(riId, bootstrapTimestamp); q.initInputKey(replicatedInputKeyBytesValue); - s.innerUpdateLock.lock(); if (isDeleted) { + s.innerUpdateLock.lock(); mh.m().remoteOperations.remove(this); } else { + s.innerWriteLock.lock(); mh.m().remoteOperations.put(this, replicatedInputValueBytesValue); } } + + @Override + public void remotePut(Data newValue, byte remoteIdentifier, long timestamp) { + mh.m().setLastModificationTime(remoteIdentifier, timestamp); + s.innerUpdateLock.lock(); + mh.m().remoteOperations.put(this, newValue); + } + + @Override + public void remoteRemove(byte remoteIdentifier, long timestamp) { + mh.m().setLastModificationTime(remoteIdentifier, timestamp); + s.innerWriteLock.lock(); + mh.m().remoteOperations.remove(this); + } } 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 a71fb4014..5bd3bcb06 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 @@ -18,11 +18,10 @@ package net.openhft.chronicle.map.impl.stage.iter; import net.openhft.chronicle.hash.Data; import net.openhft.chronicle.hash.impl.stage.iter.HashSegmentIteration; -import net.openhft.chronicle.map.MapContext; import net.openhft.chronicle.map.MapEntry; import net.openhft.chronicle.map.impl.IterationContext; import net.openhft.chronicle.map.impl.stage.entry.MapEntryStages; -import net.openhft.chronicle.map.impl.stage.map.WrappedValueInstanceValueHolder; +import net.openhft.chronicle.map.impl.stage.map.WrappedValueInstanceDataHolderAccess; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; import org.jetbrains.annotations.NotNull; @@ -32,7 +31,7 @@ public abstract class MapSegmentIteration extends HashSegmentIteration< implements MapEntry, IterationContext { @StageRef MapEntryStages entry; - @StageRef WrappedValueInstanceValueHolder wrappedValueInstanceValueHolder; + @StageRef WrappedValueInstanceDataHolderAccess wrappedValueInstanceDataHolderAccess; @Override public void doReplaceValue(Data newValue) { @@ -46,7 +45,7 @@ public abstract class MapSegmentIteration extends HashSegmentIteration< @NotNull @Override - public WrappedValueInstanceValueHolder context() { - return wrappedValueInstanceValueHolder; + public WrappedValueInstanceDataHolderAccess context() { + return wrappedValueInstanceDataHolderAccess; } } 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 076f03b45..89327c5de 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 @@ -23,7 +23,7 @@ import net.openhft.chronicle.hash.replication.ReplicableEntry; import net.openhft.chronicle.map.MapAbsentEntry; import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages; import net.openhft.chronicle.map.impl.stage.map.DefaultValue; -import net.openhft.chronicle.map.impl.stage.map.WrappedValueInstanceValueHolder; +import net.openhft.chronicle.map.impl.stage.map.WrappedValueInstanceDataHolderAccess; import net.openhft.chronicle.set.SetAbsentEntry; import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @@ -40,7 +40,7 @@ public class ReplicatedMapAbsentDelegatingForIteration @NotNull @Override - public WrappedValueInstanceValueHolder context() { + public WrappedValueInstanceDataHolderAccess context() { return delegate.context(); } diff --git a/src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueBytesDataAccess.java b/src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueBytesDataAccess.java new file mode 100644 index 000000000..e80aa0225 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueBytesDataAccess.java @@ -0,0 +1,43 @@ +/* + * 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.bytes.BytesStore; +import net.openhft.chronicle.hash.Data; +import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation; +import net.openhft.chronicle.hash.impl.util.Objects; +import net.openhft.chronicle.map.MapContext; +import net.openhft.chronicle.map.impl.stage.data.bytes.WrappedValueBytesData; +import net.openhft.sg.StageRef; +import net.openhft.sg.Staged; + +@Staged +public abstract class WrappedValueBytesDataAccess implements MapContext { + + @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation; + @StageRef WrappedValueBytesData wrappedValueBytesData; + + @Override + public Data wrapValueBytesAsData(BytesStore bytesStore, long offset, long size) { + Objects.requireNonNull(bytesStore); + checkOnEachPublicOperation.checkOnEachPublicOperation(); + WrappedValueBytesData wrapped = this.wrappedValueBytesData; + wrapped = wrapped.getUnusedWrappedValueBytesData(); + wrapped.initWrappedValueBytesStore(bytesStore, offset, size); + return wrapped; + } +} 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/WrappedValueInstanceDataHolderAccess.java similarity index 90% rename from src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueInstanceValueHolder.java rename to src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueInstanceDataHolderAccess.java index 33f6e26aa..51663b873 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/WrappedValueInstanceDataHolderAccess.java @@ -25,17 +25,16 @@ import net.openhft.sg.StageRef; import net.openhft.sg.Staged; @Staged -public abstract class WrappedValueInstanceValueHolder +public abstract class WrappedValueInstanceDataHolderAccess implements MapContext, SetContext { @StageRef CheckOnEachPublicOperation checkOnEachPublicOperation; - @StageRef - WrappedValueInstanceDataHolder wrappedValueInstanceValueHolder; + @StageRef WrappedValueInstanceDataHolder wrappedValueInstanceDataHolder; @Override public Data wrapValueAsData(V value) { checkOnEachPublicOperation.checkOnEachPublicOperation(); - WrappedValueInstanceDataHolder wrapped = this.wrappedValueInstanceValueHolder; + WrappedValueInstanceDataHolder wrapped = this.wrappedValueInstanceDataHolder; wrapped = wrapped.getUnusedWrappedValueHolder(); wrapped.initValue(value); return wrapped.wrappedData; 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 b2a1e1eb2..5b24e9094 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 @@ -16,7 +16,9 @@ package net.openhft.chronicle.map.impl.stage.query; +import net.openhft.chronicle.bytes.BytesStore; import net.openhft.chronicle.hash.Data; +import net.openhft.chronicle.hash.impl.stage.data.bytes.InputKeyBytesData; import net.openhft.chronicle.hash.impl.stage.query.HashQuery; import net.openhft.chronicle.hash.impl.stage.query.KeySearch; import net.openhft.chronicle.hash.impl.stage.query.SearchAllocatedChunks; @@ -47,6 +49,7 @@ public abstract class MapQuery extends HashQuery @StageRef SearchAllocatedChunks allocatedChunks; @StageRef KeySearch ks; @StageRef public AcquireHandle acquireHandle; + @StageRef InputKeyBytesData inputKeyBytesData; @StageRef public DefaultReturnValue defaultReturnValue; @StageRef public UsingReturnValue usingReturnValue; @@ -117,4 +120,10 @@ public abstract class MapQuery extends HashQuery checkOnEachPublicOperation.checkOnEachPublicOperation(); return this; } + + @Override + public Data getInputKeyBytesAsData(BytesStore bytesStore, long offset, long size) { + inputKeyBytesData.initInputKeyBytesStore(bytesStore, offset, size); + return inputKeyBytesData; + } } diff --git a/src/main/java/net/openhft/chronicle/set/SetFromMap.java b/src/main/java/net/openhft/chronicle/set/SetFromMap.java index f158d99b6..086ff4bb7 100644 --- a/src/main/java/net/openhft/chronicle/set/SetFromMap.java +++ b/src/main/java/net/openhft/chronicle/set/SetFromMap.java @@ -16,7 +16,9 @@ package net.openhft.chronicle.set; +import net.openhft.chronicle.bytes.BytesStore; import net.openhft.chronicle.hash.Data; +import net.openhft.chronicle.hash.impl.util.Objects; import net.openhft.chronicle.map.ChronicleMap; import net.openhft.chronicle.map.VanillaChronicleMap; import org.jetbrains.annotations.NotNull; @@ -129,22 +131,32 @@ class SetFromMap extends AbstractSet return m.keyClass(); } + // TODO test queryContext methods + @NotNull @Override public ExternalSetQueryContext queryContext(E key) { - //TODO - throw new UnsupportedOperationException(); + //noinspection unchecked + return (ExternalSetQueryContext) m.queryContext(key); } @NotNull @Override public ExternalSetQueryContext queryContext(Data key) { - //TODO - throw new UnsupportedOperationException(); + //noinspection unchecked + return (ExternalSetQueryContext) m.queryContext(key); + } + + @NotNull + @Override + public ExternalSetQueryContext queryContext(BytesStore keyBytes, long offset, long size) { + //noinspection unchecked + return (ExternalSetQueryContext) m.queryContext(keyBytes, offset, size); } @Override public SetSegmentContext segmentContext(int segmentIndex) { + // TODO throw new UnsupportedOperationException(); } @@ -153,14 +165,18 @@ class SetFromMap extends AbstractSet return m.segments(); } + // TODO test forEach methods + @Override public boolean forEachEntryWhile(Predicate> predicate) { - throw new UnsupportedOperationException(); + Objects.requireNonNull(predicate); + return m.forEachEntryWhile(e -> predicate.test(((SetEntry) e))); } @Override public void forEachEntry(Consumer> action) { - throw new UnsupportedOperationException(); + Objects.requireNonNull(action); + m.forEachEntry(e -> action.accept(((SetEntry) e))); } @Override diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index 2a03911ff..dadfe9211 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.bytes.BytesStore; import net.openhft.chronicle.core.io.Closeable; import net.openhft.chronicle.core.util.SerializableFunction; import org.jetbrains.annotations.NotNull; @@ -278,6 +279,13 @@ public class ReplicationCheckingMap implements ChronicleMap { throw new UnsupportedOperationException(); } + @NotNull + @Override + public ExternalMapQueryContext queryContext( + BytesStore keyBytes, long offset, long size) { + throw new UnsupportedOperationException(); + } + @Override public MapSegmentContext segmentContext(int segmentIndex) { throw new UnsupportedOperationException();