From bfc2dcac25940477faaa624fd2346834e7354c4f Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 13 Nov 2014 18:55:15 +0000 Subject: [PATCH] HCOLL-81 StatelessClient : Add a visitor pattern to Chronicle Map --- .../chronicle/map/ReplicatedChronicleMap.java | 112 ++++++++++-------- .../chronicle/map/StatelessChronicleMap.java | 83 +++++++++---- .../openhft/chronicle/map/TcpReplicator.java | 57 ++++++++- .../chronicle/map/StatelessClientTest.java | 35 +++++- 4 files changed, 210 insertions(+), 77 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index dc7bd211e..c68a813f3 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -955,67 +955,77 @@ final class ReplicatedChronicleMap, return keySize == keySizeMarshaller.readSize(entry) && entry.startsWith(keyBytes); } + @Override + V putWithoutLock(ThreadLocalCopies copies, MKI metaKeyInterop, KI keyInterop, K key, V value, long hash2, boolean replaceIfPresent) { + return putWithoutLock(copies, metaKeyInterop, keyInterop, key, value, hash2, replaceIfPresent, + localIdentifier, timeProvider.currentTimeMillis()); + } + V put(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, K key, V value, long hash2, boolean replaceIfPresent, final byte identifier, final long timestamp) { writeLock(); try { - MultiMap hashLookup = hashLookupLiveAndDeleted; - long keySize = metaKeyWriter.size(keyWriter, key); - hashLookup.startSearch(hash2); - for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { - long offset = offsetFromPos(pos); - MultiStoreBytes entry = entry(offset); - if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry)) - continue; - // key is found - entry.skip(keySize); - - final long timeStampPos = entry.positionAddr(); - - if (shouldIgnore(entry, timestamp, identifier)) - return null; - - boolean wasDeleted = entry.readBoolean(); - - // if wasDeleted==true then we even if replaceIfPresent==false we can treat it - // the same way as replaceIfPresent true - if (replaceIfPresent || wasDeleted) { - - entry.positionAddr(timeStampPos); - entry.writeLong(timestamp); - entry.writeByte(identifier); - // deleted flag - entry.writeBoolean(false); - - final V prevValue = replaceValueOnPut(copies, key, value, entry, pos, - offset, !wasDeleted && !putReturnsNull, hashLookup); - - if (wasDeleted) { - // remove() would have got rid of this so we have to add it back in - hashLookupLiveOnly.put(hash2, pos); - incrementSize(); - return null; - } else { - return prevValue; - } - } else { - // we wont replaceIfPresent and the entry is not deleted - return putReturnsNull ? null : readValue(copies, entry, null); - } - } - - // key is not found - long offset = putEntry(copies, metaKeyWriter, keyWriter, key, keySize, hash2, value, - identifier, timestamp, hashLookup); - incrementSize(); - notifyPut(offset, true, key, value, null, posFromOffset(offset)); - return null; + return putWithoutLock(copies, metaKeyWriter, keyWriter, key, value, hash2, replaceIfPresent, identifier, timestamp); } finally { writeUnlock(); } } + private V putWithoutLock(ThreadLocalCopies copies, MKI metaKeyWriter, KI keyWriter, K key, V value, long hash2, boolean replaceIfPresent, byte identifier, long timestamp) { + MultiMap hashLookup = hashLookupLiveAndDeleted; + long keySize = metaKeyWriter.size(keyWriter, key); + hashLookup.startSearch(hash2); + for (long pos; (pos = hashLookup.nextPos()) >= 0L; ) { + long offset = offsetFromPos(pos); + MultiStoreBytes entry = entry(offset); + if (!keyEquals(keyWriter, metaKeyWriter, key, keySize, entry)) + continue; + // key is found + entry.skip(keySize); + + final long timeStampPos = entry.positionAddr(); + + if (shouldIgnore(entry, timestamp, identifier)) + return null; + + boolean wasDeleted = entry.readBoolean(); + + // if wasDeleted==true then we even if replaceIfPresent==false we can treat it + // the same way as replaceIfPresent true + if (replaceIfPresent || wasDeleted) { + + entry.positionAddr(timeStampPos); + entry.writeLong(timestamp); + entry.writeByte(identifier); + // deleted flag + entry.writeBoolean(false); + + final V prevValue = replaceValueOnPut(copies, key, value, entry, pos, + offset, !wasDeleted && !putReturnsNull, hashLookup); + + if (wasDeleted) { + // remove() would have got rid of this so we have to add it back in + hashLookupLiveOnly.put(hash2, pos); + incrementSize(); + return null; + } else { + return prevValue; + } + } else { + // we wont replaceIfPresent and the entry is not deleted + return putReturnsNull ? null : readValue(copies, entry, null); + } + } + + // key is not found + long offset = putEntry(copies, metaKeyWriter, keyWriter, key, keySize, hash2, value, + identifier, timestamp, hashLookup); + incrementSize(); + notifyPut(offset, true, key, value, null, posFromOffset(offset)); + return null; + } + /** * Used only with replication, its sometimes possible to receive an old ( or stale update ) from a diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 5b9be429b..9c10faa0f 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -35,6 +35,7 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.nio.channels.SocketChannel; import java.util.*; +import java.util.function.Function; import static java.nio.ByteBuffer.allocateDirect; import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; @@ -90,13 +91,15 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { TO_STRING, PUT_ALL, PUT_ALL_WITHOUT_ACC, - HASH_CODE + HASH_CODE, + MAP_FOR_KEY } private long transactionID; StatelessChronicleMap(final StatelessMapConfig builder, - final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException { + final AbstractChronicleMapBuilder chronicleMapBuilder) throws + IOException { this.builder = builder; this.keyValueSerializer = new KeyValueSerializer(chronicleMapBuilder.keyBuilder, chronicleMapBuilder.valueBuilder); @@ -105,8 +108,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { this.maxEntrySize = chronicleMapBuilder.entrySize(true); this.vClass = chronicleMapBuilder.valueBuilder.eClass; this.kClass = chronicleMapBuilder.keyBuilder.eClass; - // this.putReturnsNull = putReturnsNull; - // this.removeReturnsNull = removeReturnsNull; + attemptConnect(builder.remoteAddress()); buffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder()); @@ -136,6 +138,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { result = AbstractChannelReplicator.openSocketChannel(closeables); result.connect(builder.remoteAddress()); result.socket().setTcpNoDelay(true); + result.socket().setReuseAddress(true); doHandShaking(result); break; @@ -451,9 +454,44 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } + public synchronized R mapForKey(K key, @NotNull Function function) { + + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + final long sizeLocation = writeEventAnSkip(MAP_FOR_KEY); + + writeKey(key, local); + writeObject(function); + + final Bytes reader = blockingFetch(sizeLocation); + return (R) keyValueSerializer.readObject(reader); + } + + + private void writeObject(@NotNull Function function) { + for (; ; ) { + try { + + bytes.writeObject(function); + return; + } catch (IllegalStateException e) { + + Throwable cause = e.getCause(); + + if (cause instanceof IOException && cause.getMessage().contains("Not enough available space")) { + LOG.debug("resizing buffer"); + resizeBuffer(buffer.capacity() + 1024); + + } else + throw e; + + } + } + } + + public synchronized void putAll(Map map) { - final long sizeLocation = (putReturnsNull) ? writeEvent(PUT_ALL_WITHOUT_ACC) : writeEventAnSkip(PUT_ALL); + final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) : writeEventAnSkip(PUT_ALL); writeEntries(map); @@ -560,8 +598,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { // we will back the size estimate on what we have so far final double percentageComplete = (double) numberOfEntriesReadSoFar / (double) numberOfEntries; return (long) ((double) bytes.position() / percentageComplete); - - } @@ -571,30 +607,38 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { assert size < Integer.MAX_VALUE; - final ByteBuffer result = ByteBuffer.allocateDirect((int) size).order(ByteOrder.nativeOrder()); + final ByteBuffer result = ByteBuffer.allocate((int) size).order(ByteOrder.nativeOrder()); - final int bufferPosition = buffer.position(); + /* final int bufferPosition = buffer.position(); final int bufferLimit = buffer.limit(); - +*/ final long bytesPosition = bytes.position(); + bytes = new ByteBufferBytes(result.slice()); buffer.position(0); buffer.limit((int) bytesPosition); - for (int i = 0; i < bytesPosition; i++) { - result.put(buffer.get()); + int numberOfLongs = (int) bytesPosition / 8; + + // chunk in longs first + for (int i = 0; i < numberOfLongs; i++) { + bytes.writeLong(buffer.getLong()); + } + + for (int i = numberOfLongs * 8; i < bytesPosition; i++) { + bytes.writeByte(buffer.get()); } buffer = result; - buffer.limit(bufferLimit); + /* buffer.limit(bufferLimit); buffer.position(bufferPosition); - +*/ assert buffer.capacity() == bytes.capacity(); - bytes.limit(bytes.capacity()); + /* bytes.limit(bytes.capacity()); bytes.position(bytesPosition); - +*/ assert buffer.capacity() == size; assert buffer.capacity() == bytes.capacity(); assert bytes.limit() == bytes.capacity(); @@ -731,7 +775,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { buffer.clear(); bytes.clear(); - //bytes.limit(buffer.capacity()); bytes.write((byte) event.ordinal()); long sizeLocation = markSizeLocation(); @@ -807,6 +850,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public final String toString() { return getKey() + "=" + getValue(); } + } private Bytes blockingFetch(long sizeLocation) { @@ -954,7 +998,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { bytes.writeInt((int) size - SIZE_OF_SIZE); bytes.writeLong(transactionId); bytes.position(pos); - } @@ -978,7 +1021,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { resizeBuffer(bytes.capacity() + maxEntrySize); } - } @@ -988,7 +1030,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } catch (IllegalArgumentException e) { resizeBuffer(bytes.capacity() + maxEntrySize); } - } private V readKey(final long sizeLocation) { @@ -1005,7 +1046,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } catch (IllegalArgumentException e) { resizeBuffer(bytes.capacity() + maxEntrySize); } - } private void writeValue(V value, ThreadLocalCopies local) { @@ -1014,7 +1054,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } catch (IllegalArgumentException e) { resizeBuffer(bytes.capacity() + maxEntrySize); } - } } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 6f912692b..dff0d86e6 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -37,6 +37,8 @@ import java.nio.ByteOrder; import java.nio.channels.*; import java.util.*; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; import static java.nio.channels.SelectionKey.*; import static java.util.concurrent.TimeUnit.MILLISECONDS; @@ -1253,7 +1255,7 @@ class StatelessServerConnector { final StatelessChronicleMap.EventId event = StatelessChronicleMap.EventId.values()[eventId]; - + // these methods don't return a result switch (event) { case KEY_SET: @@ -1268,7 +1270,6 @@ class StatelessServerConnector { case PUT_WITHOUT_ACC: return put(reader); - case PUT_ALL_WITHOUT_ACC: return putAll(reader); @@ -1279,6 +1280,9 @@ class StatelessServerConnector { final long sizeLocation = reflectTransactionId(reader, writer); + + // these methods return a result + switch (event) { case LONG_SIZE: @@ -1326,6 +1330,8 @@ class StatelessServerConnector { case HASH_CODE: return hashCode(reader, writer, sizeLocation); + case MAP_FOR_KEY: + return mapForKey(reader, writer, sizeLocation); default: throw new IllegalStateException("unsupported event=" + event); @@ -1334,6 +1340,38 @@ class StatelessServerConnector { } + // an object poll of 1 using an atomic reference + private AtomicReference using = new AtomicReference(); + + public Work mapForKey(ByteBufferBytes reader, Bytes writer, long sizeLocation) { + + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + final K key = readKey(reader, local); + + V value = using.getAndSet(null); + if (value == null) { + // a simple way to create a value + value = map.getUsing(key, null); + } + + final Function o = (Function) reader.readObject(); + + try (WriteContext wc = map.acquireUsingLocked(key, value)) { + Object result = o.apply(value); + keyValueSerializer.writeObject(result, writer); + + } catch (RuntimeException 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) { @@ -1916,6 +1954,11 @@ class KeyValueSerializer { return valueSerializer.readMarshallable(reader, local); } + O readObject(Bytes reader) { + if (nullCheck(reader)) return null; + return (O) reader.readObject(); + } + private boolean nullCheck(Bytes in) { return in.readBoolean(); @@ -1976,6 +2019,16 @@ class KeyValueSerializer { } + void writeObject(Object value, Bytes writer) { + assert writer.limit() == writer.capacity(); + writer.writeBoolean(value == null); + + if (value != null) { + writer.writeObject(value); + } + + } + public ThreadLocalCopies threadLocalCopies() { return valueSerializer.threadLocalCopies(); } diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 507c94ff0..db8a40030 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -23,11 +23,13 @@ import org.junit.Assert; import org.junit.Test; import java.io.IOException; +import java.io.Serializable; import java.net.InetSocketAddress; import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.Set; +import java.util.function.Function; /** @@ -39,6 +41,37 @@ public class StatelessClientTest { public static final int SIZE = 100000; + static class StringBuilderToStringFunction implements Function, Serializable { + + @Override + public String apply(StringBuilder stringBuilder) { + return stringBuilder.toString(); + } + } + + + @Test + public void testMapForKey() throws IOException, InterruptedException { + + + try (ChronicleMap serverMap = ChronicleMapBuilder.of(Integer.class, + StringBuilder.class) + .replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) { + + serverMap.put(10, new StringBuilder("Hello World")); + + try (ChronicleMap statelessMap = ChronicleMapBuilder.of(Integer + .class, StringBuilder.class) + .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { + + String actual = (String) ((StatelessChronicleMap) statelessMap).mapForKey(10, new StringBuilderToStringFunction()); + + Assert.assertEquals("Hello World", actual); + } + } + } + + @Test(timeout = 5000) public void testBufferOverFlowPutAllAndEntrySet() throws IOException, InterruptedException { @@ -167,7 +200,6 @@ public class StatelessClientTest { .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { - serverMap.put(10, "EXAMPLE-10"); Assert.assertEquals("EXAMPLE-10", statelessMap.get(10)); @@ -301,7 +333,6 @@ public class StatelessClientTest { final ChronicleMap statelessMap2; - // server { serverMap1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)