diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java index b73730be2..74cc5200e 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java @@ -193,5 +193,16 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash { */ R mapForKey(K key, @NotNull Function 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 return type. + * @return the result of the function. + */ + R updateForKey(K key, @NotNull Mutator mutator); + } diff --git a/src/main/java/net/openhft/chronicle/map/Mutator.java b/src/main/java/net/openhft/chronicle/map/Mutator.java new file mode 100644 index 000000000..69c68455c --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/Mutator.java @@ -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. + *

+ *

This is not functional as it can alter an argument. + * + * @param the type of the mutable input + * @param the type of the result of the mutator + */ + +public interface Mutator extends Serializable { + /** + * Applies this mutator to the given mutable argument. + * + * @param t the mutator argument + * @return the result + */ + R update(T t); +} diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 04c291fe5..2724725f4 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -91,7 +91,8 @@ class StatelessChronicleMap implements ChronicleMap, 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 implements ChronicleMap, Closeable { return (R) keyValueSerializer.readObject(reader); } + @Override + public synchronized R updateForKey(K key, @NotNull Mutator 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 void writeObject(@NotNull Object function) { long start = bytes.position(); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 377cc6ba8..43ba1f2cf 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1233,6 +1233,7 @@ class StatelessServerConnector { ("mapSupportsBytes")); private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class .getName()); + public static final StatelessChronicleMap.EventId[] VALUES = StatelessChronicleMap.EventId.values(); private final KeyValueSerializer keyValueSerializer; private final VanillaChronicleMap map; private double maxEntrySizeBytes; @@ -1252,7 +1253,7 @@ class StatelessServerConnector { @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 { 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 { final K key = readKey(reader, local); final V value = using.getAndSet(null); - final Function o = (Function) reader.readObject(); - - try (WriteContext wc = map.lookupUsing(key, value, VanillaChronicleMap.LockType.WRITE_LOCK, - false)) { - Object result = o.apply(wc.value()); + final Function function = (Function) 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 { 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 mutator = (Mutator) 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 { 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 { 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 { 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 { try { writer.writeInt(map.hashCode()); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(writer, sizeLocation, e); } @@ -1442,7 +1464,7 @@ class StatelessServerConnector { 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 { } - 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 { try { writer.writeBoolean(map.isEmpty()); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(writer, sizeLocation, e); } @@ -1487,7 +1509,7 @@ class StatelessServerConnector { try { writer.writeBoolean(map.containsKey(k)); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(writer, sizeLocation, e); } } @@ -1505,7 +1527,7 @@ class StatelessServerConnector { try { writer.writeBoolean(map.containsValue(v)); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(writer, sizeLocation, e); } } @@ -1522,7 +1544,7 @@ class StatelessServerConnector { try { writeValue(map.get(k), writer, local); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(writer, sizeLocation, e); } } @@ -1555,7 +1577,7 @@ class StatelessServerConnector { 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 { try { writeValue(map.remove(key), writer, local); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(writer, sizeLocation, e); } @@ -1605,7 +1627,7 @@ class StatelessServerConnector { try { map.putAll(m); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(writer, sizeLocation, e); } @@ -1631,7 +1653,7 @@ class StatelessServerConnector { try { map.clear(); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(writer, sizeLocation, e); } @@ -1648,7 +1670,7 @@ class StatelessServerConnector { try { values = map.values(); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(reader, writer, e); } @@ -1691,7 +1713,7 @@ class StatelessServerConnector { try { ks = map.keySet(); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(reader, writer, e); } @@ -1735,7 +1757,7 @@ class StatelessServerConnector { try { entries = map.entrySet(); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(reader, writer, e); } @@ -1792,7 +1814,7 @@ class StatelessServerConnector { try { result = map.putIfAbsent(key, v); - } catch (RuntimeException e) { + } catch (Throwable e) { return sendException(writer, sizeLocation, e); } @@ -1815,7 +1837,7 @@ class StatelessServerConnector { 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 { 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 { } - 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); } diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 3e4cfb765..f1cb4eb73 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -345,8 +345,6 @@ class VanillaChronicleMap, } - - enum LockType {READ_LOCK, WRITE_LOCK} @Override @@ -375,6 +373,14 @@ class VanillaChronicleMap, } } + @Override + public R updateForKey(K key, @NotNull Mutator mutator) { + try (Context entry = lookupUsing(key, null, + LockType.WRITE_LOCK, false)) { + return mutator.update(entry.value()); + } + } + @NotNull @Override public WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue) { @@ -1102,7 +1108,7 @@ class VanillaChronicleMap, /** * Removes a key (or key-value pair) from the Segment. - * + *

*

The entry will only be removed if {@code expectedValue} equals to {@code null} or the value * previously corresponding to the specified key. * diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index 9bfc62abc..4740fc87b 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -250,7 +250,12 @@ public class ReplicationCheckingMap implements ChronicleMap { } @Override - public V acquireUsing(@NotNull K key, V usingValue) { + public R updateForKey(K key, @NotNull Mutator mutator) { + return map1.updateForKey(key, mutator); + } + + @Override + public V acquireUsing(@NotNull K key, V usingValue) { return map1.acquireUsing(key, usingValue); } diff --git a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java index 539d8db76..88edb989b 100755 --- a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java @@ -171,6 +171,11 @@ public class StatelessChronicleMapTest extends JSR166TestCase { return d.mapForKey(key, function); } + @Override + public R updateForKey(K key, @NotNull Mutator mutator) { + return d.updateForKey(key, mutator); + } + @Override public V putIfAbsent(K key, V value) { return d.putIfAbsent(key, value);