HCOLL-119 Add Lamda support to SHM

This commit is contained in:
Rob Austin
2014-07-24 11:07:36 +01:00
parent 91b546561a
commit fd91368197
15 changed files with 959 additions and 1609 deletions
@@ -0,0 +1,325 @@
/*
* Copyright 2014 Higher Frequency Trading
* <p/>
* http://www.higherfrequencytrading.com
* <p/>
* 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
* <p/>
* http://www.apache.org/licenses/LICENSE-2.0
* <p/>
* 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.collections.SharedHashMap;
import java.util.Map;
import java.util.Objects;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
/**
* @author Rob Austin.
*/
public interface ChronicleMap<K, V> extends SharedHashMap<K, V> { // }, ConcurrentHashMap3 {
/**
* {@inheritDoc}
*
* @throws ClassCastException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
* @implNote This implementation assumes that the ConcurrentMap cannot contain null values and {@code
* get()} returning null unambiguously means the key is absent. Implementations which support null values
* <strong>must</strong> override this default implementation.
* @since 1.8
*/
default V getOrDefault(Object key, V defaultValue) {
V v;
return ((v = get(key)) != null) ? v : defaultValue;
}
/**
* {@inheritDoc}
*
* @throws NullPointerException {@inheritDoc}
* @implSpec The default implementation is equivalent to, for this {@code map}:
* <pre> {@code
* for ((Map.Entry<K, V> entry : map.entrySet())
* action.accept(entry.getKey(), entry.getValue());
* }</pre>
* @implNote The default implementation assumes that {@code IllegalStateException} thrown by {@code
* getKey()} or {@code getValue()} indicates that the entry has been removed and cannot be processed.
* Operation continues for subsequent entries.
* @since 1.8
*/
default void forEach(BiConsumer<? super K, ? super V> action) {
Objects.requireNonNull(action);
for (Map.Entry<K, V> entry : entrySet()) {
K k;
V v;
try {
k = entry.getKey();
v = entry.getValue();
} catch (IllegalStateException ise) {
// this usually means the entry is no longer in the map.
continue;
}
action.accept(k, v);
}
}
/**
* {@inheritDoc}
*
* @throws UnsupportedOperationException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
* @throws ClassCastException {@inheritDoc}
* @throws IllegalArgumentException {@inheritDoc}
* @implSpec <p>The default implementation is equivalent to, for this {@code map}:
* <pre> {@code
* for ((Map.Entry<K, V> entry : map.entrySet())
* do {
* K k = entry.getKey();
* V v = entry.getValue();
* } while(!replace(k, v, function.apply(k, v)));
* }</pre>
*
* The default implementation may retry these steps when multiple threads attempt updates including
* potentially calling the function repeatedly for a given key.
*
* <p>This implementation assumes that the ConcurrentMap cannot contain null values and {@code get()}
* returning null unambiguously means the key is absent. Implementations which support null values
* <strong>must</strong> override this default implementation.
* @since 1.8
*/
default void replaceAll(BiFunction<? super K, ? super V, ? extends V> function) {
Objects.requireNonNull(function);
forEach((k, v) -> {
while (!replace(k, v, function.apply(k, v))) {
// v changed or k is gone
if ((v = get(k)) == null) {
// k is no longer in the map.
break;
}
}
});
}
/**
* {@inheritDoc}
*
* @throws UnsupportedOperationException {@inheritDoc}
* @throws ClassCastException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
* @implSpec The default implementation is equivalent to the following steps for this {@code map}, then
* returning the current value or {@code null} if now absent:
*
* <pre> {@code
* if (map.get(key) == null) {
* V newValue = mappingFunction.apply(key);
* if (newValue != null)
* return map.putIfAbsent(key, newValue);
* }
* }</pre>
*
* The default implementation may retry these steps when multiple threads attempt updates including
* potentially calling the mapping function multiple times.
*
* <p>This implementation assumes that the ConcurrentMap cannot contain null values and {@code get()}
* returning null unambiguously means the key is absent. Implementations which support null values
* <strong>must</strong> override this default implementation.
* @since 1.8
*/
default V computeIfAbsent(K key,
Function<? super K, ? extends V> mappingFunction) {
Objects.requireNonNull(mappingFunction);
V v, newValue;
return ((v = get(key)) == null &&
(newValue = mappingFunction.apply(key)) != null &&
(v = putIfAbsent(key, newValue)) == null) ? newValue : v;
}
/**
* {@inheritDoc}
*
* @throws UnsupportedOperationException {@inheritDoc}
* @throws ClassCastException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
* @implSpec The default implementation is equivalent to performing the following steps for this {@code
* map}, then returning the current value or {@code null} if now absent. :
*
* <pre> {@code
* if (map.get(key) != null) {
* V oldValue = map.get(key);
* V newValue = remappingFunction.apply(key, oldValue);
* if (newValue != null)
* map.replace(key, oldValue, newValue);
* else
* map.remove(key, oldValue);
* }
* }</pre>
*
* The default implementation may retry these steps when multiple threads attempt updates including
* potentially calling the remapping function multiple times.
*
* <p>This implementation assumes that the ConcurrentMap cannot contain null values and {@code get()}
* returning null unambiguously means the key is absent. Implementations which support null values
* <strong>must</strong> override this default implementation.
* @since 1.8
*/
default V computeIfPresent(K key,
BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
Objects.requireNonNull(remappingFunction);
V oldValue;
while ((oldValue = get(key)) != null) {
V newValue = remappingFunction.apply(key, oldValue);
if (newValue != null) {
if (replace(key, oldValue, newValue))
return newValue;
} else if (remove(key, oldValue))
return null;
}
return oldValue;
}
/**
* {@inheritDoc}
*
* @throws UnsupportedOperationException {@inheritDoc}
* @throws ClassCastException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
* @implSpec The default implementation is equivalent to performing the following steps for this {@code
* map}, then returning the current value or {@code null} if absent:
*
* <pre> {@code
* V oldValue = map.get(key);
* V newValue = remappingFunction.apply(key, oldValue);
* if (oldValue != null ) {
* if (newValue != null)
* map.replace(key, oldValue, newValue);
* else
* map.remove(key, oldValue);
* } else {
* if (newValue != null)
* map.putIfAbsent(key, newValue);
* else
* return null;
* }
* }</pre>
*
* The default implementation may retry these steps when multiple threads attempt updates including
* potentially calling the remapping function multiple times.
*
* <p>This implementation assumes that the ConcurrentMap cannot contain null values and {@code get()}
* returning null unambiguously means the key is absent. Implementations which support null values
* <strong>must</strong> override this default implementation.
* @since 1.8
*/
default V compute(K key,
BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
Objects.requireNonNull(remappingFunction);
V oldValue = get(key);
for (; ; ) {
V newValue = remappingFunction.apply(key, oldValue);
if (newValue == null) {
// delete mapping
if (oldValue != null || containsKey(key)) {
// something to remove
if (remove(key, oldValue)) {
// removed the old value as expected
return null;
}
// some other value replaced old value. try again.
oldValue = get(key);
} else {
// nothing to do. Leave things as they were.
return null;
}
} else {
// add or replace old mapping
if (oldValue != null) {
// replace
if (replace(key, oldValue, newValue)) {
// replaced as expected.
return newValue;
}
// some other value replaced old value. try again.
oldValue = get(key);
} else {
// add (replace if oldValue was null)
if ((oldValue = putIfAbsent(key, newValue)) == null) {
// replaced
return newValue;
}
// some other value replaced old value. try again.
}
}
}
}
/**
* {@inheritDoc}
*
* @throws UnsupportedOperationException {@inheritDoc}
* @throws ClassCastException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
* @implSpec The default implementation is equivalent to performing the following steps for this {@code
* map}, then returning the current value or {@code null} if absent:
*
* <pre> {@code
* V oldValue = map.get(key);
* V newValue = (oldValue == null) ? value :
* remappingFunction.apply(oldValue, value);
* if (newValue == null)
* map.remove(key);
* else
* map.put(key, newValue);
* }</pre>
*
* <p>The default implementation may retry these steps when multiple threads attempt updates including
* potentially calling the remapping function multiple times.
*
* <p>This implementation assumes that the ConcurrentMap cannot contain null values and {@code get()}
* returning null unambiguously means the key is absent. Implementations which support null values
* <strong>must</strong> override this default implementation.
* @since 1.8
*/
default V merge(K key, V value,
BiFunction<? super V, ? super V, ? extends V> remappingFunction) {
Objects.requireNonNull(remappingFunction);
Objects.requireNonNull(value);
V oldValue = get(key);
for (; ; ) {
if (oldValue != null) {
V newValue = remappingFunction.apply(oldValue, value);
if (newValue != null) {
if (replace(key, oldValue, newValue))
return newValue;
} else if (remove(key, oldValue)) {
return null;
}
oldValue = get(key);
} else {
if ((oldValue = putIfAbsent(key, value)) == null) {
return value;
}
}
}
}
}
@@ -0,0 +1,30 @@
/*
* Copyright 2014 Higher Frequency Trading
* <p/>
* http://www.higherfrequencytrading.com
* <p/>
* 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
* <p/>
* http://www.apache.org/licenses/LICENSE-2.0
* <p/>
* 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.collections.ChronicleMapBuilder0;
/**
* @author Rob Austin.
*/
public class ChronicleMapBuilder extends ChronicleMapBuilder0 {
public ChronicleMapBuilder(Class aClass, Class aClass2) {
super(aClass, aClass2);
}
}
@@ -0,0 +1,150 @@
/*
* Copyright 2014 Higher Frequency Trading
* <p/>
* http://www.higherfrequencytrading.com
* <p/>
* 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
* <p/>
* http://www.apache.org/licenses/LICENSE-2.0
* <p/>
* 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.collections;
import net.openhft.chronicle.map.ChronicleMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
/**
* @author Rob Austin.
*/
public class ChronicleMapBuilder0<K, V> extends SharedHashMapBuilder<K, V> implements Cloneable {
private static final Logger LOG = LoggerFactory.getLogger(ChronicleMapBuilder0.class.getName());
public ChronicleMapBuilder0(Class<K> kClass, Class<V> vClass) {
super(kClass, vClass);
}
public static <K, V> ChronicleMapBuilder0<K, V> of(Class<K> kClass, Class<V> vClass) {
return new ChronicleMapBuilder0<K, V>(kClass, vClass);
}
/**
* Not supported yet.
*
* @param transactional if the built map should be transactional
* @return this {@code SharedHashMapBuilder} back
*/
public ChronicleMapBuilder0<K, V> transactional(boolean transactional) {
this.transactional = transactional;
return this;
}
public boolean transactional() {
return transactional;
}
public ChronicleMapBuilder0 kClass(Class kClass) {
this.kClass = kClass;
return this;
}
public ChronicleMapBuilder0 vClass(Class vClass) {
this.vClass = vClass;
return this;
}
public ChronicleMapBuilder0<K, V> file(File file) {
this.file = file;
return this;
}
public ChronicleMap<K, V> create() throws IOException {
super.create();
if (kClass == null)
throw new IllegalArgumentException("missing mandatory parameter kClass");
if (vClass == null)
throw new IllegalArgumentException("missing mandatory parameter vClass");
if (file == null)
throw new IllegalArgumentException("missing mandatory parameter file");
ChronicleMapBuilder0<K, V> builder = toBuilder();
if (!canReplicate())
return new ChronicleMap0<K, V>(builder, file, kClass, vClass);
if (identifier <= 0)
throw new IllegalArgumentException("Identifier must be positive, " + identifier + " given");
final ChronicleReplicatedHashMap<K, V> result =
new ChronicleReplicatedHashMap<K, V>(builder, kClass, vClass);
if (externalReplicatorBuilder != null)
externalReplicator = applyExternalReplicator(result, externalReplicatorBuilder, kClass, vClass);
if (tcpReplicatorBuilder != null)
applyTcpReplication(result, tcpReplicatorBuilder);
if (udpReplicatorBuilder != null) {
if (tcpReplicatorBuilder == null)
LOG.warn("MISSING TCP REPLICATION : The UdpReplicator only attempts to read data (" +
"it does not enforce or guarantee delivery), you should use the UdpReplicator if " +
"you have a large number of nodes, and you wish to receive the data before it " +
"becomes available on TCP/IP. Since data delivery is not guaranteed, it is " +
"recommended that you only use the UDP" +
" " +
"Replicator " +
"in conjunction with a TCP Replicator");
applyUdpReplication(result, udpReplicatorBuilder);
}
return result;
}
ChronicleMapBuilder0<K, V> toBuilder() throws IOException {
ChronicleMapBuilder0<K, V> builder = clone();
for (int i = 0; i < 10; i++) {
if (file.exists() && file.length() > 0) {
readFile(file, builder);
break;
}
if (file.createNewFile() || file.length() == 0) {
newFile(file);
break;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
throw new IOException(e);
}
}
if (builder == null || !file.exists())
throw new FileNotFoundException("Unable to create " + file);
return builder;
}
@Override
public ChronicleMapBuilder0<K, V> clone() {
return (ChronicleMapBuilder0<K, V>) super.clone();
}
}
@@ -0,0 +1,35 @@
/*
* Copyright 2014 Higher Frequency Trading
* <p/>
* http://www.higherfrequencytrading.com
* <p/>
* 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
* <p/>
* http://www.apache.org/licenses/LICENSE-2.0
* <p/>
* 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.collections;
import net.openhft.chronicle.map.ChronicleMap;
import org.jetbrains.annotations.NotNull;
import java.io.IOException;
/**
* @author Rob Austin.
*/
public class ChronicleReplicatedHashMap<K, V> extends VanillaSharedReplicatedHashMap<K, V> implements ChronicleMap<K, V> {
public ChronicleReplicatedHashMap(@NotNull SharedHashMapBuilder builder, @NotNull Class aClass, @NotNull Class aClass2) throws IOException {
super(builder, aClass, aClass2);
}
}
@@ -0,0 +1,25 @@
/*
* Copyright 2014 Higher Frequency Trading
* <p/>
* http://www.higherfrequencytrading.com
* <p/>
* 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
* <p/>
* http://www.apache.org/licenses/LICENSE-2.0
* <p/>
* 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.collections;
/**
* @author Rob Austin.
*/
public interface ConcurrentHashMap2 {
}
@@ -0,0 +1,159 @@
/*
* Copyright 2014 Higher Frequency Trading
* <p/>
* http://www.higherfrequencytrading.com
* <p/>
* 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
* <p/>
* http://www.apache.org/licenses/LICENSE-2.0
* <p/>
* 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.collections;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.*;
/**
* @author Rob Austin.
*/
public interface ConcurrentHashMap3<K, V> {
long mappingCount();
ConcurrentHashMap.KeySetView<K, V> keySet(V mappedValue);
void forEach(long parallelismThreshold,
BiConsumer<? super K, ? super V> action);
<U> void forEach(long parallelismThreshold,
BiFunction<? super K, ? super V, ? extends U> transformer,
Consumer<? super U> action);
<U> U search(long parallelismThreshold,
BiFunction<? super K, ? super V, ? extends U> searchFunction);
<U> U reduce(long parallelismThreshold,
BiFunction<? super K, ? super V, ? extends U> transformer,
BiFunction<? super U, ? super U, ? extends U> reducer);
double reduceToDouble(long parallelismThreshold,
ToDoubleBiFunction<? super K, ? super V> transformer,
double basis,
DoubleBinaryOperator reducer);
long reduceToLong(long parallelismThreshold,
ToLongBiFunction<? super K, ? super V> transformer,
long basis,
LongBinaryOperator reducer);
int reduceToInt(long parallelismThreshold,
ToIntBiFunction<? super K, ? super V> transformer,
int basis,
IntBinaryOperator reducer);
void forEachKey(long parallelismThreshold,
Consumer<? super K> action);
<U> void forEachKey(long parallelismThreshold,
Function<? super K, ? extends U> transformer,
Consumer<? super U> action);
<U> U searchKeys(long parallelismThreshold,
Function<? super K, ? extends U> searchFunction);
K reduceKeys(long parallelismThreshold,
BiFunction<? super K, ? super K, ? extends K> reducer);
<U> U reduceKeys(long parallelismThreshold,
Function<? super K, ? extends U> transformer,
BiFunction<? super U, ? super U, ? extends U> reducer);
double reduceKeysToDouble(long parallelismThreshold,
ToDoubleFunction<? super K> transformer,
double basis,
DoubleBinaryOperator reducer);
long reduceKeysToLong(long parallelismThreshold,
ToLongFunction<? super K> transformer,
long basis,
LongBinaryOperator reducer);
int reduceKeysToInt(long parallelismThreshold,
ToIntFunction<? super K> transformer,
int basis,
IntBinaryOperator reducer);
void forEachValue(long parallelismThreshold,
Consumer<? super V> action);
<U> void forEachValue(long parallelismThreshold,
Function<? super V, ? extends U> transformer,
Consumer<? super U> action);
<U> U searchValues(long parallelismThreshold,
Function<? super V, ? extends U> searchFunction);
V reduceValues(long parallelismThreshold,
BiFunction<? super V, ? super V, ? extends V> reducer);
<U> U reduceValues(long parallelismThreshold,
Function<? super V, ? extends U> transformer,
BiFunction<? super U, ? super U, ? extends U> reducer);
double reduceValuesToDouble(long parallelismThreshold,
ToDoubleFunction<? super V> transformer,
double basis,
DoubleBinaryOperator reducer);
long reduceValuesToLong(long parallelismThreshold,
ToLongFunction<? super V> transformer,
long basis,
LongBinaryOperator reducer);
int reduceValuesToInt(long parallelismThreshold,
ToIntFunction<? super V> transformer,
int basis,
IntBinaryOperator reducer);
void forEachEntry(long parallelismThreshold,
Consumer<? super Map.Entry<K, V>> action);
<U> void forEachEntry(long parallelismThreshold,
Function<Map.Entry<K, V>, ? extends U> transformer,
Consumer<? super U> action);
<U> U searchEntries(long parallelismThreshold,
Function<Map.Entry<K, V>, ? extends U> searchFunction);
Map.Entry<K, V> reduceEntries(long parallelismThreshold,
BiFunction<Map.Entry<K, V>, Map.Entry<K, V>, ? extends Map.Entry<K, V>> reducer);
<U> U reduceEntries(long parallelismThreshold,
Function<Map.Entry<K, V>, ? extends U> transformer,
BiFunction<? super U, ? super U, ? extends U> reducer);
double reduceEntriesToDouble(long parallelismThreshold,
ToDoubleFunction<Map.Entry<K, V>> transformer,
double basis,
DoubleBinaryOperator reducer);
long reduceEntriesToLong(long parallelismThreshold,
ToLongFunction<Map.Entry<K, V>> transformer,
long basis,
LongBinaryOperator reducer);
int reduceEntriesToInt(long parallelismThreshold,
ToIntFunction<Map.Entry<K, V>> transformer,
int basis,
IntBinaryOperator reducer);
}
@@ -0,0 +1,36 @@
/*
* Copyright 2014 Higher Frequency Trading
* <p/>
* http://www.higherfrequencytrading.com
* <p/>
* 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
* <p/>
* http://www.apache.org/licenses/LICENSE-2.0
* <p/>
* 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.collections;
import net.openhft.chronicle.map.ChronicleMap;
import java.io.File;
import java.io.IOException;
/**
* @author Rob Austin.
*/
class ChronicleMap0<K, V> extends VanillaSharedHashMap<K, V> implements ChronicleMap<K, V> {
public ChronicleMap0(ChronicleMapBuilder0<K, V> builder, File file, Class<K> kClass, Class<V> vClass)
throws IOException {
super(builder, file, kClass, vClass);
}
}