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 methods for Engine support

Browse Source
This commit is contained in:
leventov
2015-12-18 03:00:38 +03:00
parent 5280dc2b60
commit 3773b3d60a
19 changed files with 427 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/ChronicleHash.java
+13
View File
@@ -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<K, E extends HashEntry<K>, SC extends HashSegment
*/
@NotNull EQC queryContext(Data<K> 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}.
src/main/java/net/openhft/chronicle/hash/impl/stage/data/bytes/InputKeyBytesData.java
+105
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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<K> extends AbstractData<K> {
@StageRef KeyBytesInterop<K> 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);
}
}
src/main/java/net/openhft/chronicle/map/MapContext.java
+15
View File
@@ -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<K, V, R>
* @return the value as {@code Data}
*/
Data<V> 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}.
*
* <p>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<V> wrapValueBytesAsData(BytesStore valueBytes, long offset, long size);
}
src/main/java/net/openhft/chronicle/map/Replica.java
+8
View File
@@ -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<K, V> {
void remotePut(Data<V> newValue, byte remoteIdentifier, long timestamp);
void remoteRemove(byte remoteIdentifier, long timestamp);
}
/**
* Provides the unique Identifier associated with this map instance. <p> An identifier is used
* to determine which replicating node made the change. <p> If two nodes update their map at the
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+12
View File
@@ -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<K, V, R>
return q;
}
@NotNull
@Override
public ExternalMapQueryContext<K, V, ?> queryContext(
BytesStore keyBytes, long offset, long size) {
Objects.requireNonNull(keyBytes);
QueryContextInterface<K, V, R> q = mapContext();
q.initInputKey(q.getInputKeyBytesAsData(keyBytes, offset, size));
return q;
}
@Override
public MapSegmentContext<K, V, ?> segmentContext(int segmentIndex) {
IterationContext<K, V, ?> c = iterationContext();
src/main/java/net/openhft/chronicle/map/impl/MapIterationContext.java
+5 -5
View File
@@ -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,
})
src/main/java/net/openhft/chronicle/map/impl/MapQueryContext.java
+7 -1
View File
@@ -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,
src/main/java/net/openhft/chronicle/map/impl/QueryContextInterface.java
+3
View File
@@ -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<K, V, R> extends ExternalMapQueryContext<K, V, R> {
void initInputKey(Data<K> inputKey);
Data<K> getInputKeyBytesAsData(BytesStore bytesStore, long offset, long size);
DataAccess<K> inputKeyDataAccess();
src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapIterationContext.java
+4 -1
View File
@@ -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,
src/main/java/net/openhft/chronicle/map/impl/ReplicatedMapQueryContext.java
+7 -1
View File
@@ -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,
src/main/java/net/openhft/chronicle/map/impl/stage/data/bytes/WrappedValueBytesData.java
+137
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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<V> extends AbstractData<V> {
@StageRef ValueBytesInterop<V> vi;
@StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
private WrappedValueBytesData<V> next;
boolean nextInit() {
return true;
}
void closeNext() {
// do nothing
}
@Stage("Next")
public WrappedValueBytesData<V> 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);
}
}
src/main/java/net/openhft/chronicle/map/impl/stage/input/ReplicatedInput.java
+20 -4
View File
@@ -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<K, V, R>
implements RemoteOperationContext<K>, MapRemoteQueryContext<K, V, R> {
public abstract class ReplicatedInput<K, V, R> implements RemoteOperationContext<K>,
MapRemoteQueryContext<K, V, R>, Replica.QueryContext<K, V> {
@StageRef ReplicatedChronicleMapHolder<K, V, R> mh;
@StageRef ReplicationUpdate<K> ru;
@StageRef ReplicatedInputKeyBytesData<K> replicatedInputKeyBytesValue;
@@ -90,11 +91,26 @@ public abstract class ReplicatedInput<K, V, R>
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<V> 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);
}
}
src/main/java/net/openhft/chronicle/map/impl/stage/iter/MapSegmentIteration.java
+4 -5
View File
@@ -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<K, V, R> extends HashSegmentIteration<
implements MapEntry<K, V>, IterationContext<K, V, R> {
@StageRef MapEntryStages<K, V> entry;
@StageRef WrappedValueInstanceValueHolder<K, V, ?> wrappedValueInstanceValueHolder;
@StageRef WrappedValueInstanceDataHolderAccess<K, V, ?> wrappedValueInstanceDataHolderAccess;
@Override
public void doReplaceValue(Data<V> newValue) {
@@ -46,7 +45,7 @@ public abstract class MapSegmentIteration<K, V, R> extends HashSegmentIteration<
@NotNull
@Override
public WrappedValueInstanceValueHolder<K, V, ?> context() {
return wrappedValueInstanceValueHolder;
public WrappedValueInstanceDataHolderAccess<K, V, ?> context() {
return wrappedValueInstanceDataHolderAccess;
}
}
src/main/java/net/openhft/chronicle/map/impl/stage/iter/ReplicatedMapAbsentDelegatingForIteration.java
+2 -2
View File
@@ -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<K, V>
@NotNull
@Override
public WrappedValueInstanceValueHolder<K, V, ?> context() {
public WrappedValueInstanceDataHolderAccess<K, V, ?> context() {
return delegate.context();
}
src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueBytesDataAccess.java
+43
View File
@@ -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 <http://www.gnu.org/licenses/>.
*/
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<K, V, R> implements MapContext<K, V, R> {
@StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
@StageRef WrappedValueBytesData<V> wrappedValueBytesData;
@Override
public Data<V> wrapValueBytesAsData(BytesStore bytesStore, long offset, long size) {
Objects.requireNonNull(bytesStore);
checkOnEachPublicOperation.checkOnEachPublicOperation();
WrappedValueBytesData<V> wrapped = this.wrappedValueBytesData;
wrapped = wrapped.getUnusedWrappedValueBytesData();
wrapped.initWrappedValueBytesStore(bytesStore, offset, size);
return wrapped;
}
}
src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueInstanceValueHolder.java → src/main/java/net/openhft/chronicle/map/impl/stage/map/WrappedValueInstanceDataHolderAccess.java
+3 -4
View File
@@ -25,17 +25,16 @@ import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
@Staged
public abstract class WrappedValueInstanceValueHolder<K, V, R>
public abstract class WrappedValueInstanceDataHolderAccess<K, V, R>
implements MapContext<K, V, R>, SetContext<K, R> {
@StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
@StageRef
WrappedValueInstanceDataHolder<V> wrappedValueInstanceValueHolder;
@StageRef WrappedValueInstanceDataHolder<V> wrappedValueInstanceDataHolder;
@Override
public Data<V> wrapValueAsData(V value) {
checkOnEachPublicOperation.checkOnEachPublicOperation();
WrappedValueInstanceDataHolder<V> wrapped = this.wrappedValueInstanceValueHolder;
WrappedValueInstanceDataHolder<V> wrapped = this.wrappedValueInstanceDataHolder;
wrapped = wrapped.getUnusedWrappedValueHolder();
wrapped.initValue(value);
return wrapped.wrappedData;
src/main/java/net/openhft/chronicle/map/impl/stage/query/MapQuery.java
+9
View File
@@ -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<K, V, R> extends HashQuery<K>
@StageRef SearchAllocatedChunks allocatedChunks;
@StageRef KeySearch<K> ks;
@StageRef public AcquireHandle<K, V> acquireHandle;
@StageRef InputKeyBytesData<K> inputKeyBytesData;
@StageRef public DefaultReturnValue<V> defaultReturnValue;
@StageRef public UsingReturnValue<V> usingReturnValue;
@@ -117,4 +120,10 @@ public abstract class MapQuery<K, V, R> extends HashQuery<K>
checkOnEachPublicOperation.checkOnEachPublicOperation();
return this;
}
@Override
public Data<K> getInputKeyBytesAsData(BytesStore bytesStore, long offset, long size) {
inputKeyBytesData.initInputKeyBytesStore(bytesStore, offset, size);
return inputKeyBytesData;
}
}
src/main/java/net/openhft/chronicle/set/SetFromMap.java
+22 -6
View File
@@ -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<E> extends AbstractSet<E>
return m.keyClass();
}
// TODO test queryContext methods
@NotNull
@Override
public ExternalSetQueryContext<E, ?> queryContext(E key) {
//TODO
throw new UnsupportedOperationException();
//noinspection unchecked
return (ExternalSetQueryContext<E, ?>) m.queryContext(key);
}
@NotNull
@Override
public ExternalSetQueryContext<E, ?> queryContext(Data<E> key) {
//TODO
throw new UnsupportedOperationException();
//noinspection unchecked
return (ExternalSetQueryContext<E, ?>) m.queryContext(key);
}
@NotNull
@Override
public ExternalSetQueryContext<E, ?> queryContext(BytesStore keyBytes, long offset, long size) {
//noinspection unchecked
return (ExternalSetQueryContext<E, ?>) m.queryContext(keyBytes, offset, size);
}
@Override
public SetSegmentContext<E, ?> segmentContext(int segmentIndex) {
// TODO
throw new UnsupportedOperationException();
}
@@ -153,14 +165,18 @@ class SetFromMap<E> extends AbstractSet<E>
return m.segments();
}
// TODO test forEach methods
@Override
public boolean forEachEntryWhile(Predicate<? super SetEntry<E>> predicate) {
throw new UnsupportedOperationException();
Objects.requireNonNull(predicate);
return m.forEachEntryWhile(e -> predicate.test(((SetEntry<E>) e)));
}
@Override
public void forEachEntry(Consumer<? super SetEntry<E>> action) {
throw new UnsupportedOperationException();
Objects.requireNonNull(action);
m.forEachEntry(e -> action.accept(((SetEntry<E>) e)));
}
@Override