HCOLL-81 make it clear that mapForKey is read only as it uses a Function and updateForKey is read/write and takes a mutator.

This commit is contained in:
Peter Lawrey
2014-11-15 12:56:37 +00:00
parent 80ba83a669
commit 3750bc4f8e
7 changed files with 120 additions and 35 deletions
@@ -193,5 +193,16 @@ public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>, ChronicleHash {
*/
<R> R mapForKey(K key, @NotNull Function<? super V, R> function);
/**
* Apply a mutator to the value for a key and return a result.
* A write lock is assumed.
*
* @param key to apply the mapping to
* @param mutator to alter the value and calculate a result
* @param <R> return type.
* @return the result of the function.
*/
<R> R updateForKey(K key, @NotNull Mutator<? super V, R> mutator);
}
@@ -0,0 +1,22 @@
package net.openhft.chronicle.map;
import java.io.Serializable;
/**
* Represents a mutator that accepts one mutable argument, which it may alter and produces a result.
* <p/>
* <p>This is not functional as it can alter an argument.
*
* @param <T> the type of the mutable input
* @param <R> the type of the result of the mutator
*/
public interface Mutator<T, R> extends Serializable {
/**
* Applies this mutator to the given mutable argument.
*
* @param t the mutator argument
* @return the result
*/
R update(T t);
}
@@ -91,7 +91,8 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
PUT_ALL,
PUT_ALL_WITHOUT_ACC,
HASH_CODE,
MAP_FOR_KEY
MAP_FOR_KEY,
UPDATE_FOR_KEY
}
private long transactionID;
@@ -470,6 +471,19 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
return (R) keyValueSerializer.readObject(reader);
}
@Override
public synchronized <R> R updateForKey(K key, @NotNull Mutator<? super V, R> mutator) {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final long sizeLocation = writeEventAnSkip(UPDATE_FOR_KEY);
writeKey(key, local);
writeObject(mutator);
final Bytes reader = blockingFetch(sizeLocation);
return (R) keyValueSerializer.readObject(reader);
}
private <R> void writeObject(@NotNull Object function) {
long start = bytes.position();
@@ -1233,6 +1233,7 @@ class StatelessServerConnector<K, V> {
("mapSupportsBytes"));
private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class
.getName());
public static final StatelessChronicleMap.EventId[] VALUES = StatelessChronicleMap.EventId.values();
private final KeyValueSerializer<K, V> keyValueSerializer;
private final VanillaChronicleMap<K, ?, ?, V, ?, ?> map;
private double maxEntrySizeBytes;
@@ -1252,7 +1253,7 @@ class StatelessServerConnector<K, V> {
@NotNull final ByteBufferBytes reader) {
final StatelessChronicleMap.EventId event = StatelessChronicleMap.EventId.values()[eventId];
final StatelessChronicleMap.EventId event = VALUES[eventId];
// these methods don't return a result
switch (event) {
@@ -1332,6 +1333,9 @@ class StatelessServerConnector<K, V> {
case MAP_FOR_KEY:
return mapForKey(reader, writer, sizeLocation);
case UPDATE_FOR_KEY:
return updateForKey(reader, writer, sizeLocation);
default:
throw new IllegalStateException("unsupported event=" + event);
@@ -1348,14 +1352,12 @@ class StatelessServerConnector<K, V> {
final K key = readKey(reader, local);
final V value = using.getAndSet(null);
final Function<V, ?> o = (Function<V, ?>) reader.readObject();
try (WriteContext<K, V> wc = map.lookupUsing(key, value, VanillaChronicleMap.LockType.WRITE_LOCK,
false)) {
Object result = o.apply(wc.value());
final Function<V, ?> function = (Function<V, ?>) reader.readObject();
try {
Object result = map.mapForKey(key, function);
keyValueSerializer.writeObject(result, writer);
} catch (RuntimeException e) {
} catch (Throwable e) {
LOG.info("", e);
return sendException(writer, sizeLocation, e);
} finally {
@@ -1364,9 +1366,29 @@ class StatelessServerConnector<K, V> {
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
public Work updateForKey(ByteBufferBytes reader, Bytes writer, long sizeLocation) {
final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies();
final K key = readKey(reader, local);
final V value = using.getAndSet(null);
final Mutator<V, ?> mutator = (Mutator<V, ?>) reader.readObject();
try {
Object result = map.updateForKey(key, mutator);
keyValueSerializer.writeObject(result, writer);
} catch (Throwable e) {
LOG.info("", e);
return sendException(writer, sizeLocation, e);
} finally {
using.set(value);
}
writeSizeAndFlags(sizeLocation, false, writer);
return null;
}
private Work removeWithValue(Bytes reader, Bytes writer, final long sizeLocation) {
if (MAP_SUPPORTS_BYTES) {
@@ -1379,7 +1401,7 @@ class StatelessServerConnector<K, V> {
final boolean result;
try {
result = map.remove(key, readValue);
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
writer.writeBoolean(result);
@@ -1400,7 +1422,7 @@ class StatelessServerConnector<K, V> {
try {
replaced = map.replace(key, oldValue, newValue);
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
writer.writeBoolean(replaced);
@@ -1414,7 +1436,7 @@ class StatelessServerConnector<K, V> {
try {
writer.writeLong(map.longSize());
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
writeSizeAndFlags(sizeLocation, false, writer);
@@ -1426,7 +1448,7 @@ class StatelessServerConnector<K, V> {
try {
writer.writeInt(map.hashCode());
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
@@ -1442,7 +1464,7 @@ class StatelessServerConnector<K, V> {
final long remaining = writer.remaining();
try {
str = map.toString();
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
@@ -1459,8 +1481,8 @@ class StatelessServerConnector<K, V> {
}
private Work sendException(Bytes writer, long sizeLocation, RuntimeException e) {
writeException(e, writer);
private Work sendException(Bytes writer, long sizeLocation, Throwable e) {
writeException(writer, e);
writeSizeAndFlags(sizeLocation, true, writer);
return null;
}
@@ -1470,7 +1492,7 @@ class StatelessServerConnector<K, V> {
try {
writer.writeBoolean(map.isEmpty());
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
@@ -1487,7 +1509,7 @@ class StatelessServerConnector<K, V> {
try {
writer.writeBoolean(map.containsKey(k));
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
}
@@ -1505,7 +1527,7 @@ class StatelessServerConnector<K, V> {
try {
writer.writeBoolean(map.containsValue(v));
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
}
@@ -1522,7 +1544,7 @@ class StatelessServerConnector<K, V> {
try {
writeValue(map.get(k), writer, local);
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
}
@@ -1555,7 +1577,7 @@ class StatelessServerConnector<K, V> {
try {
writeValue(map.put(k, v), writer, local);
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
}
@@ -1585,7 +1607,7 @@ class StatelessServerConnector<K, V> {
try {
writeValue(map.remove(key), writer, local);
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
@@ -1605,7 +1627,7 @@ class StatelessServerConnector<K, V> {
try {
map.putAll(m);
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
@@ -1631,7 +1653,7 @@ class StatelessServerConnector<K, V> {
try {
map.clear();
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
@@ -1648,7 +1670,7 @@ class StatelessServerConnector<K, V> {
try {
values = map.values();
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(reader, writer, e);
}
@@ -1691,7 +1713,7 @@ class StatelessServerConnector<K, V> {
try {
ks = map.keySet();
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(reader, writer, e);
}
@@ -1735,7 +1757,7 @@ class StatelessServerConnector<K, V> {
try {
entries = map.entrySet();
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(reader, writer, e);
}
@@ -1792,7 +1814,7 @@ class StatelessServerConnector<K, V> {
try {
result = map.putIfAbsent(key, v);
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
@@ -1815,7 +1837,7 @@ class StatelessServerConnector<K, V> {
try {
replaced = map.replace(k, v);
} catch (RuntimeException e) {
} catch (Throwable e) {
return sendException(writer, sizeLocation, e);
}
writeValue(replaced, writer, local);
@@ -1863,7 +1885,7 @@ class StatelessServerConnector<K, V> {
out.position(currentPosition);
}
private void writeException(RuntimeException e, Bytes out) {
private void writeException(Bytes out, Throwable e) {
out.writeObject(e);
}
@@ -1881,7 +1903,7 @@ class StatelessServerConnector<K, V> {
}
private Work sendException(Bytes reader, Bytes writer, RuntimeException e) {
private Work sendException(Bytes reader, Bytes writer, Throwable e) {
final long sizeLocation = reflectTransactionId(reader, writer);
return sendException(writer, sizeLocation, e);
}
@@ -345,8 +345,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
enum LockType {READ_LOCK, WRITE_LOCK}
@Override
@@ -375,6 +373,14 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
}
@Override
public <R> R updateForKey(K key, @NotNull Mutator<? super V, R> mutator) {
try (Context<K, V> entry = lookupUsing(key, null,
LockType.WRITE_LOCK, false)) {
return mutator.update(entry.value());
}
}
@NotNull
@Override
public WriteContext<K, V> acquireUsingLocked(@NotNull K key, @NotNull V usingValue) {
@@ -1102,7 +1108,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
/**
* Removes a key (or key-value pair) from the Segment.
*
* <p/>
* <p>The entry will only be removed if {@code expectedValue} equals to {@code null} or the value
* previously corresponding to the specified key.
*