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

HCOLL-318 added many documentation and examples; Empowerted ReplicableEntry interface, and changed Map/SetRemoteOperations contracts to allow implementing more agile reconciliation strategies

Browse Source
This commit is contained in:
leventov
2015-06-20 05:43:34 +03:00
parent 9746084998
commit 1d7f54635a
28 changed files with 4164 additions and 2862 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/ChronicleHash.java
+27 -1
View File
@@ -19,6 +19,8 @@
package net.openhft.chronicle.hash;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.MapMethods;
import net.openhft.chronicle.map.MapQueryContext;
import net.openhft.chronicle.set.ChronicleSet;
import org.jetbrains.annotations.NotNull;
@@ -40,7 +42,7 @@ public interface ChronicleHash<K, C extends KeyContext<K>, EQC extends ExternalH
*
* @return the file this {@link ChronicleMap} or {@link ChronicleSet} is mapped to,
* or {@code null} if it is not mapped to any file
* @see ChronicleHashBuilder#createPersistedTo(java.io.File)
* @see ChronicleHashBuilder#createPersistedTo(File)
*/
File file();
@@ -57,9 +59,33 @@ public interface ChronicleHash<K, C extends KeyContext<K>, EQC extends ExternalH
*/
Class<K> keyClass();
/**
* Returns the context to perform arbitrary operations with the given key in this map.
* Conventionally, try-with-resources block should wrap the returned context: <pre>{@code
* try (ExternalHashQueryContext<K> q = hash.queryContext(key)) {
* // ... do something
* }}</pre>
* See {@link HashQueryContext} and {@link MapMethods} for a lot of inspiration about using this
* functionality.
*
* @param key the queried key
* @return the context to perform operations with the key
* @see HashQueryContext
* @see MapQueryContext
* @see ExternalHashQueryContext
* @see MapMethods
*/
@NotNull
EQC queryContext(K key);
/**
* Equivalent to {@link #queryContext(Object)}, but accepts {@code Data} instead of key as
* an object. Useful, when you already have {@code Data}, calling this method instead of {@link
* #queryContext(Object)} might help to avoid unnecessary deserialization.
*
* @param key the queried key as {@code Data}
* @return the context to perform operations with the key
*/
@NotNull EQC queryContext(Data<K, ?> key);
/**
src/main/java/net/openhft/chronicle/hash/HashQueryContext.java
+59 -10
View File
@@ -2,7 +2,11 @@ package net.openhft.chronicle.hash;
import net.openhft.chronicle.hash.locks.InterProcessLock;
import net.openhft.chronicle.hash.locks.InterProcessReadWriteUpdateLock;
import net.openhft.chronicle.map.MapEntryOperations;
import net.openhft.chronicle.map.MapMethods;
import net.openhft.chronicle.map.MapQueryContext;
import net.openhft.chronicle.set.SetEntryOperations;
import net.openhft.chronicle.set.SetQueryContext;
/**
* Context of {@link ChronicleHash} operations with <i>individual keys</i>.
@@ -21,10 +25,11 @@ import net.openhft.chronicle.map.MapMethods;
* try (ExternalMapQueryContext<K, V, ?> q = map.queryContext(key)) {
* // q.entry(), checks if the entry is present in the map, and acquires
* // the read lock for that.
* if (q.entry() != null) {
* MapEntry<K, V> entry = q.entry();
* if (entry != null) {
* // Tries to acquire the write lock to perform modification,
* // but this is an illegal upgrade: read -> write, throws IllegalMonitorStateException
* q.remove(q.entry());
* q.remove(entry);
* }
* }}</pre>
* So, to workaround this, you should acquire the {@linkplain #updateLock() update lock},
@@ -33,9 +38,9 @@ import net.openhft.chronicle.map.MapMethods;
* // CORRECT
* try (ExternalMapQueryContext<K, V, ?> q = map.queryContext(key)) {
* q.updateLock().lock(); // acquire the update lock before checking the entry presence.
* if (q.entry() != null) {
* q.remove(q.entry());
* }
* MapEntry<K, V> entry = q.entry();
* if (entry != null)
* q.remove(entry);
* }}</pre>
* </li>
* <li>You want to try to acquire some lock heuristically, in order to improve total {@code
@@ -46,14 +51,14 @@ import net.openhft.chronicle.map.MapMethods;
* // For acquireUsing(), it is assumed to be very probable, that the entry is already
* // present in the map, so we will perform the whole acquireUsing() without exclusive locking
* if (q.readLock().tryLock()) {
* MapEntry<?, V> entry = q.entry();
* MapEntry<K, V> entry = q.entry();
* if (entry != null) {
* // Entry is present, return
* returnValue.returnValue(entry.value());
* return;
* }
* // Key is absent
* // Need to unlock, to lock to update lock later. Direct updrage is forbidden.
* // Need to unlock, to lock to update lock later. Direct upgrade is forbidden.
* q.readLock().unlock();
* }
* // We are here, either if we:
@@ -63,7 +68,7 @@ import net.openhft.chronicle.map.MapMethods;
* // 2) Seen the entry is absent under the read lock. This means we need to insert
* // the default value into the map. that requires update-level access as well
* q.updateLock().lock();
* MapEntry<?, V> entry = q.entry();
* MapEntry<K, V> entry = q.entry();
* if (entry != null) {
* // Entry is present, return
* returnValue.returnValue(entry.value());
@@ -74,7 +79,7 @@ import net.openhft.chronicle.map.MapMethods;
* returnValue.returnValue(q.entry().value());
* }}</pre></li>
* <li>If the default {@link InterProcessLock#lock()} policy of trying to acquire the
* lock for some time, and then throw {@code RuntimeException}, or need custom timeout:
* lock for some time, and then throw {@code RuntimeException}, or need a custom timeout:
* <pre>{@code
* try (ExternalHashQueryContext<K> q = hash.queryContext(key)) {
* if (q.writeLock().tryLock(5, TimeUnit.SECONDS)) {
@@ -84,9 +89,53 @@ import net.openhft.chronicle.map.MapMethods;
* }
* }}</pre></li>
* </ul>
*
* <p>{@code HashQueryContext} defines the common pattern for working with {@code ChronicleHash}
* contexts: it has a pair of methods, {@link #entry()} and {@link #absentEntry()}, at any moment
* one of them returns an (absent) entry context object, another - {@code null}, depending on the
* presence of the {@linkplain #queriedKey() queried key} in the {@code ChronicleHash}. Thus,
* block of code that uses {@code HashQueryContext} usually has an if-else statement,
* with "then" branch for dealing with the present entry, {@code else} branch for dealing with
* the absent entry, or vise-versa. For example: <pre>{@code
* interface Point {
* double getX();
* void setX(double x);
* double addX(double xAdd);
*
* double getY();
* void setY(double y);
* double addY(double yAdd);
* }
*
* <K> Point movePoint(ChronicleMap<K, Point> map, K key, double xMove, double yMove,
* Point using) {
* // Moves existing point by [xMove, yMove], if absent - assumes the default point is [0, 0].
* // Returns the resulting point
* try (ExternalMapQueryContext<K, Point, ?> q = map.queryContext(key)) {
* Point offHeapPoint;
* q.updateLock().lock();
* MapEntry<K, Point> entry = q.entry();
* if (entry != null) {
* // Key is present
* offHeapPoint = entry.value().getUsing(using);
* } else {
* // Key is absent
* q.insert(q.absentEntry(), q.defaultValue(q.absentEntry()));
* offHeapPoint = q.entry().value().getUsing(using);
* }
* offHeapPoint.addX(xMove);
* offHeapPoint.addY(yMove);
* return offHeapPoint;
* }
* }}</pre>
*
* <p>{@code HashQueryContext} is the base interface defining the structure, but it has no methods
* to anything "interesting" with {@code entry()} or {@code absentEntry()}. Use {@link
* MapQueryContext} or {@link SetQueryContext} interfaces, which provide access to {@link
* MapEntryOperations} and {@link SetEntryOperations} respectively.
*
* @param <K> the hash key type
* @see ChronicleHash#queryContext(Object)
* @see ChronicleHash#queryContext(Object)
*/
public interface HashQueryContext<K> extends HashContext<K>, InterProcessReadWriteUpdateLock {
src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
-1
View File
@@ -16,7 +16,6 @@
package net.openhft.chronicle.hash.impl;
import net.openhft.chronicle.core.OS;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ExternalHashQueryContext;
import net.openhft.chronicle.hash.KeyContext;
src/main/java/net/openhft/chronicle/hash/AcceptanceDecision.java → src/main/java/net/openhft/chronicle/hash/replication/AcceptanceDecision.java
+5 -9
View File
@@ -16,25 +16,21 @@
* limitations under the License.
*/
package net.openhft.chronicle.hash;
package net.openhft.chronicle.hash.replication;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.replication.DefaultEventualConsistencyStrategy;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
import net.openhft.chronicle.set.replication.SetRemoteOperations;
/**
* Decision, if {@link MapRemoteOperations remote modification operation} should be accepted
* or discarded.
* *
* @see MapRemoteOperations
* @see SetRemoteOperations
* or discarded. Used in {@link DefaultEventualConsistencyStrategy}.
*/
public enum AcceptanceDecision {
/**
* Acceptance decision -- the remote modification operation is applied to the local
* {@link ChronicleHash} state. This decision should also been returned, when the actual
* state modification is not needed (for example, remote {@code remove entry} operation, while
* the entry is not present in the ChronicleHash locally), but the operation is generally
* considered acceptable.
* {@link ChronicleHash} state.
*/
ACCEPT,
src/main/java/net/openhft/chronicle/hash/replication/DefaultEventualConsistencyStrategy.java
+2 -3
View File
@@ -18,14 +18,13 @@
package net.openhft.chronicle.hash.replication;
import net.openhft.chronicle.hash.AcceptanceDecision;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
import net.openhft.chronicle.set.replication.SetRemoteOperations;
import static net.openhft.chronicle.hash.AcceptanceDecision.ACCEPT;
import static net.openhft.chronicle.hash.AcceptanceDecision.DISCARD;
import static net.openhft.chronicle.hash.replication.AcceptanceDecision.ACCEPT;
import static net.openhft.chronicle.hash.replication.AcceptanceDecision.DISCARD;
/**
* Specifies the default eventual consistency strategy for {@link ChronicleHashBuilder#replication(
src/main/java/net/openhft/chronicle/hash/replication/HashReplicableEntry.java
+1 -1
View File
@@ -20,6 +20,6 @@ package net.openhft.chronicle.hash.replication;
import net.openhft.chronicle.hash.HashEntry;
public interface HashReplicableEntry<K> extends HashEntry<K>,ReplicableEntry {
public interface HashReplicableEntry<K> extends HashEntry<K>, ReplicableEntry {
}
src/main/java/net/openhft/chronicle/hash/replication/ReplicableEntry.java
+65
View File
@@ -1,7 +1,72 @@
package net.openhft.chronicle.hash.replication;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.HashAbsentEntry;
import net.openhft.chronicle.hash.HashEntry;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.Replica;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
import net.openhft.chronicle.set.replication.SetRemoteOperations;
/**
* Abstracts an in replicated {@link ChronicleHash}. A {@link HashAbsentEntry} might be a {@code
* ReplicableEntry} as well as {@link HashEntry}.
*/
public interface ReplicableEntry {
/**
* The identifier, associated with this entry. Originally, and by default, this identifier means
* the identifier of the {@code ChronicleHash} node on which this entry was updated last.
*
* <p>When the entry is replicated, this identifier comes to remote nodes as {@link
* HashRemoteQueryContext#remoteIdentifier()}.
*
* <p>On {@code ChronicleHash} local operations with entries, like {@link ChronicleMap#put(
* Object, Object)}, this identifier is overwritten to the own {@code ChronicleHash} {@link
* Replica#identifier()}. On remote operations, proxied through {@link MapRemoteOperations} or
* {@link SetRemoteOperations}, you are free to overwrite this identifier to any value, using
* {@link #updateOrigin(byte, long)} method.
*
* @return the identifier, associated with this entry
*/
byte originIdentifier();
/**
* The timestamp, associated with this entry. Originally, and by default, this timestamp means
* the time of the last.
*
* <p>When the entry is replicated, this timestamp comes to remote nodes as {@link
* HashRemoteQueryContext#remoteTimestamp()}.
*
* <p>On {@code ChronicleHash} local operations with entries, like {@link ChronicleMap#put(
* Object, Object)}, this timestamp is overwritten using {@linkplain
* ChronicleHashBuilder#timeProvider(TimeProvider) the configured time provider}. On remote
* operations, proxied through {@link MapRemoteOperations} or {@link SetRemoteOperations},
* you are free to overwrite this timestamp by any value, using {@link #updateOrigin(byte, long)
* } method.
*
* @return the timestamp, associated with this entry
* @see TimeProvider
*/
long originTimestamp();
/**
* Overwrite the entry's associated identifier and timestamp. After this call {@link
* #originIdentifier()} returns the given {@code newIdentifier} and {@link #originTimestamp()}
* returns the given {@code newTimestamp} respectively.
*
* @param newIdentifier the new identifier for the entry
* @param newTimestamp the new timestamp for the entry
*/
void updateOrigin(byte newIdentifier, long newTimestamp);
/**
* Suppress the entry, if it was scheduled to be replicated over to remote Chronicle nodes.
*/
void dropChanged();
/**
* Propagate the entry, schedule it to be replicated over to remote Chronicle nodes.
*/
void raiseChanged();
}
src/main/java/net/openhft/chronicle/map/ChannelFactory.java
-26
View File
@@ -1,26 +0,0 @@
/*
* Copyright 2014 Higher Frequency Trading
*
* http://www.higherfrequencytrading.com
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.openhft.chronicle.map;
/**
* Created by Rob Austin
*/
public interface ChannelFactory {
void createChannel(short channelID);
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+11 -2
View File
@@ -28,6 +28,7 @@ import net.openhft.chronicle.hash.replication.*;
import net.openhft.chronicle.hash.serialization.*;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter;
import net.openhft.chronicle.hash.serialization.internal.MetaProvider;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
import net.openhft.chronicle.set.ChronicleSetBuilder;
import net.openhft.lang.Maths;
import net.openhft.lang.io.Bytes;
@@ -168,6 +169,7 @@ public final class ChronicleMapBuilder<K, V> implements
MapMethods<K, V, ?> methods = DefaultSpi.mapMethods();
MapEntryOperations<K, V, ?> entryOperations = mapEntryOperations();
MapRemoteOperations<K, V, ?> remoteOperations = mapRemoteOperations();
ChronicleMapBuilder(Class<K> keyClass, Class<V> valueClass) {
keyBuilder = new SerializationBuilder<>(keyClass, SerializationBuilder.Role.KEY);
@@ -1495,8 +1497,8 @@ public final class ChronicleMapBuilder<K, V> implements
}
/**
* Inject your SPI code around basic {@code ChronicleMap}'s operations with present entries:
* removing entries and replacing the entries' value.
* Inject your SPI code around basic {@code ChronicleMap}'s operations with entries:
* removing entries, replacing the entries' value and inserting the new entry.
*
* This affects behaviour of ordinary map.put(), map.remove(), etc. calls, as well as removes
* and replacing values <i>during iterations</i>, <i>remote map calls</i> and
@@ -1520,5 +1522,12 @@ public final class ChronicleMapBuilder<K, V> implements
this.methods = mapMethods;
return this;
}
public ChronicleMapBuilder<K, V> remoteOperations(
MapRemoteOperations<K, V, ?> remoteOperations) {
Objects.requireNonNull(remoteOperations);
this.remoteOperations = remoteOperations;
return this;
}
}
src/main/java/net/openhft/chronicle/map/MapContext.java
+5 -2
View File
@@ -22,11 +22,14 @@ import net.openhft.chronicle.hash.HashContext;
import net.openhft.chronicle.hash.Data;
/**
* Context, in which {@link MapEntry MapEntries} are accessed.
* Context, in which {@link MapEntry MapEntries} are accessed. {@code MapContext} allows to access
* {@link MapEntryOperations}, configured for the accessed {@link ChronicleMap} via {@link
* ChronicleMapBuilder#entryOperations(MapEntryOperations)}. {@code MapContext} implements {@code
* MapEntryOperations} by delegation to the configured {@code entryOperations}.
*
* @param <K> the map key type
* @param <V> the map value type
* @param <R> the return type of {@link MapEntryOperations} specialized for the queried map
* @param <R> the return type of {@link MapEntryOperations} specified for the queried map
*/
public interface MapContext<K, V, R> extends HashContext<K>, MapEntryOperations<K, V, R> {
/**
src/main/java/net/openhft/chronicle/map/MapEntryOperations.java
+80 -4
View File
@@ -19,17 +19,85 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.Data;
import net.openhft.chronicle.map.replication.MapRemoteOperations;
import org.jetbrains.annotations.NotNull;
/**
* SPI interface for customizing "low-level" modification operations on {@link ChronicleMap}
* entries. All {@code ChronicleMap} modifications operate via and instance of this interface:
* entries. All {@code ChronicleMap} modifications operate via an instance of this interface:
* <ul>
* <li>Ordinary query operations: {@code map.put()}, {@code map.compute()}. In fact, the</li>
* <li>Ordinary query operations: {@code map.put()}, {@code map.compute()}, {@code map.remove()}
* </li>
* <li>Operations with entry during iterations on the map, or the key/value/entry views,
* like {@code iterator.remove()}.</li>
* <li>Remote map operation calls and Replicated {@code ChronicleMap} reconciliation operations:
* see {@link MapRemoteOperations}.</li>
* </ul>
*
* <p>By default {@link #remove}, {@link #insert} and {@link #replaceValue} return {@code null},
* but subclasses could return something more sensible, to be used in higher-level SPI interfaces,
* namely {@link MapMethods} and {@link MapRemoteOperations}. For example, in bidirectional map
* implementation (i. e. a map that preserves the uniqueness of its values as well as that of
* its keys), that includes two {@code ChronicleMaps}, the {@code MapEntryOperations}' methods return
* type could be used to indicate if we were successful to lock both maps before performing
* the update: <pre>{@code
* enum DualLockSuccess {SUCCESS, FAIL}
*
* class BiMapEntryOperations<K, V> implements MapEntryOperations<K, V, DualLockSuccess> {
* ChronicleMap<V, K> reverse;
*
* public void setReverse(ChronicleMap<V, K> reverse) {
* this.reverse = reverse;
* }
*
* @Override
* public DualLockSuccess remove(@NotNull MapEntry<K, V> entry) {
* try (ExternalMapQueryContext<V, K, ?> rq = reverse.queryContext(entry.value())) {
* if (!rq.updateLock().tryLock()) {
* if (entry.context() instanceof MapQueryContext)
* return FAIL;
* throw new IllegalStateException("Concurrent modifications to reverse map " +
* "during remove during iteration");
* }
* MapEntry<V, K> reverseEntry = rq.entry();
* if (reverseEntry != null) {
* entry.doRemove();
* reverseEntry.doRemove();
* return SUCCESS;
* } else {
* throw new IllegalStateException(entry.key() + " maps to " + entry.value() +
* ", but in the reverse map this value is absent");
* }
* }
* }
*
* // ... other methods
* }
*
* class BiMapMethods<K, V> implements MapMethods<K, V, DualLockSuccess> {
* @Override
* public void remove(MapQueryContext<K, V, DualLockSuccess> q, ReturnValue<V> returnValue) {
* while (true) {
* q.updateLock().lock();
* try {
* MapEntry<K, V> entry = q.entry();
* if (entry != null) {
* returnValue.returnValue(entry.value());
* if (q.remove(entry) == SUCCESS)
* return;
* }
* } finally {
* q.readLock().unlock();
* }
* }
* }
*
* // ... other methods
* }}</pre>
*
* @param <K> the map key type
* @param <V> the map value type
* @param <R> methods return type, used for communication between lower- and higher-level SPI
* @see ChronicleMapBuilder#entryOperations(MapEntryOperations)
*/
public interface MapEntryOperations<K, V, R> {
@@ -43,6 +111,7 @@ public interface MapEntryOperations<K, V, R> {
* @param entry the entry to remove
* @throws IllegalStateException if some locking/state conditions required to perform remove
* operation are not met
* @throws RuntimeException if removal was unconditionally unsuccessful due to any reason
*/
default R remove(@NotNull MapEntry<K, V> entry) {
entry.doRemove();
@@ -52,9 +121,14 @@ public interface MapEntryOperations<K, V, R> {
/**
* Replaces the given entry's value with the new one.
*
* @implNote default implementation calls {@link MapEntry#doReplaceValue(Data)
* entry.doReplaceValue(newValue)} and returns {@code null}.
*
* @param entry the entry to replace the value in
* @throws IllegalStateException if some locking/state conditions required to perform replace
* operation are not met
* @throws RuntimeException if value replacement was unconditionally unsuccessful due
* to any reason
*/
default R replaceValue(@NotNull MapEntry<K, V> entry, Data<V, ?> newValue) {
entry.doReplaceValue(newValue);
@@ -64,11 +138,13 @@ public interface MapEntryOperations<K, V, R> {
/**
* Inserts the new entry into the map, of {@link MapAbsentEntry#absentKey() the key} from
* the given insertion context (<code>absentEntry</code>) and the given {@code value}.
* Returns {@code true} if the insertion was successful, {@code false} if it was discarded
* for any reason.
*
* @implNote default implementation calls {@link MapAbsentEntry#doInsert(Data)
* absentEntry.doInsert(value)} and returns {@code null}.
*
* @throws IllegalStateException if some locking/state conditions required to perform insertion
* operation are not met
* @throws RuntimeException if insertion was unconditionally unsuccessful due to any reason
*/
default R insert(@NotNull MapAbsentEntry<K, V> absentEntry, Data<V, ?> value) {
absentEntry.doInsert(value);
src/main/java/net/openhft/chronicle/map/MapMethods.java
+266 -7
View File
@@ -21,6 +21,7 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.Data;
import java.util.function.BiFunction;
import java.util.function.Function;
import static net.openhft.chronicle.hash.Data.bytesEquivalent;
import static net.openhft.chronicle.map.MapMethodsSupport.returnCurrentValueIfPresent;
@@ -28,6 +29,8 @@ import static net.openhft.chronicle.map.MapMethodsSupport.tryReturnCurrentValueI
/**
* SPI interface for customizing behaviour of the specific Map's methods with individual keys.
*
* @see ChronicleMapBuilder#mapMethods(MapMethods)
*/
public interface MapMethods<K, V, R> {
@@ -38,11 +41,34 @@ public interface MapMethods<K, V, R> {
/**
* Backing {@link ChronicleMap#get}, {@link ChronicleMap#getUsing} and
* {@link ChronicleMap#getOrDefault} methods.
*
* @implNote the default implementation is equivalent to <pre>{@code
* MapEntry<?, V> entry = q.entry();
* if (entry != null)
* returnValue.returnValue(entry.value());
* }</pre>
*/
default void get(MapQueryContext<K, V, R> q, ReturnValue<V> returnValue) {
returnCurrentValueIfPresent(q, returnValue);
}
/**
* Backing {@link ChronicleMap#put(Object, Object)} method.
*
* @implNote the default implementation is equivalent to <pre>{@code
* // We cannot read the previous value under read lock, because then we will need
* // to release the read lock -> acquire write lock, the value might be updated in
* // between, that will break ConcurrentMap.put() atomicity guarantee. So, we acquire
* // update lock from the start:
* q.updateLock().lock();
* MapEntry<K, V> entry = q.entry();
* if (entry != null) {
* returnValue.returnValue(entry.value());
* q.replaceValue(entry, value);
* } else {
* q.insert(q.absentEntry(), value);
* }}</pre>
*/
default void put(MapQueryContext<K, V, R> q, Data<V, ?> value, ReturnValue<V> returnValue) {
// We cannot read the previous value under read lock, because then we will need
// to release the read lock -> acquire write lock, the value might be updated in
@@ -58,6 +84,29 @@ public interface MapMethods<K, V, R> {
}
}
/**
* Backing {@link ChronicleMap#putIfAbsent(Object, Object)} method.
*
* @implNote the default implementation is equivalent to <pre>{@code
* if (q.readLock().tryLock()) {
* MapEntry<?, V> entry = q.entry();
* if (entry != null) {
* returnValue.returnValue(entry.value());
* return;
* }
* // Key is absent
* q.readLock().unlock();
* }
* q.updateLock().lock();
* MapEntry<?, V> entry = q.entry();
* if (entry != null) {
* returnValue.returnValue(entry.value());
* return;
* }
* // Key is absent
* q.insert(q.absentEntry(), value);
* }</pre>
*/
default void putIfAbsent(MapQueryContext<K, V, R> q,
Data<V, ?> value, ReturnValue<V> returnValue) {
if (tryReturnCurrentValueIfPresent(q, returnValue))
@@ -66,6 +115,33 @@ public interface MapMethods<K, V, R> {
q.insert(q.absentEntry(), value);
}
/**
* Backing {@link ChronicleMap#acquireUsing(Object, Object)} method.
*
* @implNote the default implementation is equivalent to <pre>{@code
* if (q.readLock().tryLock()) {
* MapEntry<?, V> entry = q.entry();
* if (entry != null) {
* returnValue.returnValue(entry.value());
* return;
* }
* // Key is absent
* q.readLock().unlock();
* }
* q.updateLock().lock();
* MapEntry<?, V> entry = q.entry();
* if (entry != null) {
* returnValue.returnValue(entry.value());
* return;
* }
* // Key is absent
* q.insert(q.absentEntry(), q.defaultValue(q.absentEntry()));
* // Meaningful to return the default as newly-inserted, not the default entry itself.
* // map.acquireUsing() is most useful for data-value generated values, for which it makes
* // big difference -- what bytes to refer. Consider map.acquireUsing(...).incrementValue();
* returnValue.returnValue(q.entry().value());
* }</pre>
*/
default void acquireUsing(MapQueryContext<K, V, R> q, ReturnValue<V> returnValue) {
if (tryReturnCurrentValueIfPresent(q, returnValue))
return;
@@ -73,10 +149,60 @@ public interface MapMethods<K, V, R> {
q.insert(q.absentEntry(), q.defaultValue(q.absentEntry()));
// meaningful to return the default as newly-inserted, not the default entry itself.
// map.acquireUsing() is most useful for data-value generated values, for which it makes
// big difference -- what bytes to refer. map.acquireUsing().incrementValue();
// big difference -- what bytes to refer. consider map.acquireUsing(...).incrementValue();
// The same reasoning is applied in all same occurrences in this class file
returnValue.returnValue(q.entry().value());
}
/**
* Backing {@link ChronicleMap#computeIfAbsent(Object, Function)} method.
*
* @implNote the default implementation is equivalent to <pre>{@code
* if (q.readLock().tryLock()) {
* MapEntry<?, V> entry = q.entry();
* if (entry != null) {
* returnValue.returnValue(entry.value());
* return;
* }
* // Key is absent
* q.readLock().unlock();
* }
* q.updateLock().lock();
* MapEntry<?, V> entry = q.entry();
* if (entry != null) {
* returnValue.returnValue(entry.value());
* return;
* }
* // Key is absent
* q.insert(q.absentEntry(), q.wrapValueAsValue(mappingFunction.apply(q.queriedKey().get())));
* returnValue.returnValue(q.entry().value());
* }</pre>
*/
default void computeIfAbsent(MapQueryContext<K, V, R> q,
Function<? super K, ? extends V> mappingFunction,
ReturnValue<V> returnValue) {
if (tryReturnCurrentValueIfPresent(q, returnValue))
return;
// Key is absent
q.insert(q.absentEntry(), q.wrapValueAsValue(mappingFunction.apply(q.queriedKey().get())));
returnValue.returnValue(q.entry().value());
}
/**
* Backing {@link ChronicleMap#remove(Object)} method.
*
* @implNote the default implementation is equivalent to <pre>{@code
* // We cannot read the previous value under read lock, because then we will need
* // to release the read lock -> acquire write lock, the value might be updated in
* // between, that will break ConcurrentMap.remove() atomicity guarantee. So, we acquire
* // update lock from the start:
* q.updateLock().lock();
* MapEntry<K, V> entry = q.entry();
* if (entry != null) {
* returnValue.returnValue(entry.value());
* q.remove(entry);
* }}</pre>
*/
default void remove(MapQueryContext<K, V, R> q, ReturnValue<V> returnValue) {
// We cannot read the previous value under read lock, because then we will need
// to release the read lock -> acquire write lock, the value might be updated in
@@ -90,6 +216,24 @@ public interface MapMethods<K, V, R> {
}
}
/**
* Backing {@link ChronicleMap#remove(Object, Object)} method.
*
* @implNote the default implementation is equivalent to <pre>{@code
* // remove(key, value) should find the entry & remove most of the time,
* // so don't try to check key presence and value equivalence under read lock first,
* // as in putIfAbsent()/acquireUsing(), start with update lock:
* q.updateLock().lock();
* MapEntry<K, V> entry = q.entry();
* if (entry != null && Data.bytesEquivalent(entry.value(), value)) {
* q.remove(entry);
* return true;
* } else {
* return false;
* }}</pre>
*
* @return if the entry was removed
*/
default boolean remove(MapQueryContext<K, V, R> q, Data<V, ?> value) {
// remove(key, value) should find the entry & remove most of the time,
// so don't try to check key presence and value equivalence under read lock first,
@@ -104,6 +248,20 @@ public interface MapMethods<K, V, R> {
}
}
/**
* Backing {@link ChronicleMap#replace(Object, Object)} method.
*
* @implNote the default implementation is equivalent to <pre>{@code
* // replace(key, value) should find the key & put the value most of the time,
* // so don't try to check key presence under read lock first,
* // as in putIfAbsent()/acquireUsing(), start with update lock:
* q.updateLock().lock();
* MapEntry<K, V> entry = q.entry();
* if (entry != null) {
* returnValue.returnValue(entry.value());
* q.replaceValue(entry, value);
* }}</pre>
*/
default void replace(MapQueryContext<K, V, R> q,
Data<V, ?> value, ReturnValue<V> returnValue) {
// replace(key, value) should find the key & put the value most of the time,
@@ -117,6 +275,24 @@ public interface MapMethods<K, V, R> {
}
}
/**
* Backing {@link ChronicleMap#replace(Object, Object, Object)} method.
*
* @implNote the default implementation is equivalent to <pre>{@code
* // replace(key, old, new) should find the entry & put new value most of the time,
* // so don't try to check key presence and value equivalence under read lock first,
* // as in putIfAbsent()/acquireUsing(), start with update lock:
* q.updateLock().lock();
* MapEntry<K, V> entry = q.entry();
* if (entry != null && Data.bytesEquivalent(((MapEntry<K, V>) entry).value(), oldValue)) {
* q.replaceValue(entry, newValue);
* return true;
* } else {
* return false;
* }}</pre>
*
* @return if the entry was replaced
*/
default boolean replace(MapQueryContext<K, V, R> q,
Data<V, ?> oldValue, Data<V, ?> newValue) {
// replace(key, old, new) should find the entry & put new value most of the time,
@@ -132,9 +308,30 @@ public interface MapMethods<K, V, R> {
}
}
/**
* Backing {@link ChronicleMap#compute(Object, BiFunction)} method.
*
* @implNote the default implementation is equivalent to <pre>{@code
* q.updateLock().lock();
* MapEntry<K, V> entry = q.entry();
* V oldValue = entry != null ? entry.value().get() : null;
* V newValue = remappingFunction.apply(q.queriedKey().get(), oldValue);
* if (newValue != null) {
* Data<V, ?> newValueData = q.wrapValueAsValue(newValue);
* if (entry != null) {
* q.replaceValue(entry, newValueData);
* } else {
* q.insert(q.absentEntry(), newValueData);
* entry = q.entry();
* }
* returnValue.returnValue(entry.value());
* } else if (entry != null) {
* q.remove(entry);
* }}</pre>
*/
default void compute(MapQueryContext<K, V, R> q,
BiFunction<? super K, ? super V, ? extends V> remappingFunction,
ReturnValue<V> returnValue) {
BiFunction<? super K, ? super V, ? extends V> remappingFunction,
ReturnValue<V> returnValue) {
q.updateLock().lock();
MapEntry<K, V> entry = q.entry();
V oldValue = entry != null ? entry.value().get() : null;
@@ -145,16 +342,76 @@ public interface MapMethods<K, V, R> {
q.replaceValue(entry, newValueData);
} else {
q.insert(q.absentEntry(), newValueData);
entry = q.entry();
assert entry != null;
}
returnValue.returnValue(newValueData);
returnValue.returnValue(entry.value());
} else if (entry != null) {
q.remove(entry);
}
}
/**
* Backing {@link ChronicleMap#computeIfPresent(Object, BiFunction)} method.
*
* @implNote the default implementation is equivalent to <pre>{@code
* q.updateLock().lock();
* MapEntry<K, V> entry = q.entry();
* if (entry != null) {
* V oldValue = entry.value().get();
* V newValue = remappingFunction.apply(q.queriedKey().get(), oldValue);
* if (newValue != null ) {
* q.replaceValue(entry, q.wrapValueAsValue(newValue));
* returnValue.returnValue(q.entry().value());
* } else {
* q.remove(entry);
* }
* }}</pre>
*/
default void computeIfPresent(MapQueryContext<K, V, R> q,
BiFunction<? super K, ? super V, ? extends V> remappingFunction,
ReturnValue<V> returnValue) {
q.updateLock().lock();
MapEntry<K, V> entry = q.entry();
if (entry != null) {
V oldValue = entry.value().get();
V newValue = remappingFunction.apply(q.queriedKey().get(), oldValue);
if (newValue != null ) {
q.replaceValue(entry, q.wrapValueAsValue(newValue));
returnValue.returnValue(q.entry().value());
} else {
q.remove(entry);
}
}
}
/**
* Backing {@link ChronicleMap#merge(Object, Object, BiFunction)} method.
*
* @implNote the default implementation is equivalent to <pre>{@code
* q.updateLock().lock();
* Data<V, ?> newValueData;
* MapEntry<K, V> entry = q.entry();
* if (entry != null) {
* V oldValue = entry.value().get();
* V newValue = remappingFunction.apply(oldValue, value.get());
* if (newValue == null) {
* q.remove(entry);
* return;
* }
* newValueData = q.wrapValueAsValue(newValue);
* q.replaceValue(entry, newValueData);
* } else {
* newValueData = value;
* q.insert(q.absentEntry(), newValueData);
* entry = q.entry();
* }
* returnValue.returnValue(entry.value());
* }</pre>
*/
default void merge(MapQueryContext<K, V, R> q, Data<V, ?> value,
BiFunction<? super V, ? super V, ? extends V> remappingFunction,
ReturnValue<V> returnValue) {
BiFunction<? super V, ? super V, ? extends V> remappingFunction,
ReturnValue<V> returnValue) {
q.updateLock().lock();
Data<V, ?> newValueData;
MapEntry<K, V> entry = q.entry();
@@ -170,7 +427,9 @@ public interface MapMethods<K, V, R> {
} else {
newValueData = value;
q.insert(q.absentEntry(), newValueData);
entry = q.entry();
assert entry != null;
}
returnValue.returnValue(newValueData);
returnValue.returnValue(entry.value());
}
}
src/main/java/net/openhft/chronicle/map/MapQueryContext.java
+4 -2
View File
@@ -24,8 +24,10 @@ import org.jetbrains.annotations.Nullable;
/**
* The context of {@link ChronicleMap} operations with <i>individual keys</i>
* (most: {@code get()}, {@code put()}, etc., opposed to <i>bulk</i> operations). This is the main
* context type of {@link MapMethods} and {@link MapRemoteOperations}.
* (like during {@code get()}, {@code put()}, etc., opposed to <i>bulk</i> operations).
* This is the main context type of {@link MapMethods} and {@link MapRemoteOperations}.
*
*
*
* @param <K> the map key type
* @param <V> the map value type
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+17 -5
View File
@@ -102,9 +102,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
private transient long startOfModificationIterators;
private boolean bootstrapOnlyLocalEntries;
// TODO init in builder
public transient MapRemoteOperations<K, V, R> remoteOperations =
DefaultSpi.mapRemoteOperations();
public transient MapRemoteOperations<K, V, R> remoteOperations;
transient CompiledReplicatedMapQueryContext<K, KI, MKI, V, VI, MVI, R, ?> remoteOpContext;
transient CompiledReplicatedMapIterationContext<K, KI, MKI, V, VI, MVI, R, ?> remoteItContext;
@@ -113,6 +111,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
throws IOException {
super(builder, true);
this.timeProvider = builder.timeProvider();
this.remoteOperations = (MapRemoteOperations<K, V, R>) builder.remoteOperations;
this.localIdentifier = replication.identifier();
this.bootstrapOnlyLocalEntries = replication.bootstrapOnlyLocalEntries();
@@ -290,6 +289,19 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
}
}
public void moveChange(long segmentIndex, long oldPos, long newPos) {
for (long next = modIterSet.nextSetBit(0L); next > 0L;
next = modIterSet.nextSetBit(next + 1L)) {
try {
ModificationIterator modificationIterator = modificationIterators.get((int) next);
if (modificationIterator.dropChange(segmentIndex, oldPos))
modificationIterator.raiseChange(segmentIndex, newPos);
} catch (Exception e) {
LOG.error("", e);
}
}
}
@Override
public boolean identifierCheck(@NotNull ReplicableEntry entry, int chronicleId) {
return entry.originIdentifier() == localIdentifier;
@@ -515,9 +527,9 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
modificationNotifier.onChange();
}
void dropChange(long segmentIndex, long pos) {
boolean dropChange(long segmentIndex, long pos) {
LOG.debug("drop change: id {}, segment {}, pos {}", localIdentifier, segmentIndex, pos);
changesForUpdates.clear(combine(segmentIndex, pos));
return changesForUpdates.clearIfSet(combine(segmentIndex, pos));
}
/**
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+18
View File
@@ -40,6 +40,7 @@ import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.function.Function;
import static net.openhft.chronicle.map.ChronicleMapBuilder.greatestCommonDivisor;
@@ -494,4 +495,21 @@ public class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super
return q.defaultReturnValue().returnValue();
}
}
@Override
public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
try (QueryContextInterface<K, V, R> q = queryContext(key)) {
methods.computeIfAbsent(q, mappingFunction, q.defaultReturnValue());
return q.defaultReturnValue().returnValue();
}
}
@Override
public V computeIfPresent(K key,
BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
try (QueryContextInterface<K, V, R> q = queryContext(key)) {
methods.computeIfPresent(q, remappingFunction, q.defaultReturnValue());
return q.defaultReturnValue().returnValue();
}
}
}
src/main/java/net/openhft/chronicle/map/impl/CompiledReplicatedMapIterationContext.java
+953 -998
View File
File diff suppressed because it is too large Load Diff
src/main/java/net/openhft/chronicle/map/impl/CompiledReplicatedMapQueryContext.java
+1643 -1688
View File
File diff suppressed because it is too large Load Diff
src/main/java/net/openhft/chronicle/map/replication/MapRemoteOperations.java
+140 -23
View File
@@ -18,57 +18,174 @@
package net.openhft.chronicle.map.replication;
import net.openhft.chronicle.hash.AcceptanceDecision;
import net.openhft.chronicle.hash.Data;
import net.openhft.chronicle.hash.replication.DefaultEventualConsistencyStrategy;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
import net.openhft.chronicle.map.MapAbsentEntry;
import net.openhft.chronicle.map.MapEntryOperations;
import static net.openhft.chronicle.hash.AcceptanceDecision.ACCEPT;
import static net.openhft.chronicle.hash.AcceptanceDecision.DISCARD;
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;
/**
* SPI strategy of performing remote calls and apply replication events for {@link ChronicleMap}.
*
* <p>Example: Grow-only set values CRDT: <pre><code>
* class GrowOnlySetValuedMapEntryOperations&lt;K, E&gt;
* implements MapEntryOperations&lt;K, Set&lt;E&gt;, Void&gt; {
* &#064;Override
* public Void remove(@NotNull MapEntry&lt;K, Set&lt;E&gt;&gt; entry) {
* throw new UnsupportedOperationException("Map with grow-only set values " +
* "doesn't support map value removals");
* }
* }
*
* class GrowOnlySetValuedMapRemoteOperations&lt;K, E&gt;
* implements MapRemoteOperations&lt;K, Set&lt;E&gt;, Void&gt; {
* &#064;Override
* public void put(MapRemoteQueryContext&lt;K, Set&lt;E&gt;, Void&gt; q,
* Data&lt;Set&lt;E&gt;, ?&gt; newValue) {
* MapReplicableEntry&lt;K, Set&lt;E&gt;&gt; entry = q.entry();
* if (entry != null) {
* Set&lt;E&gt; merged = new HashSet&lt;&gt;(entry.value().get());
* merged.addAll(newValue.get());
* q.replaceValue(entry, q.wrapValueAsValue(merged));
* } else {
* q.insert(q.absentEntry(), newValue);
* q.entry().updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
* }
* }
*
* &#064;Override
* public void remove(MapRemoteQueryContext&lt;K, Set&lt;E&gt;, Void&gt; q) {
* throw new UnsupportedOperationException();
* }
* }</code></pre>
*
* @param <K> the map key type
* @param <V> the map value type
* @param <R> the return type of {@link MapEntryOperations} specified fro the queried map
* @see DefaultEventualConsistencyStrategy
* @see ChronicleMapBuilder#remoteOperations(MapRemoteOperations)
*/
public interface MapRemoteOperations<K, V, R> {
default AcceptanceDecision remove(MapRemoteQueryContext<K, V, R> q) {
/**
* Handle remote {@code remove} call and {@code remove} replication event, i. e. when the entry
* with the query key ({@code q.queriedKey()}) was removed on some {@code ChronicleMap} node.
*
* @implNote the default implementation applies the remove event using {@link
* DefaultEventualConsistencyStrategy} "latest wins" strategy: <pre>{@code
* MapReplicableEntry<K, V> entry = q.entry();
* if (entry != null) {
* if (DefaultEventualConsistencyStrategy.decideOnRemoteModification(entry, q) == ACCEPT) {
* q.remove(entry);
* ReplicableEntry replicableAbsentEntry = (ReplicableEntry) q.absentEntry();
* replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
* replicableAbsentEntry.dropChanged();
* }
* } else {
* MapAbsentEntry<K, V> absentEntry = q.absentEntry();
* ReplicableEntry replicableAbsentEntry;
* if (!(absentEntry instanceof ReplicableEntry)) {
* // Note in the two following lines dummy value is inserted and removed using direct
* // entry.doXxx calls, not q.xxx(entry). The intention is to avoid calling possibly
* // overridden MapEntryOperations, because this is technical procedure of making
* // "truly absent" entry "deleted", not actual insertion and removal.
* absentEntry.doInsert(q.dummyZeroValue());
* q.entry().doRemove();
* replicableAbsentEntry = (ReplicableEntry) q.absentEntry();
* } else {
* replicableAbsentEntry = (ReplicableEntry) absentEntry;
* if (DefaultEventualConsistencyStrategy.decideOnRemoteModification(
* replicableAbsentEntry, q) == DISCARD) {
* return;
* }
* }
* replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
* replicableAbsentEntry.dropChanged();
* }}</pre>
*
* @param q the remote operation context
*/
default void remove(MapRemoteQueryContext<K, V, R> q) {
MapReplicableEntry<K, V> entry = q.entry();
if (entry != null) {
if (decideOnRemoteModification(entry, q) == ACCEPT) {
q.remove(entry);
return ACCEPT;
} else {
return DISCARD;
ReplicableEntry replicableAbsentEntry = (ReplicableEntry) q.absentEntry();
replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
replicableAbsentEntry.dropChanged();
}
} else {
MapAbsentEntry<K, V> absentEntry = q.absentEntry();
assert absentEntry != null;
if (absentEntry instanceof ReplicableEntry) {
AcceptanceDecision decision =
decideOnRemoteModification((ReplicableEntry) absentEntry, q);
return decision;
ReplicableEntry replicableAbsentEntry;
if (!(absentEntry instanceof ReplicableEntry)) {
// Note in the two following lines dummy value is inserted and removed using direct
// entry.doXxx calls, not q.xxx(entry). The intention is to avoid calling possibly
// overridden MapEntryOperations, because this is technical procedure of making
// "truly absent" entry "deleted", not actual insertion and removal.
absentEntry.doInsert(q.dummyZeroValue());
q.entry().doRemove();
replicableAbsentEntry = (ReplicableEntry) q.absentEntry();
} else {
return ACCEPT;
replicableAbsentEntry = (ReplicableEntry) absentEntry;
if (decideOnRemoteModification(replicableAbsentEntry, q) == DISCARD)
return;
}
replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
replicableAbsentEntry.dropChanged();
}
}
default AcceptanceDecision put(MapRemoteQueryContext<K, V, R> q, Data<V, ?> newValue) {
/**
* Handle remote {@code put} call or replication event, i. e. when the entry with the queried
* key ({@code q.queriedKey()}) was changed on some remote {@code ChronicleMap} node, with the
* given {@code newValue}.
*
* @implNote the default implementation applies the put event using {@link
* DefaultEventualConsistencyStrategy} "latest wins" strategy: <pre>{@code
* MapReplicableEntry<K, V> entry = q.entry();
* if (entry != null) {
* if (DefaultEventualConsistencyStrategy.decideOnRemoteModification(entry, q) == ACCEPT) {
* q.replaceValue(entry, newValue);
* entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
* entry.dropChanged();
* }
* } else {
* MapAbsentEntry<K, V> absentEntry = q.absentEntry();
* if (!(absentEntry instanceof ReplicableEntry) ||
* DefaultEventualConsistencyStrategy.decideOnRemoteModification(
* (ReplicableEntry) absentEntry, q) == ACCEPT) {
* q.insert(absentEntry, newValue);
* entry = q.entry();
* entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
* entry.dropChanged();
* }
* }}</pre>
*
* @param q the remote operation context
* @param newValue the new value to put
*/
default void put(MapRemoteQueryContext<K, V, R> q, Data<V, ?> newValue) {
MapReplicableEntry<K, V> entry = q.entry();
if (entry != null) {
if (decideOnRemoteModification(entry, q) == ACCEPT) {
q.replaceValue(entry, newValue);
return ACCEPT;
} else {
return DISCARD;
entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
entry.dropChanged();
}
} else {
MapAbsentEntry<K, V> absentEntry = q.absentEntry();
assert absentEntry != null;
if (!(absentEntry instanceof ReplicableEntry) ||
decideOnRemoteModification((ReplicableEntry) absentEntry, q) == ACCEPT) {
q.insert(absentEntry, newValue);
return ACCEPT;
} else {
return DISCARD;
entry = q.entry();
entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
entry.dropChanged();
}
}
}
src/main/java/net/openhft/chronicle/map/replication/MapRemoteQueryContext.java
+18 -1
View File
@@ -18,9 +18,12 @@
package net.openhft.chronicle.map.replication;
import net.openhft.chronicle.hash.Data;
import net.openhft.chronicle.hash.replication.HashRemoteQueryContext;
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;
import net.openhft.chronicle.map.MapQueryContext;
import org.jetbrains.annotations.Nullable;
@@ -30,7 +33,7 @@ import org.jetbrains.annotations.Nullable;
*
* @param <K> the map key type
* @param <V> the map value type
* @param <R> the return type of {@link MapEntryOperations} specialized for the queried map
* @param <R> the return type of {@link MapEntryOperations} specified for the queried map
* @see MapRemoteOperations
*/
public interface MapRemoteQueryContext<K, V, R> extends MapQueryContext<K, V, R>,
@@ -38,4 +41,18 @@ public interface MapRemoteQueryContext<K, V, R> extends MapQueryContext<K, V, R>
@Nullable
@Override
MapReplicableEntry<K, V> entry();
/**
* The value used as a tombstone, for removed entries, to save space. This value has {@linkplain
* SizeMarshaller#minEncodableSize() minimum possible size} for {@linkplain
* ChronicleMapBuilder#valueSizeMarshaller(SizeMarshaller) the configured value size
* marshaller}.
*
* <p>The returned value doesn't have object form (i. e. it's {@link Data#get()} and {@link
* Data#getUsing(Object)} methods throw {@code UnsupportedOperationException}), it has only
* bytes form, of all zero bytes.
*
* @return dummy value of all zeros, of the minimum possible size.
*/
Data<V, ?> dummyZeroValue();
}
src/main/java/net/openhft/chronicle/set/SetContext.java
+4 -2
View File
@@ -3,10 +3,12 @@ package net.openhft.chronicle.set;
import net.openhft.chronicle.hash.HashContext;
/**
* Context, in which {@link SetEntry SetEntries} are accessed.
* Context, in which {@link SetEntry SetEntries} are accessed. {@code SetContext} allows to access
* {@link SetEntryOperations}, configured for the accessed {@link ChronicleSet}. {@code SetContext}
* implements {@code SetEntryOperations} by delegation to the configured {@code entryOperations}.
*
* @param <K> the set key type
* @param <R> the return type of {@link SetEntryOperations} specialized for the queried set
* @param <R> the return type of {@link SetEntryOperations} specified for the queried set
*/
public interface SetContext<K, R> extends HashContext<K>, SetEntryOperations<K, R> {
src/main/java/net/openhft/chronicle/set/replication/SetRemoteOperations.java
+27 -18
View File
@@ -1,42 +1,50 @@
package net.openhft.chronicle.set.replication;
import net.openhft.chronicle.hash.AcceptanceDecision;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.set.SetAbsentEntry;
import static net.openhft.chronicle.hash.AcceptanceDecision.ACCEPT;
import static net.openhft.chronicle.hash.AcceptanceDecision.DISCARD;
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;
public interface SetRemoteOperations<K, R> {
default AcceptanceDecision remove(SetRemoteQueryContext<K, R> q) {
default void remove(SetRemoteQueryContext<K, R> q) {
SetReplicableEntry<K> entry = q.entry();
if (entry != null) {
if (decideOnRemoteModification(entry, q) == ACCEPT) {
q.remove(entry);
return ACCEPT;
} else {
return DISCARD;
ReplicableEntry replicableAbsentEntry = (ReplicableEntry) q.absentEntry();
replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
replicableAbsentEntry.dropChanged();
}
} else {
SetAbsentEntry<K> absentEntry = q.absentEntry();
assert absentEntry != null;
if (absentEntry instanceof ReplicableEntry) {
return decideOnRemoteModification((ReplicableEntry) absentEntry, q);
ReplicableEntry replicableAbsentEntry;
if (!(absentEntry instanceof ReplicableEntry)) {
// Note in the two following lines dummy entry is inserted and removed using direct
// entry.doXxx calls, not q.xxx(entry). The intention is to avoid calling possibly
// overridden SetEntryOperations, because this is technical procedure of making
// "truly absent" entry "deleted", not actual insertion and removal.
absentEntry.doInsert();
q.entry().doRemove();
replicableAbsentEntry = (ReplicableEntry) q.absentEntry();
} else {
return ACCEPT;
replicableAbsentEntry = (ReplicableEntry) absentEntry;
if (decideOnRemoteModification(replicableAbsentEntry, q) == DISCARD)
return;
}
replicableAbsentEntry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
replicableAbsentEntry.dropChanged();
}
}
default AcceptanceDecision insert(SetRemoteQueryContext<K, R> q) {
default void insert(SetRemoteQueryContext<K, R> q) {
SetReplicableEntry<K> entry = q.entry();
if (entry != null) {
if (decideOnRemoteModification(entry, q) == ACCEPT) {
return ACCEPT;
} else {
return DISCARD;
entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
entry.dropChanged();
}
} else {
SetAbsentEntry<K> absentEntry = q.absentEntry();
@@ -44,9 +52,10 @@ public interface SetRemoteOperations<K, R> {
if (!(absentEntry instanceof ReplicableEntry) ||
decideOnRemoteModification((ReplicableEntry) absentEntry, q) == ACCEPT) {
q.insert(absentEntry);
return ACCEPT;
} else {
return DISCARD;
entry = q.entry(); // q.entry() is not null after insert
assert entry != null;
entry.updateOrigin(q.remoteIdentifier(), q.remoteTimestamp());
entry.dropChanged();
}
}
}