From 258347ed71360834554fdff265eac6fdd9ad2585 Mon Sep 17 00:00:00 2001 From: leventov Date: Sat, 29 Nov 2014 20:39:00 +0700 Subject: [PATCH] HCOLL-185 fixed a number of bugs in old code and new VanillaChronicleMap and ReplicatedChronicleMap (generalized instance/bytes queries), refactoring and minor fixes, switched to bytes queries for stateless client server --- .../openhft/chronicle/map/ReaderWithSize.java | 69 ++++ .../chronicle/map/ReplicatedChronicleMap.java | 53 +-- .../openhft/chronicle/map/Replicators.java | 9 +- .../net/openhft/chronicle/map/Serializer.java | 114 ------- .../chronicle/map/StatelessChronicleMap.java | 228 +++++++------ .../openhft/chronicle/map/TcpReplicator.java | 304 ++++++------------ .../chronicle/map/VanillaChronicleMap.java | 172 +++++----- .../openhft/chronicle/map/WriterWithSize.java | 89 +++++ .../openhft/chronicle/map/SerializerTest.java | 18 +- .../chronicle/map/StatelessClientTest.java | 66 ++-- 10 files changed, 560 insertions(+), 562 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/ReaderWithSize.java delete mode 100644 src/main/java/net/openhft/chronicle/map/Serializer.java create mode 100644 src/main/java/net/openhft/chronicle/map/WriterWithSize.java diff --git a/src/main/java/net/openhft/chronicle/map/ReaderWithSize.java b/src/main/java/net/openhft/chronicle/map/ReaderWithSize.java new file mode 100644 index 000000000..0c02ac82b --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/ReaderWithSize.java @@ -0,0 +1,69 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com + * + * 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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.chronicle.hash.serialization.BytesReader; +import net.openhft.chronicle.hash.serialization.SizeMarshaller; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.threadlocal.Provider; +import net.openhft.lang.threadlocal.ThreadLocalCopies; +import org.jetbrains.annotations.Nullable; + + +final class ReaderWithSize { + private final SizeMarshaller sizeMarshaller; + private final BytesReader originalReader; + private final Provider> readerProvider; + ReaderWithSize(SerializationBuilder serializationBuilder) { + sizeMarshaller = serializationBuilder.sizeMarshaller(); + originalReader = serializationBuilder.reader(); + readerProvider = (Provider>) Provider.of(originalReader.getClass()); + } + + public ThreadLocalCopies getCopies(ThreadLocalCopies copies) { + return readerProvider.getCopies(copies); + } + + public T read(Bytes in, @Nullable ThreadLocalCopies copies) { + long size = sizeMarshaller.readSize(in); + copies = readerProvider.getCopies(copies); + BytesReader reader = readerProvider.get(copies, originalReader); + return reader.read(in, size); + } + + public T readNullable(Bytes in, @Nullable ThreadLocalCopies copies) { + if (in.readBoolean()) + return null; + return read(in, copies); + } + + public BytesReader readerForLoop(@Nullable ThreadLocalCopies copies) { + copies = readerProvider.getCopies(copies); + return readerProvider.get(copies, originalReader); + } + + public T readInLoop(Bytes in, BytesReader reader) { + long size = sizeMarshaller.readSize(in); + return reader.read(in, size); + } + + public T readNullableInLoop(Bytes in, BytesReader reader) { + if (in.readBoolean()) + return null; + return readInLoop(in, reader); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index e3e3f038f..f41c1d18a 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -744,10 +744,13 @@ final class ReplicatedChronicleMap>instance(), BytesBytesInterop.INSTANCE, keyBytes, keyBytes.remaining(), - keyBytesToInstance, null, null, null, hash2, true, + keyBytesToInstance, null, null, outputValueBytesToInstance, + hash2, readValueToLazyBytes, true, timestamp, identifier, true, true); if (!removed && LOG.isDebugEnabled()) { LOG.debug("Segment.remoteRemove() : key=" + keyBytes.toString().trim() + @@ -971,19 +974,19 @@ final class ReplicatedChronicleMap, - VB, VBI, MVBI extends MetaBytesInterop> + RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> Object remove(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, GetValueInterops getValueInterops, VB expectedValue, - InstanceOrBytesToInstance toValue, - long hash2, boolean resultUnused) { + InstanceOrBytesToInstance toValue, + long hash2, ReadValue readValue, boolean resultUnused) { writeLock(null); try { return removeWithoutLock(copies, segmentState, metaKeyInterop, keyInterop, key, keySize, toKey, getValueInterops, expectedValue, toValue, - hash2, resultUnused, currentTime(), localIdentifier, false, + hash2, readValue, resultUnused, currentTime(), localIdentifier, false, expectedValue != null); } finally { if (segmentState != null) @@ -999,16 +1002,16 @@ final class ReplicatedChronicleMap, - VB, VBI, MVBI extends MetaBytesInterop> + RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> Object removeWithoutLock( @Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, GetValueInterops getValueInterops, VB expectedValue, - InstanceOrBytesToInstance toValue, - long hash2, boolean resultUnused) { + InstanceOrBytesToInstance toValue, + long hash2, ReadValue readValue, boolean resultUnused) { return removeWithoutLock(copies, segmentState, metaKeyInterop, keyInterop, key, keySize, - toKey, getValueInterops, expectedValue, toValue, hash2, resultUnused, + toKey, getValueInterops, expectedValue, toValue, hash2, readValue, resultUnused, currentTime(), localIdentifier, false, expectedValue != null); } @@ -1021,14 +1024,14 @@ final class ReplicatedChronicleMap, - VB, VBI, MVBI extends MetaBytesInterop> + RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> Object removeWithoutLock( @Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, GetValueInterops getValueInterops, VB expectedValue, - InstanceOrBytesToInstance toValue, - long hash2, boolean resultUnused, + InstanceOrBytesToInstance toValue, + long hash2, ReadValue readValue, boolean resultUnused, long timestamp, byte identifier, boolean remote, boolean booleanResult) { segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); assert identifier > 0; @@ -1061,7 +1064,7 @@ final class ReplicatedChronicleMap { - - final SizeMarshaller sizeMarshaller; - final BytesReader originalReader; - transient Provider> readerProvider; - final Object originalWriter; - transient Provider writerProvider; - final MetaBytesInterop originalMetaValueWriter; - final MetaProvider metaValueWriterProvider; - - Serializer(final SerializationBuilder serializationBuilder) { - - sizeMarshaller = serializationBuilder.sizeMarshaller(); - originalMetaValueWriter = serializationBuilder.metaInterop(); - metaValueWriterProvider = serializationBuilder.metaInteropProvider(); - - originalReader = serializationBuilder.reader(); - originalWriter = serializationBuilder.interop(); - - readerProvider = Provider.of((Class) originalReader.getClass()); - writerProvider = Provider.of((Class) originalWriter.getClass()); - } - - - - 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); - } - - - - ThreadLocalCopies writeMarshallable(O value, Bytes out, ThreadLocalCopies copies0) { - - ThreadLocalCopies copies = (copies0 == null) ? threadLocalCopies() : copies0; - - Object valueWriter = writerProvider.get(copies, originalWriter); - copies = writerProvider.getCopies(copies); - - final long valueSize; - - MetaBytesWriter metaValueWriter = null; - - if ((value instanceof Byteable)) { - valueSize = ((Byteable) value).maxSize(); - } else { - copies = writerProvider.getCopies(copies); - valueWriter = writerProvider.get(copies, originalWriter); - copies = metaValueWriterProvider.getCopies(copies); - metaValueWriter = metaValueWriterProvider.get( - copies, originalMetaValueWriter, valueWriter, value); - valueSize = metaValueWriter.size(valueWriter, value); - } - - sizeMarshaller.writeSize(out, valueSize); - - if (metaValueWriter != null) { - assert out.limit() == out.capacity(); - metaValueWriter.write(valueWriter, out, value); - } 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 62bb35fde..85d9f1a7f 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -20,6 +20,7 @@ package net.openhft.chronicle.map; import com.sun.jdi.connect.spi.ClosedConnectionException; import net.openhft.chronicle.hash.RemoteCallTimeoutException; +import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; import net.openhft.lang.io.NativeBytes; @@ -63,7 +64,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private ByteBuffer buffer; private ByteBufferBytes bytes; - private final KeyValueSerializer keyValueSerializer; + private final ReaderWithSize keyReaderWithSize; + private final WriterWithSize keyWriterWithSize; + private final ReaderWithSize valueReaderWithSize; + private final WriterWithSize valueWriterWithSize; private volatile SocketChannel clientChannel; private CloseablesManager closeables; @@ -80,7 +84,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon static enum EventId { HEARTBEAT, STATEFUL_UPDATE, - LONG_SIZE, SIZE, + LONG_SIZE, + SIZE, IS_EMPTY, CONTAINS_KEY, CONTAINS_VALUE, @@ -111,8 +116,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException { this.chronicleMapBuilder = chronicleMapBuilder; this.config = config; - this.keyValueSerializer = new KeyValueSerializer(chronicleMapBuilder.keyBuilder, - chronicleMapBuilder.valueBuilder); + keyReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.keyBuilder); + keyWriterWithSize = new WriterWithSize<>(chronicleMapBuilder.keyBuilder); + valueReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.valueBuilder); + valueWriterWithSize = new WriterWithSize<>(chronicleMapBuilder.valueBuilder); this.putReturnsNull = chronicleMapBuilder.putReturnsNull(); this.removeReturnsNull = chronicleMapBuilder.removeReturnsNull(); this.maxEntrySize = chronicleMapBuilder.entrySize(true); @@ -343,25 +350,24 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon public synchronized V putIfAbsent(K key, V value) { - if (key == null) + if (key == null || value == null) throw new NullPointerException(); final long sizeLocation = writeEventAnSkip(PUT_IF_ABSENT); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - writeKey(key, local); - writeValue(value, local); - return readValue(sizeLocation, local); + ThreadLocalCopies copies = writeKey(key); + copies = writeValue(value, copies); + return readValue(sizeLocation, copies); } public synchronized boolean remove(Object key, Object value) { - if (key == null) - throw new NullPointerException("key can not be null"); + throw new NullPointerException(); + if (value == null) + return false; // ConcurrentHashMap compat final long sizeLocation = writeEventAnSkip(REMOVE_WITH_VALUE); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - writeKey((K) key, local); - writeValue((V) value, local); + ThreadLocalCopies copies = writeKey((K) key); + writeValue((V) value, copies); // get the data back from the server return blockingFetch(sizeLocation).readBoolean(); @@ -369,14 +375,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon public synchronized boolean replace(K key, V oldValue, V newValue) { - if (key == null) - throw new NullPointerException("key can not be null"); + if (key == null || oldValue == null || newValue == null) + throw new NullPointerException(); final long sizeLocation = writeEventAnSkip(REPLACE_WITH_OLD_AND_NEW_VALUE); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - writeKey(key, local); - writeValue(oldValue, local); - writeValue(newValue, local); + ThreadLocalCopies copies = writeKey(key); + copies = writeValue(oldValue, copies); + writeValue(newValue, copies); // get the data back from the server return blockingFetch(sizeLocation).readBoolean(); @@ -384,16 +389,15 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon public synchronized V replace(K key, V value) { - if (key == null) - throw new NullPointerException("key can not be null"); + if (key == null || value == null) + throw new NullPointerException(); final long sizeLocation = writeEventAnSkip(REPLACE); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - writeKey(key, local); - writeValue(value, local); + ThreadLocalCopies copies = writeKey(key); + copies = writeValue(value, copies); // get the data back from the server - return readValue(sizeLocation, local); + return readValue(sizeLocation, copies); } @@ -449,6 +453,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon public synchronized boolean containsKey(Object key) { + if (key == null) + throw keyNotNullNPE(); final long sizeLocation = writeEventAnSkip(CONTAINS_KEY); writeKey((K) key); @@ -457,10 +463,16 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } + private static NullPointerException keyNotNullNPE() { + return new NullPointerException("key can not be null"); + } + public synchronized boolean containsValue(Object value) { + if (value == null) + throw new NullPointerException(); final long sizeLocation = writeEventAnSkip(CONTAINS_VALUE); - writeValue((V) value); + writeValue((V) value, null); // get the data back from the server return blockingFetch(sizeLocation).readBoolean(); @@ -474,11 +486,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon public synchronized V get(Object key) { final long sizeLocation = writeEventAnSkip(GET); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - writeKey((K) key, local); + ThreadLocalCopies copies = writeKey((K) key); // get the data back from the server - return readValue(sizeLocation, local); + return readValue(sizeLocation, copies); } @@ -510,67 +521,58 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon public synchronized V remove(Object key) { if (key == null) - throw new NullPointerException("key can not be null"); + throw keyNotNullNPE(); final long sizeLocation = removeReturnsNull ? writeEvent(REMOVE_WITHOUT_ACC) : writeEventAnSkip(REMOVE); - writeKey((K) key); + ThreadLocalCopies copies = writeKey((K) key); if (removeReturnsNull) { sendWithoutAcc(bytes, sizeLocation); return null; } else // get the data back from the server - return readValue(sizeLocation); + return readValue(sizeLocation, copies); } public synchronized V put(K key, V value) { - if (key == null) + if (key == null || value == null) throw new NullPointerException(); + final long sizeLocation = + (putReturnsNull) ? writeEvent(PUT_WITHOUT_ACC) : writeEventAnSkip(PUT); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - - final long sizeLocation = (putReturnsNull) ? writeEvent(PUT_WITHOUT_ACC) : writeEventAnSkip(PUT); - - writeKey(key, local); - writeValue(value, local); - + ThreadLocalCopies copies = writeKey(key); + copies = writeValue(value, copies); if (putReturnsNull) { sendWithoutAcc(bytes, sizeLocation); return null; } else // get the data back from the server - return readValue(sizeLocation, local); + return readValue(sizeLocation, copies); } public synchronized R mapForKey(K key, @NotNull Function function) { - - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); final long sizeLocation = writeEventAnSkip(MAP_FOR_KEY); - - writeKey(key, local); + writeKey(key); writeObject(function); final Bytes reader = blockingFetch(sizeLocation); - return (R) keyValueSerializer.readObject(reader); + return (R) reader.readObject(); } @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); + writeKey(key); writeObject(mutator); final Bytes reader = blockingFetch(sizeLocation); - return (R) keyValueSerializer.readObject(reader); + return (R) reader.readObject(); } @@ -598,9 +600,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon 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); + writeEntriesForPutAll(map); if (putReturnsNull) sendWithoutAcc(bytes, sizeLocation); @@ -635,15 +638,24 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private synchronized void writeEntries(Map map) { + private synchronized void writeEntriesForPutAll(Map map) { final int numberOfEntries = map.size(); int numberOfEntriesReadSoFar = 0; - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); bytes.writeStopBit(numberOfEntries); assert bytes.limit() == bytes.capacity(); + + ThreadLocalCopies copies = keyWriterWithSize.getCopies(null); + Object keyWriter = keyWriterWithSize.writerForLoop(copies); + copies = valueWriterWithSize.getCopies(copies); + Object valueWriter = valueWriterWithSize.writerForLoop(copies); + for (final Map.Entry e : map.entrySet()) { + final K key = e.getKey(); + final V value = e.getValue(); + if (key == null || value == null) + throw new NullPointerException(); numberOfEntriesReadSoFar++; @@ -653,18 +665,16 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // as the volume of data could cause us to have to resize our buffers resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar, start); - final K key = e.getKey(); final Class keyClass = key.getClass(); - if (!kClass.isAssignableFrom(keyClass)) + if (!kClass.isAssignableFrom(keyClass)) { throw new ClassCastException("key=" + key + " is of type=" + keyClass + " " + "and should" + " be of type=" + kClass); + } + writeKeyInLoop(key, keyWriter, copies); - writeKey(key, local); - - final V value = e.getValue(); final Class valueClass = value.getClass(); if (!vClass.isAssignableFrom(valueClass)) @@ -673,7 +683,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon "should be of type=" + vClass); - writeValue(value, local); + writeValueInLoop(value, valueWriter, copies); final int len = (int) (bytes.position() - start); @@ -763,9 +773,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // get the data back from the server Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); final Collection result = new ArrayList(); + BytesReader valueReader = valueReaderWithSize.readerForLoop(null); for (; ; ) { final boolean hasMoreEntries = in.readBoolean(); @@ -774,7 +784,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long size = in.readInt(); for (int i = 0; i < size; i++) { - result.add(keyValueSerializer.readValue(in, local)); + result.add(valueReaderWithSize.readInLoop(in, valueReader)); } if (!hasMoreEntries) @@ -798,7 +808,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // get the data back from the server Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); + BytesReader keyReader = keyReaderWithSize.readerForLoop(copies); + copies = valueReaderWithSize.getCopies(copies); + BytesReader valueReader = valueReaderWithSize.readerForLoop(copies); final Map result = new HashMap(); for (; ; ) { @@ -809,8 +822,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon final long size = in.readInt(); for (int i = 0; i < size; i++) { - final K k = keyValueSerializer.readKey(in, local); - final V v = keyValueSerializer.readValue(in, local); + final K k = keyReaderWithSize.readInLoop(in, keyReader); + final V v = valueReaderWithSize.readInLoop(in, valueReader); result.put(k, v); } @@ -846,7 +859,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // get the data back from the server Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); + BytesReader keyReader = keyReaderWithSize.readerForLoop(null); final Set result = new HashSet<>(); for (; ; ) { @@ -857,7 +870,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long size = in.readInt(); for (int i = 0; i < size; i++) { - result.add(keyValueSerializer.readKey(in, local)); + result.add(keyReaderWithSize.readInLoop(in, keyReader)); } if (!hasMoreEntries) @@ -1144,18 +1157,16 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private void writeKey(K key) { - writeKey(key, null); + private ThreadLocalCopies writeKey(K key) { + return writeKey(key, null); } - - private void writeKey(K key, ThreadLocalCopies local) { + private ThreadLocalCopies writeKey(K key, ThreadLocalCopies copies) { long start = bytes.position(); for (; ; ) { try { - keyValueSerializer.writeKey(key, bytes, local); - return; + return keyWriterWithSize.write(bytes, key, copies); } catch (IndexOutOfBoundsException e) { resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); } catch (IllegalArgumentException e) { @@ -1169,43 +1180,64 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } - private V readValue(final long sizeLocation) { - return keyValueSerializer.readValue(blockingFetch(sizeLocation), null); - } - - private V readValue(final long sizeLocation, ThreadLocalCopies local) { - return keyValueSerializer.readValue(blockingFetch(sizeLocation), local); - } - - private void writeValue(V value) { - writeValue(value, null); - } - - private void writeValue(V value, ThreadLocalCopies local) { - - + private ThreadLocalCopies writeKeyInLoop(K key, Object writer, ThreadLocalCopies copies) { long start = bytes.position(); + for (; ; ) { try { - assert bytes.position() == start; - bytes.limit(bytes.capacity()); - keyValueSerializer.writeValue(value, bytes, local); - - return; + return keyWriterWithSize.writeInLoop(bytes, key, writer, copies); + } catch (IndexOutOfBoundsException e) { + resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); } catch (IllegalArgumentException e) { resizeToMessage(start, e); - - } catch (IllegalStateException e) { if (e.getMessage().contains("Not enough available space")) resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); else throw e; - } } + } + + private V readValue(final long sizeLocation, ThreadLocalCopies copies) { + return valueReaderWithSize.readNullable(blockingFetch(sizeLocation), copies); + } + + private ThreadLocalCopies writeValue(V value, ThreadLocalCopies copies) { + long start = bytes.position(); + for (; ; ) { + try { + assert bytes.position() == start; + bytes.limit(bytes.capacity()); + return valueWriterWithSize.write(bytes, value, copies); + } catch (IllegalArgumentException e) { + resizeToMessage(start, e); + } catch (IllegalStateException e) { + if (e.getMessage().contains("Not enough available space")) + resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); + else + throw e; + } + } + } + private ThreadLocalCopies writeValueInLoop(V value, Object writer, ThreadLocalCopies copies) { + long start = bytes.position(); + for (; ; ) { + try { + assert bytes.position() == start; + bytes.limit(bytes.capacity()); + return valueWriterWithSize.writeInLoop(bytes, value, writer, copies); + } catch (IllegalArgumentException e) { + resizeToMessage(start, e); + } catch (IllegalStateException e) { + if (e.getMessage().contains("Not enough available space")) + resizeBuffer((int) (bytes.capacity() + maxEntrySize), start); + else + throw e; + } + } } private void resizeToMessage(long start, IllegalArgumentException e) { diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 1baeb0d5a..bc56353bd 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -21,6 +21,7 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.RemoteNodeValidator; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.hash.replication.ThrottlingConfig; +import net.openhft.chronicle.hash.serialization.BytesReader; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; import net.openhft.lang.threadlocal.ThreadLocalCopies; @@ -1344,8 +1345,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable class StatelessServerConnector { public static final int TRANSACTION_ID_OFFSET = SIZE_OF_SIZE + 1; - public static final boolean MAP_SUPPORTS_BYTES = Boolean.valueOf(System.getProperty - ("mapSupportsBytes")); + public static final boolean MAP_SUPPORTS_BYTES = true; private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class .getName()); public static final StatelessChronicleMap.EventId[] VALUES @@ -1354,15 +1354,22 @@ class StatelessServerConnector { public static final int HEADER_SIZE = SIZE_OF_SIZE + SIZE_OF_IS_EXCEPTION + SIZE_OF_TRANSACTION_ID; - private final KeyValueSerializer keyValueSerializer; + private final ReaderWithSize keyReaderWithSize; + private final WriterWithSize keyWriterWithSize; + private final ReaderWithSize valueReaderWithSize; + private final WriterWithSize valueWriterWithSize; private final VanillaChronicleMap map; private double maxEntrySizeBytes; - StatelessServerConnector(@NotNull KeyValueSerializer keyValueSerializer, - @NotNull VanillaChronicleMap map, - int maxEntrySizeBytes) { - this.keyValueSerializer = keyValueSerializer; + StatelessServerConnector( + SerializationBuilder keySerializationBuilder, + SerializationBuilder valueSerializationBuilder, + @NotNull VanillaChronicleMap map, int maxEntrySizeBytes) { + keyReaderWithSize = new ReaderWithSize<>(keySerializationBuilder); + keyWriterWithSize = new WriterWithSize<>(keySerializationBuilder); + valueReaderWithSize = new ReaderWithSize<>(valueSerializationBuilder); + valueWriterWithSize = new WriterWithSize<>(valueSerializationBuilder); this.map = map; this.maxEntrySizeBytes = maxEntrySizeBytes; @@ -1465,15 +1472,11 @@ class StatelessServerConnector { public Work mapForKey(ByteBufferBytes reader, Bytes writer, long sizeLocation) { - - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K key = readKey(reader, local); - - + final K key = keyReaderWithSize.read(reader, null); final Function function = (Function) reader.readObject(); try { Object result = map.mapForKey(key, function); - keyValueSerializer.writeObject(result, writer); + writer.writeObject(result); } catch (Throwable e) { LOG.info("", e); @@ -1485,15 +1488,11 @@ class StatelessServerConnector { } public Work updateForKey(ByteBufferBytes reader, Bytes writer, long sizeLocation) { - - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K key = readKey(reader, local); - - + final K key = keyReaderWithSize.read(reader, null); final Mutator mutator = (Mutator) reader.readObject(); try { Object result = map.updateForKey(key, mutator); - keyValueSerializer.writeObject(result, writer); + writer.writeObject(result); } catch (Throwable e) { LOG.info("", e); @@ -1509,9 +1508,10 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { writer.writeBoolean(map.removeWithValue(reader)); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K key = readKey(reader, local); - final V readValue = readValue(reader, local); + ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); + copies = valueReaderWithSize.getCopies(copies); + final K key = keyReaderWithSize.read(reader, copies); + final V readValue = valueReaderWithSize.read(reader, copies); writer.writeBoolean(map.remove(key, readValue)); } } catch (Throwable e) { @@ -1526,10 +1526,11 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { map.replaceWithOldAndNew(reader, writer); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K key = readKey(reader, local); - final V oldValue = readValue(reader, local); - final V newValue = readValue(reader, local); + ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); + copies = valueReaderWithSize.getCopies(copies); + final K key = keyReaderWithSize.read(reader, copies); + final V oldValue = valueReaderWithSize.read(reader, copies); + final V newValue = valueReaderWithSize.read(reader, copies); writer.writeBoolean(map.replace(key, oldValue, newValue)); } } catch (Throwable e) { @@ -1618,8 +1619,7 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { writer.writeBoolean(map.containsKey(reader)); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K k = readKey(reader, local); + final K k = keyReaderWithSize.read(reader, null); writer.writeBoolean(map.containsKey(k)); } } catch (Throwable e) { @@ -1631,8 +1631,7 @@ class StatelessServerConnector { private Work containsValue(Bytes reader, Bytes writer, final long sizeLocation) { // todo optimize -- eliminate - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final V v = readValue(reader, local); + final V v = valueReaderWithSize.read(reader, null); try { writer.writeBoolean(map.containsValue(v)); @@ -1648,9 +1647,9 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { map.get(reader, writer); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K k = readKey(reader, local); - writeValue(map.get(k), writer, local); + final ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); + final K k = keyReaderWithSize.read(reader, copies); + valueWriterWithSize.writeNullable(writer, map.get(k), copies); } } catch (Throwable e) { return sendException(writer, sizeLocation, e); @@ -1665,9 +1664,10 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { map.put(reader); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K k = readKey(reader, local); - final V v = readValue(reader, local); + ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); + copies = valueReaderWithSize.getCopies(copies); + final K k = keyReaderWithSize.read(reader, copies); + final V v = valueReaderWithSize.read(reader, copies); map.put(k, v); } @@ -1679,10 +1679,11 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { map.put(reader, writer); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K k = readKey(reader, local); - final V v = readValue(reader, local); - writeValue(map.put(k, v), writer, local); + ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); + copies = valueReaderWithSize.getCopies(copies); + final K k = keyReaderWithSize.read(reader, copies); + final V v = valueReaderWithSize.read(reader, copies); + valueWriterWithSize.writeNullable(writer, map.put(k, v), copies); } } catch (Throwable e) { return sendException(writer, sizeLocation, e); @@ -1695,8 +1696,7 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { map.removeKeyAsBytes(reader); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K key = readKey(reader, local); + final K key = keyReaderWithSize.read(reader, null); map.remove(key); } @@ -1709,11 +1709,11 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { map.remove(reader, writer); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K key = readKey(reader, local); + ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); + final K key = keyReaderWithSize.read(reader, copies); if (LOG.isDebugEnabled()) LOG.debug("removing entry key=" + key); - writeValue(map.remove(key), writer, local); + valueWriterWithSize.writeNullable(writer, map.remove(key), copies); } } catch (Throwable e) { return sendException(writer, sizeLocation, e); @@ -1727,8 +1727,7 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { map.putAll(reader); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final Map m = readEntries(reader, local); + final Map m = readEntries(reader); map.putAll(m); } } catch (Throwable e) { @@ -1743,8 +1742,7 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { map.putAll(reader); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - map.putAll(readEntries(reader, local)); + map.putAll(readEntries(reader)); } return null; @@ -1777,32 +1775,33 @@ class StatelessServerConnector { } final Iterator iterator = values.iterator(); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); // this allows us to write more data than the buffer will allow return new Work() { @Override - public boolean doWork(Bytes writer) { + public boolean doWork(Bytes out) { + final long sizeLocation = header(out, transactionId); - final long sizeLocation = header(writer, transactionId); + ThreadLocalCopies copies = valueWriterWithSize.getCopies(null); + Object valueWriter = valueWriterWithSize.writerForLoop(copies); int count = 0; while (iterator.hasNext()) { // we've filled up the buffer, so lets give another channel a chance to send // some data, we don't know the max key size, we will use the entrySize instead - if (writer.remaining() <= maxEntrySizeBytes) { - writeHeader(writer, sizeLocation, count, true); + if (out.remaining() <= maxEntrySizeBytes) { + writeHeader(out, sizeLocation, count, true); return false; } count++; - writeValue(iterator.next(), writer, local); + valueWriterWithSize.writeInLoop(out, iterator.next(), valueWriter, copies); } - writeHeader(writer, sizeLocation, count, false); + writeHeader(out, sizeLocation, count, false); return true; } }; @@ -1819,33 +1818,34 @@ class StatelessServerConnector { return sendException(reader, writer, e); } - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); final Iterator iterator = ks.iterator(); // this allows us to write more data than the buffer will allow return new Work() { @Override - public boolean doWork(Bytes writer) { + public boolean doWork(Bytes out) { + final long sizeLocation = header(out, transactionId); - final long sizeLocation = header(writer, transactionId); + ThreadLocalCopies copies = keyWriterWithSize.getCopies(null); + Object keyWriter = keyWriterWithSize.writerForLoop(copies); int count = 0; while (iterator.hasNext()) { // we've filled up the buffer, so lets give another channel a chance to send // some data, we don't know the max key size, we will use the entrySize instead - if (writer.remaining() <= maxEntrySizeBytes) { - writeHeader(writer, sizeLocation, count, true); + if (out.remaining() <= maxEntrySizeBytes) { + writeHeader(out, sizeLocation, count, true); return false; } count++; K key = iterator.next(); - writeKey(key, writer, local); + keyWriterWithSize.writeInLoop(out, key, keyWriter, copies); } - writeHeader(writer, sizeLocation, count, false); + writeHeader(out, sizeLocation, count, false); return true; } }; @@ -1864,38 +1864,41 @@ class StatelessServerConnector { } final Iterator> iterator = entries.iterator(); - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); // this allows us to write more data than the buffer will allow return new Work() { @Override - public boolean doWork(Bytes writer) { - - if (writer.remaining() <= maxEntrySizeBytes) + public boolean doWork(Bytes out) { + if (out.remaining() <= maxEntrySizeBytes) return false; - final long sizeLocation = header(writer, transactionId); + final long sizeLocation = header(out, transactionId); + + ThreadLocalCopies copies = keyWriterWithSize.getCopies(null); + Object keyWriter = keyWriterWithSize.writerForLoop(copies); + copies = valueWriterWithSize.getCopies(copies); + Object valueWriter = valueWriterWithSize.writerForLoop(copies); int count = 0; while (iterator.hasNext()) { // we've filled up the buffer, so lets give another channel a chance to send // some data, we don't know the max key size, we will use the entrySize instead - if (writer.remaining() <= maxEntrySizeBytes) { + if (out.remaining() <= maxEntrySizeBytes) { - writeHeader(writer, sizeLocation, count, true); + writeHeader(out, sizeLocation, count, true); return false; } count++; final Map.Entry next = iterator.next(); - writeKey(next.getKey(), writer, local); - writeValue(next.getValue(), writer, local); + keyWriterWithSize.writeInLoop(out, next.getKey(), keyWriter, copies); + valueWriterWithSize.writeInLoop(out, next.getValue(), valueWriter, copies); } - writeHeader(writer, sizeLocation, count, false); + writeHeader(out, sizeLocation, count, false); return true; } @@ -1908,10 +1911,11 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { map.putIfAbsent(reader, writer); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K key = readKey(reader, local); - final V v = readValue(reader, local); - writeValue(map.putIfAbsent(key, v), writer, local); + ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); + copies = valueReaderWithSize.getCopies(copies); + final K key = keyReaderWithSize.read(reader, copies); + final V v = valueReaderWithSize.read(reader, copies); + valueWriterWithSize.writeNullable(writer, map.putIfAbsent(key, v), copies); } } catch (Throwable e) { return sendException(writer, sizeLocation, e); @@ -1926,10 +1930,11 @@ class StatelessServerConnector { if (MAP_SUPPORTS_BYTES) { map.replaceKV(reader, writer); } else { - final ThreadLocalCopies local = keyValueSerializer.threadLocalCopies(); - final K k = readKey(reader, local); - final V v = readValue(reader, local); - writeValue(map.replace(k, v), writer, local); + ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); + copies = valueReaderWithSize.getCopies(copies); + final K k = keyReaderWithSize.read(reader, copies); + final V v = valueReaderWithSize.read(reader, copies); + valueWriterWithSize.writeNullable(writer, map.replace(k, v), copies); } } catch (Throwable e) { return sendException(writer, sizeLocation, e); @@ -1938,16 +1943,6 @@ class StatelessServerConnector { return null; } - /** - * write the keysize and the key to the the {@code target} buffer - * - * @param key the key of the map - */ - - private void writeKey(K key, Bytes writer, final ThreadLocalCopies local) { - keyValueSerializer.writeKey(key, writer, local); - } - private long reflectTransactionId(Bytes reader, Bytes writer) { final long sizeLocation = writer.position(); writer.skip(TRANSACTION_ID_OFFSET); // isException + SIZE_OF_SIZE @@ -1956,38 +1951,29 @@ class StatelessServerConnector { return sizeLocation; } - private void writeValue(V value, final Bytes writer, ThreadLocalCopies local) { - keyValueSerializer.writeValue(value, writer, local); - } - - private K readKey(Bytes reader, ThreadLocalCopies local) { - return keyValueSerializer.readKey(reader, local); - } - - private V readValue(Bytes reader, ThreadLocalCopies local) { - return keyValueSerializer.readValue(reader, local); - } - private void writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) { long currentPosition = out.position(); final long size = out.position() - locationOfSize; out.writeInt(locationOfSize, (int) size); // size out.writeBoolean(locationOfSize + SIZE_OF_SIZE, isException); // isException - out.position(currentPosition); } private void writeException(Bytes out, Throwable e) { out.writeObject(e); } - private Map readEntries(Bytes reader, final ThreadLocalCopies local) { - + private Map readEntries(Bytes reader) { final long numberOfEntries = reader.readStopBit(); final HashMap result = new HashMap(); + ThreadLocalCopies copies = keyReaderWithSize.getCopies(null); + BytesReader keyReader = keyReaderWithSize.readerForLoop(copies); + copies = valueReaderWithSize.getCopies(copies); + BytesReader valueReader = valueReaderWithSize.readerForLoop(copies); + for (long i = 0; i < numberOfEntries; i++) { - K key = readKey(reader, local); - V value = readValue(reader, local); + K key = keyReaderWithSize.readInLoop(reader, keyReader); + V value = valueReaderWithSize.readInLoop(reader, valueReader); result.put(key, value); } return result; @@ -2040,106 +2026,4 @@ class StatelessServerConnector { } -class KeyValueSerializer { - - final Serializer valueSerializer; - final Serializer keySerializer; - - KeyValueSerializer(SerializationBuilder keySerializer, - SerializationBuilder valueSerializer) { - this.keySerializer = new Serializer(keySerializer); - this.valueSerializer = new Serializer(valueSerializer); - } - - V readValue(Bytes reader) { - if (nullCheck(reader)) return null; - return valueSerializer.readMarshallable(reader, null); - } - - - V readValue(Bytes reader, ThreadLocalCopies local) { - if (nullCheck(reader)) return null; - 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(); - } - - K readKey(Bytes reader) { - if (nullCheck(reader)) return null; - - return keySerializer.readMarshallable(reader, null); - } - - 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) { - - writer.writeBoolean(key == null); - if (key != null) - keySerializer.writeMarshallable(key, writer, null); - - } - - void writeValue(V value, Bytes writer) { - assert writer.limit() == writer.capacity(); - writer.writeBoolean(value == null); - - if (value != null) { - valueSerializer.writeMarshallable(value, writer, null); - } - - } - - - void writeKey(K key, Bytes writer, ThreadLocalCopies threadLocal) { - - writer.writeBoolean(key == null); - if (key != null) - keySerializer.writeMarshallable(key, writer, threadLocal); - - } - - void writeValue(V value, Bytes writer, ThreadLocalCopies threadLocal) { - - assert writer.limit() == writer.capacity(); - writer.writeBoolean(value == null); - - if (value != null) - valueSerializer.writeMarshallable(value, writer, threadLocal); - - } - - - 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/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index 328b744c3..d5174b4a7 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -118,12 +118,9 @@ class VanillaChronicleMap, transient long headerSize; private transient Set> entrySet; - private transient SerializationBuilder keyBuilder; - private transient SerializationBuilder valueBuilder; - public VanillaChronicleMap(AbstractChronicleMapBuilder builder) throws IOException { - keyBuilder = builder.keyBuilder; + SerializationBuilder keyBuilder = builder.keyBuilder; kClass = keyBuilder.eClass; keySizeMarshaller = keyBuilder.sizeMarshaller(); originalKeyReader = keyBuilder.reader(); @@ -133,7 +130,7 @@ class VanillaChronicleMap, originalMetaKeyInterop = (MKI) keyBuilder.metaInterop(); metaKeyInteropProvider = (MetaProvider) keyBuilder.metaInteropProvider(); - valueBuilder = builder.valueBuilder; + SerializationBuilder valueBuilder = builder.valueBuilder; vClass = valueBuilder.eClass; isNativeValueClass = vClass.getName().endsWith("$$Native"); valueSizeMarshaller = valueBuilder.sizeMarshaller(); @@ -348,6 +345,7 @@ class VanillaChronicleMap, ReadValueToOutputBytes readValueToOutputBytes = segmentState.readValueToOutputBytes; readValueToOutputBytes.reuse(valueSizeMarshaller, output); long keySize = keySizeMarshaller.readSize(entry); + entry.limit(entry.position() + keySize); GetRemoteBytesValueInterops getRemoteBytesValueInterops = segmentState.getRemoteBytesValueInterops; MultiStoreBytes value = getRemoteBytesValueInterops.getValueBytes( @@ -371,6 +369,7 @@ class VanillaChronicleMap, ThreadLocalCopies copies = SegmentState.getCopies(null); SegmentState segmentState = SegmentState.get(copies); long keySize = keySizeMarshaller.readSize(entry); + entry.limit(entry.position() + keySize); GetRemoteBytesValueInterops getRemoteBytesValueInterops = segmentState.getRemoteBytesValueInterops; MultiStoreBytes value = getRemoteBytesValueInterops.getValueBytes( @@ -398,7 +397,6 @@ class VanillaChronicleMap, public Bytes readValue(@NotNull ThreadLocalCopies copies, Bytes entry, Bytes usingBytes, long valueSize) { usingBytes = bytes(valueSize); - usingBytes.writeBoolean(false); valueSizeMarshaller.writeSize(usingBytes, valueSize); usingBytes.write(entry, entry.position(), valueSize); return usingBytes; @@ -418,6 +416,13 @@ class VanillaChronicleMap, return outputBytes; } + @Override + public Bytes readValue(@NotNull ThreadLocalCopies copies, Bytes entry, Bytes usingBytes, + long valueSize) { + outputBytes.writeBoolean(false); + return super.readValue(copies, entry, usingBytes, valueSize); + } + @Override public Bytes readNull() { outputBytes.writeBoolean(true); @@ -518,7 +523,8 @@ class VanillaChronicleMap, lookupUsing(copies, segmentState, DelegatingMetaBytesInterop.>instance(), BytesBytesInterop.INSTANCE, key, keySizeMarshaller.readSize(key), - keyBytesToInstance, readValueToOutputBytes, null, outputValueBytesToInstance, false); + keyBytesToInstance, readValueToOutputBytes, null, outputValueBytesToInstance, + false); } private V lookupUsing(K key, V usingValue, boolean create) { @@ -667,8 +673,7 @@ class VanillaChronicleMap, final XStream xstream = new XStream(new JettisonMappedXmlDriver()); xstream.setMode(XStream.NO_REFERENCES); - xstream.alias("chronicle-entries", net.openhft.chronicle.map.VanillaChronicleMap.EntrySet.class); - + xstream.alias("chronicle-entries", VanillaChronicleMap.EntrySet.class); // ideally we will use java serialization if we can lazyXStreamConverter().registerConverter(xstream); @@ -682,8 +687,7 @@ class VanillaChronicleMap, public synchronized void putAll(File fromFile) throws IOException { final XStream xstream = new XStream(new JettisonMappedXmlDriver()); xstream.setMode(XStream.NO_REFERENCES); - xstream.alias("chronicle-entries", net.openhft.chronicle.map.VanillaChronicleMap.EntrySet - .class); + xstream.alias("chronicle-entries", VanillaChronicleMap.EntrySet.class); // ideally we will use java serialization if we can lazyXStreamConverter().registerConverter(xstream); @@ -700,35 +704,18 @@ class VanillaChronicleMap, /** * @return a lazily created XStreamConverter */ - private synchronized XStreamConverter lazyXStreamConverter() { - if (xStreamConverter != null) - return xStreamConverter; - - final KeyValueSerializer serializers = new KeyValueSerializer(keyBuilder, - valueBuilder); - - final ThreadLocalCopies threadLocalCopies = serializers.threadLocalCopies(); - - final ByteBufferBytes buffer = new ByteBufferBytes(allocateDirect((int) entrySize) - .order(ByteOrder.nativeOrder())); - - xStreamConverter = new XStreamConverter(threadLocalCopies, buffer, serializers); - return xStreamConverter; + private XStreamConverter lazyXStreamConverter() { + return xStreamConverter != null ? xStreamConverter : + (xStreamConverter = new XStreamConverter()); } private class XStreamConverter { - - private final ThreadLocalCopies threadLocalCopies; private Bytes buffer; - private final KeyValueSerializer keyValueSerializer; - public XStreamConverter(ThreadLocalCopies threadLocalCopies, - ByteBufferBytes buffer, - final KeyValueSerializer keyValueSerializer) { - this.threadLocalCopies = threadLocalCopies; - this.buffer = buffer; - this.keyValueSerializer = keyValueSerializer; + public XStreamConverter() { + this.buffer = new ByteBufferBytes(allocateDirect((int) entrySize) + .order(ByteOrder.nativeOrder())); } private void registerConverter(XStream xstream) { @@ -827,9 +814,16 @@ class VanillaChronicleMap, } break; case "chronicle-key": - return keyValueSerializer.readKey(buffer, threadLocalCopies); + long keySize = keySizeMarshaller.readSize(buffer); + ThreadLocalCopies copies = keyReaderProvider.getCopies(null); + BytesReader keyReader = keyReaderProvider.get(copies, originalKeyReader); + return keyReader.read(buffer, keySize); case "chronicle-value": - return keyValueSerializer.readValue(buffer, threadLocalCopies); + long valueSize = valueSizeMarshaller.readSize(buffer); + copies = valueReaderProvider.getCopies(null); + BytesReader valueReader = + valueReaderProvider.get(copies, originalValueReader); + return valueReader.read(buffer, valueSize); } return null; @@ -951,7 +945,7 @@ class VanillaChronicleMap, } }; - private transient InstanceOrBytesToInstance outputValueBytesToInstance = + transient InstanceOrBytesToInstance outputValueBytesToInstance = new InstanceOrBytesToInstance() { @Override public V toInstance(@NotNull ThreadLocalCopies copies, Bytes outputBytes, long size) { @@ -1127,13 +1121,33 @@ class VanillaChronicleMap, metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyInterop, key); long keySize = metaKeyInterop.size(keyInterop, key); return remove(copies, null, metaKeyInterop, keyInterop, key, keySize, - keyIdentity(), this, expectedValue, valueIdentity(), removeReturnsNull); + keyIdentity(), this, expectedValue, valueIdentity(), this, removeReturnsNull); } final void removeKeyAsBytes(Bytes key) { + ThreadLocalCopies copies = SegmentState.getCopies(null); + SegmentState segmentState = SegmentState.get(copies); + ReadValueToBytes readValueToLazyBytes = segmentState.readValueToLazyBytes; + readValueToLazyBytes.valueSizeMarshaller(valueSizeMarshaller); + long keySize = keySizeMarshaller.readSize(key); + key.limit(key.position() + keySize); remove(null, null, DelegatingMetaBytesInterop.>instance(), - BytesBytesInterop.INSTANCE, key, keySizeMarshaller.readSize(key), - keyBytesToInstance, null, null, null, true); + BytesBytesInterop.INSTANCE, key, keySize, + keyBytesToInstance, null, null, + outputValueBytesToInstance, readValueToLazyBytes, true); + } + + final void remove(Bytes key, Bytes output) { + ThreadLocalCopies copies = SegmentState.getCopies(null); + SegmentState segmentState = SegmentState.get(copies); + ReadValueToOutputBytes readValueToOutputBytes = segmentState.readValueToOutputBytes; + readValueToOutputBytes.reuse(valueSizeMarshaller, output); + long keySize = keySizeMarshaller.readSize(key); + key.limit(key.position() + keySize); + remove(copies, segmentState, + DelegatingMetaBytesInterop.>instance(), + BytesBytesInterop.INSTANCE, key, keySize, keyBytesToInstance, null, null, + outputValueBytesToInstance, readValueToOutputBytes, false); } // for boolean remove(Object key, Object value); @@ -1141,40 +1155,46 @@ class VanillaChronicleMap, ThreadLocalCopies copies = SegmentState.getCopies(null); SegmentState segmentState = SegmentState.get(copies); long keySize = keySizeMarshaller.readSize(entry); + entry.limit(entry.position() + keySize); GetRemoteBytesValueInterops getRemoveBytesValueInterops = segmentState.getRemoteBytesValueInterops; MultiStoreBytes value = getRemoveBytesValueInterops.getValueBytes( entry, entry.position() + keySize); getRemoveBytesValueInterops.valueSizeMarshaller(valueSizeMarshaller); + ReadValueToBytes readValueToLazyBytes = segmentState.readValueToLazyBytes; + readValueToLazyBytes.valueSizeMarshaller(valueSizeMarshaller); return (Boolean) remove(copies, segmentState, DelegatingMetaBytesInterop.>instance(), BytesBytesInterop.INSTANCE, entry, keySize, keyBytesToInstance, - getRemoveBytesValueInterops, value, valueBytesToInstance, false); + getRemoveBytesValueInterops, value, valueBytesToInstance, + readValueToLazyBytes, false); } private , - VB, VBI, MVBI extends MetaBytesInterop> + RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> Object remove(ThreadLocalCopies copies, SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, GetValueInterops getValueInterops, VB expectedValue, - InstanceOrBytesToInstance toValue, - boolean resultUnused) { + InstanceOrBytesToInstance toValue, + ReadValue readValue, boolean resultUnused) { long hash = metaKeyInterop.hash(keyInterop, key); int segmentNum = getSegment(hash); long segmentHash = segmentHash(hash); Segment segment = segments[segmentNum]; return segment.remove(copies, segmentState, metaKeyInterop, keyInterop, key, keySize, toKey, getValueInterops, expectedValue, - toValue, segmentHash, resultUnused); + toValue, segmentHash, readValue, resultUnused); } /** * @throws NullPointerException if any of the arguments are null */ @Override - public final boolean replace(@net.openhft.lang.model.constraints.NotNull K key, @net.openhft.lang.model.constraints.NotNull V oldValue, @net.openhft.lang.model.constraints.NotNull V newValue) { + public final boolean replace(@net.openhft.lang.model.constraints.NotNull K key, + @net.openhft.lang.model.constraints.NotNull V oldValue, + @net.openhft.lang.model.constraints.NotNull V newValue) { checkValue(oldValue); return (Boolean) replaceIfValueIs(key, oldValue, newValue); } @@ -1184,7 +1204,8 @@ class VanillaChronicleMap, */ @Override @SuppressWarnings("unchecked") - public final V replace(@net.openhft.lang.model.constraints.NotNull final K key, @net.openhft.lang.model.constraints.NotNull final V value) { + public final V replace(@net.openhft.lang.model.constraints.NotNull final K key, + @net.openhft.lang.model.constraints.NotNull final V value) { return (V) replaceIfValueIs(key, null, value); } @@ -1236,6 +1257,7 @@ class VanillaChronicleMap, readValueToOutputBytes.reuse(valueSizeMarshaller, output); long keySize = keySizeMarshaller.readSize(keyAndNewValue); + keyAndNewValue.limit(keyAndNewValue.position() + keySize); GetRemoteBytesValueInterops getRemoteBytesValueInterops = segmentState.getRemoteBytesValueInterops; MultiStoreBytes value = getRemoteBytesValueInterops.getValueBytes( @@ -1269,6 +1291,7 @@ class VanillaChronicleMap, getNewValueInterops.valueSizeMarshaller(valueSizeMarshaller); entryAndNewValue.position(keyPosition); + entryAndNewValue.limit(keyPosition + keySize); output.writeBoolean((Boolean) replace(copies, segmentState, DelegatingMetaBytesInterop.>instance(), @@ -1337,7 +1360,9 @@ class VanillaChronicleMap, long nextEntryPosition = entries.position() + valueSize; entries.position(entryPosition); put(entries); + entries.clear(); // because used as key, altering position and limit entryPosition = nextEntryPosition; + entries.position(entryPosition); } } @@ -2045,19 +2070,19 @@ class VanillaChronicleMap, } , - VB, VBI, MVBI extends MetaBytesInterop> + RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> Object remove(@Nullable ThreadLocalCopies copies, @Nullable SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, GetValueInterops getValueInterops, VB expectedValue, - InstanceOrBytesToInstance toValue, - long hash2, boolean resultUnused) { + InstanceOrBytesToInstance toValue, + long hash2, ReadValue readValue, boolean resultUnused) { writeLock(null); try { return removeWithoutLock(copies, segmentState, metaKeyInterop, keyInterop, key, keySize, toKey, getValueInterops, expectedValue, toValue, - hash2, resultUnused); + hash2, readValue, resultUnused); } finally { if (segmentState != null) segmentState.close(); @@ -2072,14 +2097,14 @@ class VanillaChronicleMap, * returned - if resultUnused is true, null is always returned */ , - VB, VBI, MVBI extends MetaBytesInterop> + RV, VB extends RV, VBI, MVBI extends MetaBytesInterop> Object removeWithoutLock( @Nullable ThreadLocalCopies copies, final @Nullable SegmentState segmentState, MKBI metaKeyInterop, KBI keyInterop, KB key, long keySize, InstanceOrBytesToInstance toKey, GetValueInterops getValueInterops, VB expectedValue, - InstanceOrBytesToInstance toValue, - long hash2, boolean resultUnused) { + InstanceOrBytesToInstance toValue, + long hash2, ReadValue readValue, boolean resultUnused) { segmentStateNotNullImpliesCopiesNotNull(copies, segmentState); SegmentState localSegmentState = segmentState; try { @@ -2113,35 +2138,36 @@ class VanillaChronicleMap, if (!metaValueInterop.startsWith(valueInterop, entry, expectedValue)) return Boolean.FALSE; } - return removeEntry(copies, key, keySize, toKey, expectedValue, toValue, - resultUnused, hashLookup, entry, pos, valueSizePos, valueSize, - false, true, expectedValue != null); + return removeEntry(copies, key, keySize, toKey, toValue, + readValue, resultUnused, hashLookup, entry, pos, valueSizePos, + valueSize, false, true, expectedValue != null); } // key is not found - return expectedValue == null ? null : Boolean.FALSE; + if (expectedValue == null) { + return resultUnused ? null : readValue.readNull(); + } else { + return Boolean.FALSE; + } } finally { if (segmentState == null && localSegmentState != null) localSegmentState.close(); } } - final Object removeEntry( + final Object removeEntry( ThreadLocalCopies copies, KB key, long keySize, InstanceOrBytesToInstance toKey, - VB expectedValue, InstanceOrBytesToInstance toValue, - boolean resultUnused, MultiMap hashLookup, MultiStoreBytes entry, long pos, + InstanceOrBytesToInstance toValue, + ReadValue readValue, boolean resultUnused, + MultiMap hashLookup, MultiStoreBytes entry, long pos, long valueSizePos, long valueSize, boolean remote, boolean removeFromMultiMap, boolean booleanResult) { // get the removed value, if needed - V removedValue = null; + RV removedValue = null; if ((!booleanResult && !resultUnused) || eventListener != null) { - if (expectedValue != null) - removedValue = toValue.toInstance(copies, expectedValue, valueSize); - if (removedValue == null) { - // todo reuse some value - removedValue = readValue(copies, entry, null, valueSize); - entry.position(entry.position() - valueSize); - } + // todo reuse some value + removedValue = readValue.readValue(copies, entry, null, valueSize); + entry.position(entry.position() - valueSize); } // update segment state @@ -2160,8 +2186,8 @@ class VanillaChronicleMap, if (bytesEventListener != null) bytesEventListener.onRemove(entry, 0L, metaDataBytes, valueSizePos); if (eventListener != null) { - V removedValueForEventListener = removedValue != null ? removedValue : - readValue(copies, entry, null); + V removedValueForEventListener = + toValue.toInstance(copies, removedValue, valueSize); eventListener.onRemove(toKey.toInstance(copies, key, keySize), removedValueForEventListener); } @@ -2757,7 +2783,7 @@ class VanillaChronicleMap, public void removeEntry() { segment.removeWithoutLock(copies, segmentState, metaKeyInterop, keyInterop, key(), keySize, map.keyIdentity(), - map, null, map.valueIdentity(), segmentHash, true); + map, null, map.valueIdentity(), segmentHash, map, true); } } @@ -2793,7 +2819,7 @@ class VanillaChronicleMap, putOnClose = false; segment.removeWithoutLock(copies, segmentState, metaKeyInterop, keyInterop, key(), keySize, map.keyIdentity(), - map, null, map.valueIdentity(), segmentHash, true); + map, null, map.valueIdentity(), segmentHash, map, true); } } } diff --git a/src/main/java/net/openhft/chronicle/map/WriterWithSize.java b/src/main/java/net/openhft/chronicle/map/WriterWithSize.java new file mode 100644 index 000000000..62f01cbc3 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/WriterWithSize.java @@ -0,0 +1,89 @@ +/* + * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com + * + * 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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.chronicle.hash.serialization.SizeMarshaller; +import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop; +import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter; +import net.openhft.chronicle.hash.serialization.internal.MetaProvider; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.threadlocal.Provider; +import net.openhft.lang.threadlocal.StatefulCopyable; +import net.openhft.lang.threadlocal.ThreadLocalCopies; +import org.jetbrains.annotations.Nullable; + +final class WriterWithSize { + private final SizeMarshaller sizeMarshaller; + private final Object originalWriter; + private final Provider writerProvider; + private final MetaBytesInterop originalMetaWriter; + private final MetaProvider metaWriterProvider; + + WriterWithSize(SerializationBuilder serializationBuilder) { + sizeMarshaller = serializationBuilder.sizeMarshaller(); + originalWriter = serializationBuilder.interop(); + writerProvider = Provider.of((Class) originalWriter.getClass()); + originalMetaWriter = serializationBuilder.metaInterop(); + metaWriterProvider = serializationBuilder.metaInteropProvider(); + } + + + + ThreadLocalCopies getCopies(ThreadLocalCopies copies) { + copies = writerProvider.getCopies(copies); + return metaWriterProvider.getCopies(copies); + } + + public ThreadLocalCopies write(Bytes out, T t, @Nullable ThreadLocalCopies copies) { + copies = writerProvider.getCopies(copies); + Object writer = writerProvider.get(copies, originalWriter); + copies = metaWriterProvider.getCopies(copies); + MetaBytesWriter metaWriter = metaWriterProvider.get(copies, originalMetaWriter, writer, t); + sizeMarshaller.writeSize(out, metaWriter.size(writer, t)); + metaWriter.write(writer, out, t); + return copies; + } + + public ThreadLocalCopies writeNullable(Bytes out, T t, @Nullable ThreadLocalCopies copies) { + out.writeBoolean(t == null); + if (t == null) + return copies; + return write(out, t, copies); + } + + public Object writerForLoop(@Nullable ThreadLocalCopies copies) { + copies = writerProvider.getCopies(copies); + return writerProvider.get(copies, originalWriter); + } + + public ThreadLocalCopies writeInLoop(Bytes out, T t, Object writer, + @Nullable ThreadLocalCopies copies) { + copies = metaWriterProvider.getCopies(copies); + MetaBytesWriter metaWriter = metaWriterProvider.get(copies, originalMetaWriter, writer, t); + sizeMarshaller.writeSize(out, metaWriter.size(writer, t)); + metaWriter.write(writer, out, t); + return copies; + } + + public ThreadLocalCopies writeNullableInLoop(Bytes out, T t, Object writer, + @Nullable ThreadLocalCopies copies) { + out.writeBoolean(t == null); + if (t == null) + return copies; + return writeInLoop(out, t, writer, copies); + } +} diff --git a/src/test/java/net/openhft/chronicle/map/SerializerTest.java b/src/test/java/net/openhft/chronicle/map/SerializerTest.java index 7551c4037..ea21e7f9a 100644 --- a/src/test/java/net/openhft/chronicle/map/SerializerTest.java +++ b/src/test/java/net/openhft/chronicle/map/SerializerTest.java @@ -73,23 +73,23 @@ public class SerializerTest { ByteBufferBytes in = out.slice(); - ChronicleMapBuilder cBuilder = ChronicleMapBuilder.of(Integer - .class, valueClass); + ChronicleMapBuilder cBuilder = ChronicleMapBuilder.of(Integer.class, valueClass); OnHeapUpdatableChronicleMapBuilder builder = (OnHeapUpdatableChronicleMapBuilder) cBuilder.delegate; builder.preMapConstruction(false); - Serializer v = new Serializer(builder.valueBuilder); + WriterWithSize valueWriterWithSize = new WriterWithSize(builder.valueBuilder); + ReaderWithSize valueReaderWithSize = new ReaderWithSize(builder.valueBuilder); - v.writeMarshallable(value, out, null); + valueWriterWithSize.write(out, value, null); long position = out.position(); in.limit(position); - Object actual = v.readMarshallable(in, null); + Object actual = valueReaderWithSize.read(in, null); Assert.assertEquals(actual, value); } @@ -107,15 +107,15 @@ public class SerializerTest { builder.preMapConstruction(false); { - Serializer v = new Serializer(builder.keyBuilder); + WriterWithSize keyWriterWithSize = new WriterWithSize(builder.keyBuilder); + ReaderWithSize keyReaderWithSize = new ReaderWithSize(builder.keyBuilder); - - v.writeMarshallable(key, out, null); + keyWriterWithSize.write(out, key, null); long position = out.position(); in.limit(position); - Object actual = v.readMarshallable(in, null); + Object actual = keyReaderWithSize.read(in, null); Assert.assertEquals(actual, key); } diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 78a3707f8..5fb2ea699 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -27,6 +27,8 @@ import java.net.InetSocketAddress; import java.util.*; import java.util.concurrent.ExecutionException; +import static org.junit.Assert.assertEquals; + /** * @author Rob Austin. @@ -65,7 +67,7 @@ public class StatelessClientTest { String actual = statelessMap.mapForKey(10, ToString.INSTANCE); - Assert.assertEquals("Hello World", actual); + assertEquals("Hello World", actual); } } } @@ -88,7 +90,7 @@ public class StatelessClientTest { String actual = statelessMap.mapForKey(11, ToString.INSTANCE); - Assert.assertEquals(null, actual); + assertEquals(null, actual); } } } @@ -118,9 +120,9 @@ public class StatelessClientTest { Set> entries = statelessMap.entrySet(); Map.Entry next = entries.iterator().next(); - Assert.assertEquals("some value=" + next.getKey(), next.getValue()); + assertEquals("some value=" + next.getKey(), next.getValue()); - Assert.assertEquals(entries.size(), SIZE); + assertEquals(SIZE, entries.size()); } } @@ -151,7 +153,7 @@ public class StatelessClientTest { Collection values = statelessMap.values(); - Assert.assertEquals(values.size(), SIZE); + assertEquals(values.size(), SIZE); } } @@ -182,7 +184,7 @@ public class StatelessClientTest { final Set keys = statelessMap.keySet(); - Assert.assertEquals(keys.size(), payload.size()); + assertEquals(keys.size(), payload.size()); } @@ -203,8 +205,8 @@ public class StatelessClientTest { .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { - Assert.assertEquals("EXAMPLE-10", statelessMap.get(10)); - Assert.assertEquals(1, statelessMap.size()); + assertEquals("EXAMPLE-10", statelessMap.get(10)); + assertEquals(1, statelessMap.size()); } } } @@ -223,8 +225,8 @@ public class StatelessClientTest { serverMap.put(10, "EXAMPLE-10"); - Assert.assertEquals("EXAMPLE-10", statelessMap.get(10)); - Assert.assertEquals(1, statelessMap.size()); + assertEquals("EXAMPLE-10", statelessMap.get(10)); + assertEquals(1, statelessMap.size()); } @@ -249,8 +251,8 @@ public class StatelessClientTest { serverMap.put("hello", Collections.singletonMap("hello", "world")); - Assert.assertEquals(Collections.singletonMap("hello", "world"), statelessMap.get("hello")); - Assert.assertEquals(1, statelessMap.size()); + assertEquals(Collections.singletonMap("hello", "world"), statelessMap.get("hello")); + assertEquals(1, statelessMap.size()); } @@ -281,8 +283,8 @@ public class StatelessClientTest { statelessMap.put("hello", data); - Assert.assertEquals(data, serverMap.get("hello")); - Assert.assertEquals(1, statelessMap.size()); + assertEquals(data, serverMap.get("hello")); + assertEquals(1, statelessMap.size()); } @@ -312,8 +314,8 @@ public class StatelessClientTest { int value = SIZE - 10; - Assert.assertEquals("some value=" + value, statelessMap.get(value)); - Assert.assertEquals(SIZE, statelessMap.size()); + assertEquals("some value=" + value, statelessMap.get(value)); + assertEquals(SIZE, statelessMap.size()); } } @@ -343,8 +345,8 @@ public class StatelessClientTest { int value = SIZE - 10; - Assert.assertEquals("some value=" + value, statelessMap.get(value)); - Assert.assertEquals(SIZE, statelessMap.size()); + assertEquals("some value=" + value, statelessMap.get(value)); + assertEquals(SIZE, statelessMap.size()); } } @@ -365,8 +367,8 @@ public class StatelessClientTest { statelessMap.put(1, "some value"); - Assert.assertEquals("some value", statelessMap.get(1)); - Assert.assertEquals(1, statelessMap.size()); + assertEquals("some value", statelessMap.get(1)); + assertEquals(1, statelessMap.size()); } @@ -388,13 +390,13 @@ public class StatelessClientTest { statelessMap.put(1, "some value"); - Assert.assertEquals("some value", statelessMap.get(1)); - Assert.assertEquals(1, statelessMap.size()); + assertEquals("some value", statelessMap.get(1)); + assertEquals(1, statelessMap.size()); statelessMap.remove(1); - Assert.assertEquals(null, statelessMap.get(1)); - Assert.assertEquals(0, statelessMap.size()); + assertEquals(null, statelessMap.get(1)); + assertEquals(0, statelessMap.size()); } } @@ -415,13 +417,13 @@ public class StatelessClientTest { statelessMap.put(1, "some value"); - Assert.assertEquals("some value", statelessMap.getLater(1).get()); - Assert.assertEquals(1, statelessMap.size()); + assertEquals("some value", statelessMap.getLater(1).get()); + assertEquals(1, statelessMap.size()); statelessMap.remove(1); - Assert.assertEquals(null, statelessMap.getLater(1).get()); - Assert.assertEquals(0, statelessMap.size()); + assertEquals(null, statelessMap.getLater(1).get()); + assertEquals(0, statelessMap.size()); } } @@ -441,8 +443,8 @@ public class StatelessClientTest { .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { CharSequence oldValue = statelessMap.putLater(1, "some value").get(); - Assert.assertEquals("some value", statelessMap.get(1)); - Assert.assertEquals(1, statelessMap.size()); + assertEquals("some value", statelessMap.get(1)); + assertEquals(1, statelessMap.size()); } } @@ -462,8 +464,8 @@ public class StatelessClientTest { .statelessClient(new InetSocketAddress("localhost", 8056)).create()) { statelessMap.put(1, "some value"); statelessMap.removeLater(1).get(); - Assert.assertEquals(null, statelessMap.get(1)); - Assert.assertEquals(0, statelessMap.size()); + assertEquals(null, statelessMap.get(1)); + assertEquals(0, statelessMap.size()); } }