From e48a44e6e9cc62c5fe643144b3fdea4de733881d Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 15 Oct 2014 16:23:57 +0100 Subject: [PATCH] HCOLL-167 Created a stateless map - fixed test case -net.openhft.chronicle.map.SerializerTest.testKeyMarshallable --- .../net/openhft/chronicle/map/Serializer.java | 65 ++--- .../chronicle/map/StatelessMapClient.java | 273 +++++------------- .../openhft/chronicle/map/SerializerTest.java | 20 +- 3 files changed, 94 insertions(+), 264 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/Serializer.java b/src/main/java/net/openhft/chronicle/map/Serializer.java index ec112db76..52643c9f2 100644 --- a/src/main/java/net/openhft/chronicle/map/Serializer.java +++ b/src/main/java/net/openhft/chronicle/map/Serializer.java @@ -7,19 +7,16 @@ import net.openhft.chronicle.map.serialization.SizeMarshaller; import net.openhft.chronicle.map.threadlocal.Provider; import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import net.openhft.lang.io.Bytes; -import net.openhft.lang.io.serialization.BytesMarshallable; -import net.openhft.lang.io.serialization.ObjectFactory; import net.openhft.lang.model.Byteable; import net.openhft.lang.model.constraints.NotNull; /** * uses a {@code SerializationBuilder} to readMarshallable and writeMarshallable objects + * * @author Rob Austin. */ -public class Serializer> - implements BytesMarshallable { +public class Serializer> { - final Class clazz; final SizeMarshaller sizeMarshaller; final BytesReader originalReader; transient Provider> readerProvider; @@ -27,44 +24,38 @@ public class Serializer> transient Provider writerProvider; final MVW originalMetaValueWriter; final MetaProvider metaValueWriterProvider; - final ObjectFactory objectFactory; - - final ThreadLocal value = new ThreadLocal(); public Serializer(final SerializationBuilder serializationBuilder) { - clazz = serializationBuilder.eClass; + sizeMarshaller = serializationBuilder.sizeMarshaller(); originalReader = serializationBuilder.reader(); originalWriter = (VW) serializationBuilder.interop(); originalMetaValueWriter = (MVW) serializationBuilder.metaInterop(); metaValueWriterProvider = (MetaProvider) serializationBuilder.metaInteropProvider(); - objectFactory = serializationBuilder.factory(); readerProvider = Provider.of((Class) originalReader.getClass()); writerProvider = Provider.of((Class) originalWriter.getClass()); } + public O readMarshallable(@NotNull Bytes in) throws IllegalStateException { - public O getObject() { - return value.get(); + ThreadLocalCopies copies; + copies = readerProvider.getCopies(null); + 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 void setObject(O value) { - this.value.set(value); - } + public void writeMarshallable(@NotNull Bytes out, O value) { - @Override - public void readMarshallable(@NotNull Bytes in) throws IllegalStateException { - O o = readObject(in); - setObject(o); - } - - @Override - public void writeMarshallable(@NotNull Bytes out) { - O value1 = getObject(); - boolean byteableValue = value1 instanceof Byteable; + boolean byteableValue = value instanceof Byteable; ThreadLocalCopies copies; copies = readerProvider.getCopies(null); VW valueWriter = writerProvider.get(copies, originalWriter); @@ -78,10 +69,10 @@ public class Serializer> valueWriter = writerProvider.get(copies, originalWriter); copies = metaValueWriterProvider.getCopies(copies); metaValueWriter = metaValueWriterProvider.get( - copies, originalMetaValueWriter, valueWriter, value1); - valueSize = metaValueWriter.size(valueWriter, value1); + copies, originalMetaValueWriter, valueWriter, value); + valueSize = metaValueWriter.size(valueWriter, value); } else { - valueAsByteable = (Byteable) value1; + valueAsByteable = (Byteable) value; valueSize = valueAsByteable.maxSize(); } @@ -89,26 +80,10 @@ public class Serializer> if (metaValueWriter != null) { - metaValueWriter.write(valueWriter, out, value1); + metaValueWriter.write(valueWriter, out, value); } else { throw new UnsupportedOperationException(""); } } - private O readObject(Bytes in) { - - ThreadLocalCopies copies; - copies = readerProvider.getCopies(null); - BytesReader valueReader = readerProvider.get(copies, originalReader); - - try { - final O v = objectFactory.create(); - long valueSize = sizeMarshaller.readSize(in); - return valueReader.read(in, valueSize, v); - } catch (Exception e) { - throw new RuntimeException(e); - } - - } - } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java index 8d6108386..3297e5b20 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java @@ -1,15 +1,12 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.map.serialization.*; -import net.openhft.chronicle.map.threadlocal.Provider; +import net.openhft.chronicle.map.serialization.MetaBytesInterop; +import net.openhft.chronicle.map.serialization.MetaBytesWriter; import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import net.openhft.lang.io.ByteBufferBytes; -import net.openhft.lang.io.BytesStore; -import net.openhft.lang.io.serialization.ObjectFactory; -import net.openhft.lang.model.Byteable; +import net.openhft.lang.io.Bytes; import org.jetbrains.annotations.NotNull; -import javax.swing.text.Segment; import java.io.File; import java.io.IOException; import java.io.Serializable; @@ -29,43 +26,18 @@ public class StatelessMapClient, V, VW, MVW extends MetaBytesWriter> extends AbstractMap implements ChronicleMap, Serializable { - final Class kClass; + private ThreadLocal sourceBuffer = new ThreadLocal(); - final BytesReader originalKeyReader; - transient Provider> keyReaderProvider; - final KI originalKeyInterop; - transient Provider keyInteropProvider; - final MKI originalMetaKeyInterop; - final MetaProvider metaKeyInteropProvider; - - final Class vClass; - final SizeMarshaller valueSizeMarshaller; - final BytesReader originalValueReader; - transient Provider> valueReaderProvider; - final VW originalValueWriter; - transient Provider valueWriterProvider; - final MVW originalMetaValueWriter; - final MetaProvider metaValueWriterProvider; - final ObjectFactory valueFactory; - final DefaultValueProvider defaultValueProvider; Map transactionIDs = new HashMap(); AtomicLong transactionID = new AtomicLong(); - private final long lockTimeOutNS; - - transient Segment[] segments; // non-final for close() - // non-final for close() and because it is initialized out of constructor - transient BytesStore ms; - transient long headerSize; - transient Set> entrySet; - - private int bits; - private int mask; private long timeoutMs = TimeUnit.SECONDS.toMillis(20); + private final Serializer valueSerializer; + private final Serializer keySerializer; @Override @@ -111,79 +83,24 @@ public class StatelessMapClient, } - private final ByteBufferBytes target; - private final SizeMarshaller keySizeMarshaller; - - public StatelessMapClient(ByteBufferBytes target, ChronicleMapBuilder builder) { - this.target = target; - - SerializationBuilder keyBuilder = builder.keyBuilder; - kClass = keyBuilder.eClass; - keySizeMarshaller = keyBuilder.sizeMarshaller(); - originalKeyReader = keyBuilder.reader(); - - originalKeyInterop = (KI) keyBuilder.interop(); - - originalMetaKeyInterop = (MKI) keyBuilder.metaInterop(); - metaKeyInteropProvider = (MetaProvider) keyBuilder.metaInteropProvider(); - - SerializationBuilder valueBuilder = builder.valueBuilder; - vClass = valueBuilder.eClass; - valueSizeMarshaller = valueBuilder.sizeMarshaller(); - originalValueReader = valueBuilder.reader(); - - originalValueWriter = (VW) valueBuilder.interop(); - originalMetaValueWriter = (MVW) valueBuilder.metaInterop(); - metaValueWriterProvider = (MetaProvider) valueBuilder.metaInteropProvider(); - valueFactory = valueBuilder.factory(); - defaultValueProvider = builder.defaultValueProvider(); - - lockTimeOutNS = builder.lockTimeOut(TimeUnit.NANOSECONDS); - - keyReaderProvider = Provider.of((Class) originalKeyReader.getClass()); - keyInteropProvider = Provider.of((Class) originalKeyInterop.getClass()); - - valueReaderProvider = Provider.of((Class) originalValueReader.getClass()); - valueWriterProvider = Provider.of((Class) originalValueWriter.getClass()); - - if (defaultValueProvider instanceof ConstantValueProvider) { - ConstantValueProvider constantValueProvider = - (ConstantValueProvider) defaultValueProvider; - if (constantValueProvider.wasDeserialized()) { - ThreadLocalCopies copies = valueReaderProvider.getCopies(null); - BytesReader valueReader = valueReaderProvider.get(copies, originalValueReader); - constantValueProvider.initTransients(valueReader); - } - } - } + private final Bytes out; - void initTransients() { - keyReaderProvider = Provider.of((Class) originalKeyReader.getClass()); - keyInteropProvider = Provider.of((Class) originalKeyInterop.getClass()); + public StatelessMapClient(Bytes out, ChronicleMapBuilder builder) { - valueReaderProvider = Provider.of((Class) originalValueReader.getClass()); - valueWriterProvider = Provider.of((Class) originalValueWriter.getClass()); + valueSerializer = new Serializer(builder.valueBuilder); + keySerializer = new Serializer(builder.keyBuilder); + + this.out = out; - if (defaultValueProvider instanceof ConstantValueProvider) { - ConstantValueProvider constantValueProvider = - (ConstantValueProvider) defaultValueProvider; - if (constantValueProvider.wasDeserialized()) { - ThreadLocalCopies copies = valueReaderProvider.getCopies(null); - BytesReader valueReader = valueReaderProvider.get(copies, originalValueReader); - constantValueProvider.initTransients(valueReader); - } - } } @Override public V putIfAbsent(K key, V value) { - target.writeByte(PUT_IF_ABSENT.ordinal()); + out.writeByte(PUT_IF_ABSENT.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - - writeKey(key, copies); + writeKey(key); writeValue(value); return null; @@ -192,15 +109,15 @@ public class StatelessMapClient, @Override public boolean remove(Object key, Object value) { - target.writeByte(REMOVE_WITH_VALUE.ordinal()); + out.writeByte(REMOVE_WITH_VALUE.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey((K) key, copies); + + writeKey((K) key); writeValue((V) value); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return blockingFetch(transactionId).readBoolean(); @@ -209,15 +126,15 @@ public class StatelessMapClient, @Override public boolean replace(K key, V oldValue, V newValue) { - target.writeByte(REPLACE_WITH_OLD_AND_NEW_VALUE.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey(key, copies); + out.writeByte(REPLACE_WITH_OLD_AND_NEW_VALUE.ordinal()); + + writeKey(key); writeValue(oldValue); writeValue(newValue); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return blockingFetch(transactionId).readBoolean(); @@ -227,14 +144,14 @@ public class StatelessMapClient, @Override public V replace(K key, V value) { - target.writeByte(REPLACE.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey(key, copies); + out.writeByte(REPLACE.ordinal()); + + writeKey(key); writeValue(value); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return readValue(blockingFetch(transactionId)); @@ -244,11 +161,11 @@ public class StatelessMapClient, @Override public int size() { - target.writeByte(SIZE.ordinal()); + out.writeByte(SIZE.ordinal()); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return blockingFetch(transactionId).readInt(); @@ -257,11 +174,11 @@ public class StatelessMapClient, @Override public boolean isEmpty() { - target.writeByte(IS_EMPTY.ordinal()); + out.writeByte(IS_EMPTY.ordinal()); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return blockingFetch(transactionId).readBoolean(); @@ -269,13 +186,13 @@ public class StatelessMapClient, @Override public boolean containsKey(Object key) { - target.writeByte(CONTAINS_KEY.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey((K) key, copies); + out.writeByte(CONTAINS_KEY.ordinal()); + + writeKey((K) key); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return blockingFetch(transactionId).readBoolean(); @@ -284,12 +201,12 @@ public class StatelessMapClient, @Override public boolean containsValue(Object value) { - target.writeByte(CONTAINS_VALUE.ordinal()); + out.writeByte(CONTAINS_VALUE.ordinal()); writeValue((V) value); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return blockingFetch(transactionId).readBoolean(); @@ -297,11 +214,11 @@ public class StatelessMapClient, @Override public long longSize() { - target.writeByte(LONG_SIZE.ordinal()); + out.writeByte(LONG_SIZE.ordinal()); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return blockingFetch(transactionId).readLong(); @@ -309,13 +226,13 @@ public class StatelessMapClient, @Override public V get(Object key) { - target.writeByte(GET.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey((K) key, copies); + out.writeByte(GET.ordinal()); + + writeKey((K) key); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return readValue(blockingFetch(transactionId)); @@ -324,22 +241,9 @@ public class StatelessMapClient, V readValue(ByteBufferBytes source) { - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - copies = metaKeyInteropProvider.getCopies(copies); + return valueSerializer.readMarshallable(source); - copies = valueReaderProvider.getCopies(copies); - BytesReader valueReader = valueReaderProvider.get(copies, originalValueReader); - - try { - final V v = valueFactory.create(); - - - long valueSize = valueSizeMarshaller.readSize(source); - return valueReader.read(source, valueSize, v); - } catch (Exception e) { - throw new RuntimeException(e); - } } @@ -355,15 +259,15 @@ public class StatelessMapClient, @Override public V put(K key, V value) { - target.writeByte(PUT.ordinal()); + out.writeByte(PUT.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - writeKey((K) key, copies); + + writeKey((K) key); writeValue((V) value); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return readValue(blockingFetch(transactionId)); @@ -372,11 +276,11 @@ public class StatelessMapClient, @Override public V remove(Object key) { - target.writeByte(REMOVE.ordinal()); + out.writeByte(REMOVE.ordinal()); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return readValue(blockingFetch(transactionId)); @@ -385,9 +289,10 @@ public class StatelessMapClient, @Override public void putAll(Map map) { - target.writeByte(PUT_ALL.ordinal()); - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); + throw new UnsupportedOperationException("todo"); + /* out.writeByte(PUT_ALL.ordinal()); + HashMap safeCopy = new HashMap(map); @@ -395,34 +300,34 @@ public class StatelessMapClient, // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server blockingFetch(transactionId); - - return; +*/ + //return; } private void writeEntries(ThreadLocalCopies copies, HashMap safeCopy) { - target.writeLong(safeCopy.size()); + out.writeLong(safeCopy.size()); Set entries = (Set) safeCopy.entrySet(); for (Entry e : entries) { - writeKey(e.getKey(), copies); + writeKey(e.getKey()); writeValue(e.getValue()); } } @Override public void clear() { - target.writeByte(CLEAR.ordinal()); + out.writeByte(CLEAR.ordinal()); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server blockingFetch(transactionId); @@ -433,11 +338,11 @@ public class StatelessMapClient, @NotNull @Override public Set keySet() { - target.writeByte(KEY_SET.ordinal()); + out.writeByte(KEY_SET.ordinal()); // send the transaction id long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return readKeySet(blockingFetch(transactionId)); @@ -456,7 +361,7 @@ public class StatelessMapClient, private K readKey(ByteBufferBytes bufferBytes) { long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server return readKey(blockingFetch(transactionId)); @@ -466,10 +371,10 @@ public class StatelessMapClient, @NotNull @Override public Collection values() { - target.writeByte(VALUES.ordinal()); + out.writeByte(VALUES.ordinal()); long transactionId = nextUniqueTransaction(); - target.writeLong(transactionId); + out.writeLong(transactionId); // get the data back from the server ByteBufferBytes bufferBytes = blockingFetch(transactionId); @@ -487,7 +392,7 @@ public class StatelessMapClient, @NotNull @Override public Set> entrySet() { - target.writeByte(ENTRY_SET.ordinal()); + out.writeByte(ENTRY_SET.ordinal()); long transactionId = nextUniqueTransaction(); @@ -511,56 +416,16 @@ public class StatelessMapClient, void writeValue(V value) { - boolean byteableValue = value instanceof Byteable; - ThreadLocalCopies copies = keyInteropProvider.getCopies(null); - VW valueWriter = valueWriterProvider.get(copies, originalValueWriter); - copies = valueWriterProvider.getCopies(copies); - long valueSize; - MetaBytesWriter metaValueWriter = null; - Byteable valueAsByteable = null; - - if (!byteableValue) { - copies = valueWriterProvider.getCopies(copies); - valueWriter = valueWriterProvider.get(copies, originalValueWriter); - copies = metaValueWriterProvider.getCopies(copies); - metaValueWriter = metaValueWriterProvider.get( - copies, originalMetaValueWriter, valueWriter, value); - valueSize = metaValueWriter.size(valueWriter, value); - } else { - valueAsByteable = (Byteable) value; - valueSize = valueAsByteable.maxSize(); - } - - valueSizeMarshaller.writeSize(target, valueSize); - - - if (metaValueWriter != null) { - metaValueWriter.write(valueWriter, target, value); - } else { - throw new UnsupportedOperationException(""); - /* assert valueAsByteable != null; - long valueOffset = entry.positionAddr() - bytes.address(); - target.zeroOut(valueOffset, valueOffset + valueSize); - valueAsByteable.bytes(target, valueOffset);*/ - } + valueSerializer.writeMarshallable(out, value); } /** * write the keysize and the key to the the {@code target} buffer * - * @param key the key of the map - * @param copies + * @param key the key of the map */ - private void writeKey(K key, final ThreadLocalCopies copies) { - KI keyWriter = keyInteropProvider.get(copies, originalKeyInterop); - - MKI metaKeyWriter = - metaKeyInteropProvider.get(copies, originalMetaKeyInterop, keyWriter, key); - - long keySize = metaKeyWriter.size(keyWriter, key); - - keySizeMarshaller.writeSize(target, keySize); - metaKeyWriter.write(keyWriter, target, key); + private void writeKey(K key) { + keySerializer.writeMarshallable(out, key); } diff --git a/src/test/java/net/openhft/chronicle/map/SerializerTest.java b/src/test/java/net/openhft/chronicle/map/SerializerTest.java index dd6360f28..eeea8b527 100644 --- a/src/test/java/net/openhft/chronicle/map/SerializerTest.java +++ b/src/test/java/net/openhft/chronicle/map/SerializerTest.java @@ -5,7 +5,6 @@ import net.openhft.lang.io.Bytes; import net.openhft.lang.io.serialization.BytesMarshallable; import net.openhft.lang.model.constraints.NotNull; import org.junit.Assert; -import org.junit.Ignore; import org.junit.Test; import java.io.*; @@ -32,7 +31,6 @@ public class SerializerTest { @Test - @Ignore public void testKeyMarshallable() throws Exception { testReadWriteKey("Test"); testReadWriteKey(1); @@ -61,17 +59,13 @@ public class SerializerTest { Serializer v = new Serializer(builder.valueBuilder); - v.setObject(value); - v.writeMarshallable(out); + v.writeMarshallable(out, value); long position = out.position(); in.limit(position); - // let just blank the value so that we can test if it works - v.setObject(null); - v.readMarshallable(in); - Object actual = v.getObject(); + Object actual = v.readMarshallable(in); Assert.assertEquals(actual, value); } @@ -88,17 +82,13 @@ public class SerializerTest { Serializer v = new Serializer(builder.keyBuilder); - v.setObject(key); - v.writeMarshallable(out); + + v.writeMarshallable(out, key); long position = out.position(); in.limit(position); - // let just blank the value so that we can test if it works - v.setObject(null); - - v.readMarshallable(in); - Object actual = v.getObject(); + Object actual = v.readMarshallable(in); Assert.assertEquals(actual, key); }