diff --git a/src/main/java/net/openhft/chronicle/map/Serializer.java b/src/main/java/net/openhft/chronicle/map/Serializer.java index 08bee2377..dda00aa18 100644 --- a/src/main/java/net/openhft/chronicle/map/Serializer.java +++ b/src/main/java/net/openhft/chronicle/map/Serializer.java @@ -18,12 +18,12 @@ package net.openhft.chronicle.map; -import net.openhft.lang.threadlocal.Provider; -import net.openhft.lang.threadlocal.ThreadLocalCopies; import net.openhft.chronicle.hash.serialization.*; import net.openhft.lang.io.Bytes; import net.openhft.lang.model.Byteable; import net.openhft.lang.model.constraints.NotNull; +import net.openhft.lang.threadlocal.Provider; +import net.openhft.lang.threadlocal.ThreadLocalCopies; /** * uses a {@code SerializationBuilder} to read and write object(s) as bytes @@ -66,9 +66,33 @@ class Serializer { } } - public void writeMarshallable(O value, @NotNull Bytes out) { + public O readMarshallable(@NotNull Bytes in, ThreadLocalCopies local) throws + IllegalStateException { + final ThreadLocalCopies copies = (local == null) ? threadLocalCopies() : local; + + final BytesReader valueReader = readerProvider.get(copies, originalReader); + + try { + long valueSize = sizeMarshaller.readSize(in); + return valueReader.read(in, valueSize, null); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + public ThreadLocalCopies threadLocalCopies() { + return readerProvider.getCopies(null); + } + + + public ThreadLocalCopies writeMarshallable(O value, @NotNull Bytes out) { + return getThreadLocalCopies(value, out, null); + } + + private ThreadLocalCopies getThreadLocalCopies(O value, Bytes out, ThreadLocalCopies copies0) { + + ThreadLocalCopies copies = (copies0 == null) ? readerProvider.getCopies(null) : copies0; - ThreadLocalCopies copies = readerProvider.getCopies(null); Object valueWriter = writerProvider.get(copies, originalWriter); copies = writerProvider.getCopies(copies); @@ -92,9 +116,10 @@ class Serializer { if (metaValueWriter != null) { assert out.limit() == out.capacity(); metaValueWriter.write(valueWriter, out, value); - } else + } else throw new UnsupportedOperationException(""); + return copies; } } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 790a80c5e..0204cdd5b 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -24,6 +24,7 @@ import net.openhft.chronicle.hash.exceptions.IORuntimeException; import net.openhft.chronicle.hash.exceptions.TimeoutRuntimeException; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; +import net.openhft.lang.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -64,7 +65,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private int maxEntrySize; private final Class kClass; private final Class vClass; - private int headroom = 1024; + static enum EventId { HEARTBEAT, @@ -244,9 +245,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized V putIfAbsent(K key, V value) { long sizeLocation = writeEvent(PUT_IF_ABSENT); - writeKey(key); - writeValue(value); - return readKey(sizeLocation); + ThreadLocalCopies local = writeKey(key); + writeValue(value, local); + return readKey(sizeLocation, local); } @@ -256,8 +257,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { throw new NullPointerException("key can not be null"); long sizeLocation = writeEvent(REMOVE_WITH_VALUE); - writeKey((K) key); - writeValue((V) value); + ThreadLocalCopies local = writeKey((K) key); + writeValue((V) value, local); // get the data back from the server return blockingFetch(sizeLocation).readBoolean(); @@ -269,9 +270,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { throw new NullPointerException("key can not be null"); long sizeLocation = writeEvent(REPLACE_WITH_OLD_AND_NEW_VALUE); - writeKey(key); - writeValue(oldValue); - writeValue(newValue); + ThreadLocalCopies local = writeKey(key); + writeValue(oldValue, local); + writeValue(newValue, local); // get the data back from the server return blockingFetch(sizeLocation).readBoolean(); @@ -283,11 +284,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { throw new NullPointerException("key can not be null"); long sizeLocation = writeEvent(REPLACE); - writeKey(key); - writeValue(value); + ThreadLocalCopies local = writeKey(key); + writeValue(value, local); // get the data back from the server - return readKey(sizeLocation); + return readKey(sizeLocation, local); } @@ -333,9 +334,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { final long sizeLocation = writeEvent(PUT_ALL); writeEntries(map); - blockingFetch(sizeLocation); - } @@ -392,10 +391,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized V get(Object key) { long sizeLocation = writeEvent(GET); - writeKey((K) key); + ThreadLocalCopies local = writeKey((K) key); // get the data back from the server - return readKey(sizeLocation); + return readKey(sizeLocation, local); } @@ -411,11 +410,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { public synchronized V put(K key, V value) { long sizeLocation = writeEvent(PUT); - writeKey(key); - writeValue(value); + ThreadLocalCopies local = writeKey(key); + writeValue(value, local); // get the data back from the server - return readKey(sizeLocation); + return readKey(sizeLocation, local); } public synchronized V remove(Object key) { @@ -431,6 +430,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { final int numberOfEntries = map.size(); int numberOfEntriesReadSoFar = 0; + ThreadLocalCopies local = null; bytes.writeStopBit(numberOfEntries); assert bytes.limit() == bytes.capacity(); @@ -453,7 +453,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { "and should" + " be of type=" + kClass); - writeKey(key); + + local = writeKey(key, local); + assert bytes.limit() == bytes.capacity(); final V value = e.getValue(); @@ -466,18 +468,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { try { assert bytes.limit() == bytes.capacity(); - writeValue(value); + writeValue(value, local); + } catch (Exception e4) { - - } catch ( - Exception e4) { - LOG.info("bytes position=" + bytes.position() + ",capacity=" + bytes.capacity() + - ", " + - "limit=" + bytes.limit()); assert bytes.limit() == bytes.capacity(); - writeValue(value); - int i = 0; + writeValue(value, local); + throw e4; } @@ -491,7 +488,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar) { - // System.out.println("bytes=" + bytes.position()); long remaining = bytes.remaining(); @@ -577,7 +573,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { // get the data back from the server Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); - + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); final Collection result = new ArrayList(); for (; ; ) { @@ -588,7 +584,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { long size = in.readInt(); for (int i = 0; i < size; i++) { - V v = keyValueSerializer.readValue(in); + + V v = keyValueSerializer.readValue(in, local); result.add(v); } @@ -615,7 +612,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { // get the data back from the server Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); - + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); final Set> result = new HashSet>(); for (; ; ) { @@ -627,8 +624,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { for (int i = 0; i < size; i++) { - K k = keyValueSerializer.readKey(in); - V v = keyValueSerializer.readValue(in); + K k = keyValueSerializer.readKey(in,local); + V v = keyValueSerializer.readValue(in,local); result.add(new Entry(k, v)); } @@ -664,7 +661,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { // get the data back from the server Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); - + final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); final Set result = new HashSet<>(); for (; ; ) { @@ -676,7 +673,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { long size = in.readInt(); for (int i = 0; i < size; i++) { - K k = keyValueSerializer.readKey(in); + K k = keyValueSerializer.readKey(in,local); result.add(k); } @@ -913,24 +910,54 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { return position; } - private void writeKey(K key) { + + private ThreadLocalCopies writeKey(K key) { try { - keyValueSerializer.writeKey(key, bytes); + return keyValueSerializer.writeKey(key, bytes); } catch (IllegalArgumentException e) { resizeBuffer(bytes.capacity() + maxEntrySize); + return null; } + + } + + + private ThreadLocalCopies writeKey(K key, ThreadLocalCopies local) { + try { + return keyValueSerializer.writeKey(key, bytes, local); + } catch (IllegalArgumentException e) { + resizeBuffer(bytes.capacity() + maxEntrySize); + return null; + } + } private V readKey(final long sizeLocation) { return keyValueSerializer.readValue(blockingFetch(sizeLocation)); } - private void writeValue(V value) { + private V readKey(final long sizeLocation, ThreadLocalCopies local) { + return keyValueSerializer.readValue(blockingFetch(sizeLocation), local); + } + + private ThreadLocalCopies writeValue(V value) { try { - keyValueSerializer.writeValue(value, bytes); + return keyValueSerializer.writeValue(value, bytes, null); } catch (IllegalArgumentException e) { resizeBuffer(bytes.capacity() + maxEntrySize); + return null; } + + } + + private ThreadLocalCopies writeValue(V value, ThreadLocalCopies local) { + try { + return keyValueSerializer.writeValue(value, bytes, local); + } catch (IllegalArgumentException e) { + resizeBuffer(bytes.capacity() + maxEntrySize); + return null; + } + } private Set readKeySet(Bytes in) { diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 80ad3962e..e2b15605a 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -22,6 +22,7 @@ import net.openhft.chronicle.hash.TcpReplicationConfig; import net.openhft.chronicle.hash.ThrottlingConfig; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; +import net.openhft.lang.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; @@ -1878,6 +1879,13 @@ class KeyValueSerializer { return valueSerializer.readMarshallable(reader); } + + V readValue(Bytes reader, ThreadLocalCopies local) { + if (nullCheck(reader)) return null; + return valueSerializer.readMarshallable(reader, local); + } + + private boolean nullCheck(Bytes in) { try { return in.readBoolean(); @@ -1896,29 +1904,58 @@ class KeyValueSerializer { return keySerializer.readMarshallable(reader); } + K readKey(Bytes reader, ThreadLocalCopies local) { + if (nullCheck(reader)) return null; + + return keySerializer.readMarshallable(reader, local); + } + /** * write the keysize and the key to the the {@code target} buffer * * @param key the key of the map */ - void writeKey(K key, Bytes writer) { + ThreadLocalCopies writeKey(K key, Bytes writer) { writer.writeBoolean(key == null); if (key != null) - keySerializer.writeMarshallable(key, writer); + return keySerializer.writeMarshallable(key, writer); + return null; } - void writeValue(V value, Bytes writer) { + ThreadLocalCopies writeValue(V value, Bytes writer) { assert writer.limit() == writer.capacity(); writer.writeBoolean(value == null); - assert writer.limit() == writer.capacity(); + if (value != null) { - assert writer.limit() == writer.capacity(); - valueSerializer.writeMarshallable(value, writer); + return valueSerializer.writeMarshallable(value, writer); } + return null; } + ThreadLocalCopies writeKey(K key, Bytes writer, ThreadLocalCopies threadLocal) { + + writer.writeBoolean(key == null); + if (key != null) + return keySerializer.writeMarshallable(key, writer); + return null; + } + + ThreadLocalCopies writeValue(V value, Bytes writer, ThreadLocalCopies threadLocal) { + assert writer.limit() == writer.capacity(); + writer.writeBoolean(value == null); + + if (value != null) { + return valueSerializer.writeMarshallable(value, writer); + } + return null; + } + + + public ThreadLocalCopies threadLocalCopies() { + return valueSerializer.threadLocalCopies(); + } }