Commited event listening SPI / layers for efficient coding of logic for remote/local/replication queries at once

This commit is contained in:
leventov
2015-05-16 00:48:10 +07:00
parent 0c6d348cab
commit 8e1b065f9c
14 changed files with 687 additions and 10 deletions
@@ -0,0 +1,48 @@
/*
* Copyright 2015 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.hash;
import net.openhft.chronicle.map.MapEntry;
import net.openhft.chronicle.map.MapEntryOperations;
import org.jetbrains.annotations.NotNull;
/**
* Access to the <i>present</i> {@code ChronicleHash} entry.
*
* @param <K> type of the key in {@code ChronicleHash}
* @see MapEntryOperations
*/
public interface HashEntry<K> {
/**
* Returns the entry key.
*/
@NotNull Value<K, ?> key();
/**
* Removes the entry from the {@code ChronicleHash}. Returns {@code true} if the remove was
* successful, {@code false} if it failed for any reason.
*
* <p>This method if the default implementation for {@link MapEntryOperations#remove(MapEntry)},
* which might be customized over the default.
*
* @throws IllegalStateException if some locking/state conditions required to perform remove
* operation are not met
*/
boolean defaultRemove();
}
@@ -1,17 +1,19 @@
/*
* Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com
* Copyright 2015 Higher Frequency Trading
*
* 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.higherfrequencytrading.com
*
* http://www.apache.org/licenses/LICENSE-2.0
* 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
*
* 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.
* 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.hash;
@@ -34,6 +36,7 @@ import java.util.ConcurrentModificationException;
*
* @param <K> key type
*/
@Deprecated
public interface KeyContext<K> extends AutoCloseable, InterProcessReadWriteUpdateLock {
/**
@@ -0,0 +1,97 @@
/*
* Copyright 2015 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.hash;
import net.openhft.chronicle.bytes.*;
import net.openhft.chronicle.algo.hashing.LongHashFunction;
import org.jetbrains.annotations.Nullable;
/**
* Dual bytes/object access to keys/values/elements.
*
* <p>Bytes access: {@link #access()} + {@link #handle()} + {@link #size()}.
*
* <p>Object access: {@link #get()}.
*
* In most cases, each particular value wraps either some object or some bytes. Object is marshalled
* to bytes lazily on demand, and bytes are lazily deserialized to object, accordingly.
*
* @param <V> type of the accessed objects
* @param <T> type of the handle for bytes access
*/
public interface Value<V, T> {
ReadAccess<T> access();
T handle();
long offset();
long size();
default long hash(LongHashFunction f) {
return f.hash(handle(), access(), offset(), size());
}
default <T2> boolean equivalent(ReadAccess<T2> access, T2 handle, long offset) {
return Access.equivalent(access(), handle(), offset(), access, handle, offset, size());
}
default <S, T2, A extends ReadAccess<T2>> boolean equivalent(
Accessor<S, T2, A> accessor, S source, long index) {
return equivalent(accessor.access(source), accessor.handle(source),
accessor.offset(source, index));
}
default <T2> void writeTo(WriteAccess<T2> access, T2 handle, long offset) {
Access.copy(access(), handle(), offset(), access, handle, offset, size());
}
default <T2> void writeTo(StreamingDataOutput<?, ?, T2> output) {
long offset = output.accessPositionOffset();
output.skip(size());
writeTo(output.access(), output.accessHandle(), offset);
}
default <S, T2, A extends WriteAccess<T2>> void writeTo(
Accessor<S, T2, A> accessor, S source, long index) {
writeTo(accessor.access(source), accessor.handle(source), accessor.offset(source, index));
}
/**
* Returns "cached" object, generally not eligible for using outside some context, or a block,
* synchronized with locks, or lambda, etc.
*
* If the {@code Value} is object wrapper -- this method just returns this object.
*/
V get();
/**
* Reads the object from the value's bytes, trying to reuse the given object
* (might be {@code null}).
*/
V getUsing(@Nullable V usingInstance);
static <T1, T2> boolean bytesEquivalent(Value<?, T1> v1, Value<?, T2> v2) {
if (v1.size() != v2.size())
return false;
return Access.equivalent(v1.access(), v1.handle(), v1.offset(),
v2.access(), v2.handle(), v2.offset(), v1.size());
}
}
@@ -1481,5 +1481,40 @@ public final class ChronicleMapBuilder<K, V> implements
return (int) Math.min(Math.max(2L, entries() >> 10),
Math.min(64, maxChunksPerEntry()));
}
/**
* Inject your SPI code around basic {@code ChronicleMap}'s operations with present entries:
* removing entries and replacing the entries' value.
*
* 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
* <i>internal replication operations</i>.
*/
public ChronicleMapBuilder<K, V> customizeEntryOperations(
MapEntryOperations<K, V> customEntryOperations) {
throw new UnsupportedOperationException();
}
/**
* Inject your SPI code around basic operation of inserting a new entry into the map.
*
* This affects behaviour of ordinary map.put(), map.computeIfAbsent(), etc. calls,
* as well as similar <i>remote map calls</i> and <i>internal replication operation</i>.
*/
public ChronicleMapBuilder<K, V> customizeAbsentEntryOperations(
MapAbsentEntryOperations<K, V> customAbsentEntryOperations) {
throw new UnsupportedOperationException();
}
/**
* Inject your SPI around logic of all {@code ChronicleMap}'s operations with individual keys:
* from {@link ChronicleMap#containsKey} to {@link ChronicleMap#acquireUsing} and
* {@link ChronicleMap#merge}.
*
* This affects behaviour of ordinary map calls, as well as <i>remote calls</i>.
*/
public ChronicleMapBuilder<K, V> customizeMapMethods(MapMethods<K, V> customMapMethods) {
throw new UnsupportedOperationException();
}
}
@@ -0,0 +1,30 @@
/*
* Copyright 2015 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;
import net.openhft.chronicle.hash.Value;
public interface MapAbsentEntry<K, V> {
Value<K, ?> absentKey();
boolean defaultInsert(Value<V, ?> value);
Value<V, ?> defaultValue();
}
@@ -0,0 +1,33 @@
/*
* Copyright 2015 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;
import net.openhft.chronicle.hash.Value;
import org.jetbrains.annotations.NotNull;
public interface MapAbsentEntryOperations<K, V> {
default boolean insert(@NotNull MapAbsentEntry<K, V> absentEntry, Value<V, ?> value) {
return absentEntry.defaultInsert(value);
}
default Value<V, ?> defaultValue(@NotNull MapAbsentEntry<K, V> absentEntry) {
return absentEntry.defaultValue();
}
}
@@ -0,0 +1,52 @@
/*
* Copyright 2015 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;
import net.openhft.chronicle.hash.HashEntry;
import net.openhft.chronicle.hash.Value;
/**
* Access to the <i>present</i> {@link ChronicleMap} entry.
*
* @param <K> type of the key in {@code ChronicleMap}
* @param <V> type of the value in {@code ChronicleMap}
*
* @see MapEntryOperations
*/
public interface MapEntry<K, V> extends HashEntry<K> {
/**
* Returns the entry value.
*/
Value<V, ?> value();
/**
* Replaces the entry's value with the new one. Returns {@code true} if the replace operation
* was successful, {@code false} if it failed for any reason.
*
* <p>This method if the default implementation for {@link MapEntryOperations#replaceValue},
* which might be customized over the default.
*
* @throws IllegalStateException if some locking/state conditions required to perform replace
* operation are not met
*/
boolean defaultReplaceValue(Value<V, ?> newValue);
}
@@ -0,0 +1,61 @@
/*
* Copyright 2015 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;
import net.openhft.chronicle.hash.Value;
import org.jetbrains.annotations.NotNull;
/**
* SPI interface for customizing "low-level" modification operations on the <i>present</i>
* {@link ChronicleMap} entry.
*
* @param <K> type of the key in {@code ChronicleMap}
* @param <V> type of the value in {@code ChronicleMap}
*/
public interface MapEntryOperations<K, V> {
/**
* Removes the given entry from the {@code ChronicleHash}. Returns {@code true} if the remove
* was successful, {@code false} if it failed for any reason.
*
* @implNote simply delegates to {@link MapEntry#defaultRemove()}.
*
* @param entry the entry to remove
* @throws IllegalStateException if some locking/state conditions required to perform remove
* operation are not met
*/
default boolean remove(@NotNull MapEntry<K, V> entry) {
return entry.defaultRemove();
}
/**
* Replaces the given entry's value with the new one. Returns {@code true} if the replace
* operation was successful, {@code false} if it failed for any reason.
*
* @implNote simply delegates to {@link MapEntry#defaultReplaceValue(Value)
* entry.defaultReplace(newValue)}
*
* @param entry the entry to replace the value in
* @throws IllegalStateException if some locking/state conditions required to perform replace
* operation are not met
*/
default boolean replaceValue(@NotNull MapEntry<K, V> entry, Value<V, ?> newValue) {
return entry.defaultReplaceValue(newValue);
}
}
@@ -18,6 +18,7 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.KeyContext;
@Deprecated
public interface MapKeyContext<K, V> extends KeyContext<K> {
long valueOffset();
@@ -0,0 +1,179 @@
/*
* Copyright 2015 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;
import net.openhft.chronicle.hash.Value;
import java.util.function.BiFunction;
import java.util.function.Function;
import static net.openhft.chronicle.hash.Value.bytesEquivalent;
/**
* SPI interface for customizing behaviour of the specific Map's methods with individual keys.
*/
public interface MapMethods<K, V> {
default boolean containsKey(MapQueryContext<K, V> q) {
return q.entry() != null;
}
/**
* Backing {@link ChronicleMap#get}, {@link ChronicleMap#getUsing} and
* {@link ChronicleMap#getOrDefault} methods.
*/
default void get(MapQueryContext<K, V> q, ReturnValue<? super V> returnValue) {
MapMethodsSupport.returnCurrentValueIfPresent(q, returnValue);
}
default boolean put(MapQueryContext<K, V> q,
Value<V, ?> value, ReturnValue<? super 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
// 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());
return q.replaceValue(entry, value);
} else {
return q.insert(q.absentEntry(), value);
}
}
default boolean putIfAbsent(MapQueryContext<K, V> q,
Value<V, ?> value, ReturnValue<? super V> returnValue) {
if (MapMethodsSupport.tryReturnCurrentValueIfPresent(q, returnValue))
return false;
// Key is absent
return q.insert(q.absentEntry(), value);
}
default boolean acquireUsing(MapQueryContext<K, V> q, ReturnValue<? super V> returnValue) {
if (MapMethodsSupport.tryReturnCurrentValueIfPresent(q, returnValue))
return false;
// Key is absent
Value<V, ?> defaultValue = q.defaultValue(q.absentEntry());
boolean inserted = q.insert(q.absentEntry(), defaultValue);
returnValue.returnValue(defaultValue);
return inserted;
}
default boolean remove(MapQueryContext<K, V> q, ReturnValue<? super 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
// 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());
return q.remove(entry);
} else {
return false;
}
}
default boolean remove(MapQueryContext<K, V> q, Value<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,
// as in putIfAbsent()/acquireUsing(), start with update lock:
q.updateLock().lock();
MapEntry<K, V> entry = q.entry();
return entry != null &&
bytesEquivalent(entry.value(), value) &&
q.remove(entry);
}
default boolean replace(MapQueryContext<K, V> q,
Value<V, ?> value, ReturnValue<? super V> returnValue) {
// 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());
return q.replaceValue(entry, value);
} else {
return false;
}
}
default boolean replace(MapQueryContext<K, V> q,
Value<V, ?> oldValue, Value<V, ?> newValue) {
// 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();
return entry != null &&
bytesEquivalent(((MapEntry<K, V>) entry).value(), oldValue) &&
q.replaceValue(entry, newValue);
}
default boolean compute(MapQueryContext<K, V> q,
BiFunction<? super K, ? super V, ? extends V> remappingFunction,
Function<V, Value<V, ?>> newValueObjectToValue,
ReturnValue<? super V> returnValue) {
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) {
Value<V, ?> newValueValue = newValueObjectToValue.apply(newValue);
boolean updated;
if (entry != null) {
updated = q.replaceValue(entry, newValueValue);
} else {
updated = q.insert(q.absentEntry(), newValueValue);
}
returnValue.returnValue(newValueValue);
return updated;
} else if (entry != null) {
return q.remove(entry);
} else {
return false;
}
}
default boolean merge(MapQueryContext<K, V> q, Value<V, ?> value,
BiFunction<? super V, ? super V, ? extends V> remappingFunction,
Function<V, Value<V, ?>> newValueObjectToValue,
ReturnValue<? super V> returnValue) {
q.updateLock().lock();
Value<V, ?> newValueValue;
boolean updated;
MapEntry<K, V> entry = q.entry();
if (entry != null) {
V oldValue = entry.value().get();
V newValue = remappingFunction.apply(oldValue, value.get());
if (newValue == null)
return q.remove(entry);
newValueValue = newValueObjectToValue.apply(newValue);
updated = q.replaceValue(entry, newValueValue);
} else {
newValueValue = value;
updated = q.insert(q.absentEntry(), newValueValue);
}
returnValue.returnValue(newValueValue);
return updated;
}
}
@@ -0,0 +1,51 @@
/*
* Copyright 2015 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;
final class MapMethodsSupport {
static <K, V> void returnCurrentValueIfPresent(
MapQueryContext<K, V> q, ReturnValue<? super V> returnValue) {
MapEntry<K, V> entry = q.entry();
if (entry != null)
returnValue.returnValue(entry.value());
}
static <K, V> boolean tryReturnCurrentValueIfPresent(
MapQueryContext<K, V> q, ReturnValue<? super V> returnValue) {
if (q.readLock().tryLock()) {
MapEntry<K, V> entry = q.entry();
if (entry != null) {
returnValue.returnValue(entry.value());
return true;
}
// Key is absent
q.readLock().unlock();
}
q.updateLock().lock();
MapEntry<K, V> entry = q.entry();
if (entry != null) {
returnValue.returnValue(entry.value());
return true;
}
return false;
}
private MapMethodsSupport() {}
}
@@ -0,0 +1,55 @@
/*
* Copyright 2015 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;
import net.openhft.chronicle.hash.Value;
import net.openhft.chronicle.hash.locks.InterProcessReadWriteUpdateLock;
import org.jetbrains.annotations.Nullable;
/**
* The context of Map operations with <i>individual keys</i> (most: get(), put(), etc., opposed
* to bulk operations).
*
* @param <K>
* @param <V>
*/
public interface MapQueryContext<K, V> extends InterProcessReadWriteUpdateLock,
MapEntryOperations<K, V>, MapAbsentEntryOperations<K, V> {
/**
* Returns the queried key as a value.
*/
Value<K, ?> queriedKey();
/**
* Returns the entry access object, if the entry with the queried key is <i>present</i>
* in the map, returns {@code null} is the entry is <i>absent</i>. Might acquire
* {@link #readLock} before searching for the key, if the context is not locked yet.
*/
@Nullable
MapEntry<K, V> entry();
/**
* Returns the special <i>absent entry</i> object, if the entry with the queried key
* is <i>absent</i> in the map, returns {@code null}, if the entry is <i>present</i>. Might
* acquire {@link #readLock} before searching for the key, if the context is not locked yet.
*/
@Nullable
MapAbsentEntry<K, V> absentEntry();
}
@@ -0,0 +1,26 @@
/*
* Copyright 2015 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;
import net.openhft.chronicle.hash.Value;
import org.jetbrains.annotations.NotNull;
public interface ReturnValue<V> {
void returnValue(@NotNull Value<? extends V, ?> value);
}